fix(backend): address Greptile review findings on the orchestrator

backend/server/contracts/jobs.py

@@ -120,6 +120,9 @@ class RunRecord(Strict):
     run_id: str
     capture_id: str
     packet_sha256: str
+    # Storage key of the accepted packet manifest. An amendment's packet lives
+    # under its amendment prefix, not captures/{capture_id}/packet.json.
+    packet_uri: str | None = None
     parent_run_id: str | None = None
     status: StageStatus
     branch: Branch

backend/server/contracts/schemas/jobs.v0.schema.json

@@ -514,6 +514,18 @@
           "title": "Packet Sha256",
           "type": "string"
         },
+        "packet_uri": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null,
+          "title": "Packet Uri"
+        },
         "parent_run_id": {
           "anyOf": [
             {

backend/server/orchestrator/api.py

@@ -73,6 +73,7 @@ def start_run(
         run_id=run_id,
         capture_id=capture_id,
         packet_sha256=digest,
+        packet_uri=packet_uri,
         parent_run_id=parent_run_id,
         status="pending",
         branch=branch_of(packet),  # type: ignore[arg-type]
@@ -162,16 +163,39 @@ def on_confirmation(
     )
     try:
         from server.reads.policy import confirm
+        from server.reads.types import Reads
     except ImportError:
         return
-    confirm(
-        capture_id,
+    key = _latest_reads_key(services, capture_id)
+    if key is None:
+        return
+    try:
+        reads = Reads.model_validate(json.loads(services.storage.get_bytes(key)))  # type: ignore[attr-defined]
+    except (OSError, ValueError):
+        return
+    updated = confirm(
+        reads,
         still_id=confirmation.imageId,
         choice=confirmation.value if confirmation.kind == "confirmed" else None,
         typed=confirmation.value if confirmation.kind == "corrected" else None,
-        by=by,
+        by=by,  # type: ignore[arg-type]
         field_id=confirmation.fieldId,
     )
+    # Persist the corrected record so later results read the answer.
+    services.storage.put_bytes(key, updated.model_dump_json().encode(), "application/json")  # type: ignore[attr-defined]
+
+
+def _latest_reads_key(services: object, capture_id: str) -> str | None:
+    """The reads artifact of the capture's newest run, or None before reads exist."""
+
+    runs = services.state.runs_for_capture(capture_id)  # type: ignore[attr-defined]
+    if not runs:
+        return None
+    run = max(runs, key=lambda item: item.created_at)
+    from server.orchestrator.artifacts import overlay_stage_artifacts
+
+    key = overlay_stage_artifacts(services.state, run).artifacts.get("reads/reads.json")  # type: ignore[attr-defined]
+    return key or None
 
 
 def cancel_capture(capture_id: str) -> None:

backend/server/orchestrator/artifacts.py

@@ -188,21 +188,24 @@ def hydrate_packet(
     *,
     include_files: bool = True,
     cache_dir: Path | None = None,
+    packet_key: str | None = None,
 ) -> None:
     """Copy the packet into ``packet_dir``.
 
     ``include_files`` is false for stages that only read ``packet.json``. A warm
     volume cache skips GCS for the frames a later stage does need.
+    ``packet_key`` is the manifest's storage key: an amendment run passes its
+    own ``captures/{capture_id}/amendments/{amendment_id}/packet.json``.
     """
 
     packet_dir.mkdir(parents=True, exist_ok=True)
     if _copy_packet_cache(cache_dir, packet_dir, include_files=include_files):
         return
-    packet_key = f"captures/{capture_id}/packet.json"
-    _download_all(storage, [(packet_dir / "packet.json", packet_key)])
+    manifest_key = packet_key or f"captures/{capture_id}/packet.json"
+    _download_all(storage, [(packet_dir / "packet.json", manifest_key)])
     if not include_files:
         return
-    _download_all(storage, _packet_file_jobs(state, capture_id, packet_dir))
+    _download_all(storage, _packet_file_jobs(state, capture_id, packet_dir, manifest_key))
     if cache_dir is not None and (packet_dir / "packet.json").is_file():
         _publish_packet_cache(packet_dir, cache_dir)
 
@@ -242,7 +245,7 @@ def hydrate_artifacts(storage: object, dest_root: Path, artifacts: dict[str, str
     return present
 
 
-def _packet_file_jobs(state: object, capture_id: str, packet_dir: Path) -> list[tuple[Path, str]]:
+def _packet_file_jobs(state: object, capture_id: str, packet_dir: Path, packet_key: str) -> list[tuple[Path, str]]:
     packet: dict | None = None
     manifest = packet_dir / "packet.json"
     if manifest.is_file():
@@ -252,6 +255,9 @@ def _packet_file_jobs(state: object, capture_id: str, packet_dir: Path) -> list[
             loaded = None
         if isinstance(loaded, dict):
             packet = loaded
+    # Manifest paths are relative to the packet. A base packet sits at
+    # captures/{capture_id}/packet.json, an amendment under its own prefix.
+    base = packet_key.rsplit("packet.json", 1)[0] or f"captures/{capture_id}/"
     jobs: list[tuple[Path, str]] = []
     seen: set[str] = set()
     files = getattr(state, "files", None)
@@ -271,7 +277,7 @@ def _packet_file_jobs(state: object, capture_id: str, packet_dir: Path) -> list[
         rel = _safe_rel(rel_text)
         if rel is None:
             continue
-        jobs.append((packet_dir / rel, f"captures/{capture_id}/{rel.as_posix()}"))
+        jobs.append((packet_dir / rel, f"{base}{rel.as_posix()}"))
         seen.add(rel.as_posix())
     return jobs
 

backend/server/orchestrator/coordinator.py

@@ -42,6 +42,16 @@
 from server.orchestrator.stateio import get_json, put_json
 
 _FINAL = {"verdict_ready", "retake_request", "failed"}
+_PROFILES = {"A", "B", "C", "S"}
+
+
+def _run_profile(run: RunRecord) -> str:
+    """The run's selected rules profile, or the C default when unset."""
+
+    profile = getattr(run, "profile_default", None)
+    return profile if profile in _PROFILES else "C"
+
+
 # Capture status that matches a composed outcome. Views needed is needs_views.
 _KIND_STATUS = {
     "eligible": "complete",
@@ -261,6 +271,9 @@ def _spawn_one(self, run: RunRecord, name: StageName) -> bool:
         rec.worker_call_id = call_id
         rec.started_at = self.s.clock.iso()
         rec.heartbeat = self.s.clock.iso()
+        # entered_at belongs to the attempt that set it. A replacement worker
+        # must not inherit the previous attempt's elapsed budget.
+        rec.entered_at = None
         rec.error = None
         rec.error_code = None
         self.s.state.upsert_stage(rec)
@@ -557,7 +570,9 @@ def _apply_success(self, run: RunRecord) -> None:
             self._finish_with_result(run, decision, composed)
             self._verify_final_event(run)
             return
-        verdict = _read_json(self.s, self._overlay_artifacts(run), "verdicts/verdict.C.json")
+        verdict = _read_json(self.s, self._overlay_artifacts(run), f"verdicts/verdict.{_run_profile(run)}.json")
+        if verdict is None and _run_profile(run) != "C":
+            verdict = _read_json(self.s, self._overlay_artifacts(run), "verdicts/verdict.C.json")
         site = (verdict or {}).get("site") or {}
         decision_name = str(site.get("decision") or "manual_review")
         views = list((verdict or {}).get("views_needed") or [])

backend/server/orchestrator/services.py

@@ -49,13 +49,15 @@ def stage_context(self, run: RunRecord, rec: StageRecord, call_id: str) -> Stage
         packet_dir = work / "packet"
         packet_dir.mkdir(exist_ok=True)
         reads = self.reads_for(run.capture_id)
+        packet_uri = getattr(run, "packet_uri", None) or f"captures/{run.capture_id}/packet.json"
         hydrate_packet(
             reads,
             self.state,
             run.capture_id,
             packet_dir,
             include_files=stage_needs_packet_files(rec.stage),
             cache_dir=packet_cache_dir(run.capture_id, run.packet_sha256),
+            packet_key=packet_uri,
         )
         # Artifacts stay on raw storage. Scene merge replaces scene.json in place
         # after objects_3d has already read it, and a cache would keep the pre-merge bytes.

backend/server/orchestrator/tests/test_greptile_followup.py

@@ -0,0 +1,150 @@
+"""Regressions from the Greptile apex review of PR #247.
+
+Each test names the finding it pins: amendment packet hydration, confirmation
+persistence, per-attempt timeout budgets, and the run's own rules profile.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import timedelta
+from pathlib import Path
+
+from server.contracts.api import ConfirmationRequest
+from server.contracts.jobs import RunRecord, StageRecord
+from server.orchestrator.api import on_confirmation
+from server.orchestrator.clock import Clock
+from server.orchestrator.coordinator import RunCoordinator
+from server.orchestrator.events import emit_failed
+from server.orchestrator.services import bind
+from server.orchestrator.tests.conftest import build_services
+from server.reads.types import ConfirmBlock, MeterRead, Reads
+
+
+def _run(services, run_id: str, **overrides: object) -> RunRecord:
+    fields: dict[str, object] = {
+        "run_id": run_id,
+        "capture_id": "cap_test",
+        "packet_sha256": "a" * 64,
+        "status": "running",
+        "branch": "image",
+        "flow": "guided",
+        "tier": "arkit",
+        "profile_default": "C",
+        "created_at": services.clock.iso(),
+        "deadline_at": (services.clock.now() + timedelta(minutes=20)).isoformat(),
+        "artifacts": {},
+    }
+    fields.update(overrides)
+    return RunRecord(**fields)  # type: ignore[arg-type]
+
+
+def test_amendment_run_hydrates_its_own_packet(tmp_path: Path) -> None:
+    """Finding: artifacts.py:202 — an amendment run read the base packet."""
+    services, _uri, _digest = build_services(tmp_path, "memory")
+    amend_key = "captures/cap_test/amendments/amd-1/packet.json"
+    amend_packet = {
+        "keyframes": [{"id": "door-2"}],
+        "stills": [],
+        "files": [{"path": "extra/photo.jpg", "sha256": "ab" * 32, "bytes": 4}],
+    }
+    services.storage.put_bytes(amend_key, json.dumps(amend_packet).encode(), "application/json")
+    services.storage.put_bytes("captures/cap_test/amendments/amd-1/extra/photo.jpg", b"jpeg", "image/jpeg")
+    run = _run(services, "run_amend", packet_uri=amend_key)
+    rec = StageRecord(run_id=run.run_id, stage="poses", status="running", attempt=1, heartbeat=services.clock.iso())
+    with bind(services):
+        ctx = services.stage_context(run, rec, "call")
+    packet_dir = Path(ctx.inputs["packet_dir"])
+    # The amendment manifest, not captures/cap_test/packet.json.
+    assert json.loads((packet_dir / "packet.json").read_text()) == amend_packet
+    # Manifest-listed files resolve under the amendment prefix.
+    assert (packet_dir / "extra" / "photo.jpg").read_bytes() == b"jpeg"
+
+
+def test_confirmation_updates_the_reads_record(tmp_path: Path) -> None:
+    """Finding: api.py:173 — confirm() got a capture_id string, not a Reads."""
+    services, _uri, _digest = build_services(tmp_path, "memory")
+    key = "captures/cap_test/runs/run_reads/reads/reads.json"
+    reads = Reads(
+        capture_id="cap_test",
+        status="complete",
+        meter=MeterRead(status="auto_fill", value="00000", core="00000", confidence=0.5),
+        confirm=ConfirmBlock(candidates=["00000"]),
+    )
+    services.storage.put_bytes(key, reads.model_dump_json().encode(), "application/json")
+    run = _run(services, "run_reads", status="done", artifacts={"reads/reads.json": key})
+    services.state.put_run(run)
+    request = ConfirmationRequest(
+        fieldId="meter_number",
+        kind="corrected",
+        value="12345",
+        imageId="s1",
+        clientAt="2026-10-15T00:00:00.000Z",
+    )
+    with bind(services):
+        on_confirmation("cap_test", request, "cfm_1", by="device")
+    stored = Reads.model_validate(json.loads(services.storage.get_bytes(key)))
+    assert stored.meter is not None
+    assert stored.meter.value == "12345"
+    assert stored.meter.status == "auto_fill"
+    assert stored.confirm is not None
+    assert stored.confirm.answer == "12345"
+
+
+def test_new_attempt_clears_entered_at(tmp_path: Path) -> None:
+    """Finding: coordinator.py:266 — a retry inherited the timed-out budget."""
+    services, _uri, _digest = build_services(tmp_path, "memory")
+    # No worker runs, so the persisted attempt-start record is observable.
+    services.executor.spawn = lambda call_id, ctx: None  # type: ignore[method-assign]
+    run = _run(services, "run_retry")
+    services.state.put_run(run)
+    old = services.clock.iso()
+    services.state.upsert_stage(
+        StageRecord(
+            run_id=run.run_id,
+            stage="validate",
+            status="pending",
+            attempt=1,
+            failures=1,
+            heartbeat=old,
+            entered_at=old,
+            error_code="timeout",
+            error="timeout: validate",
+        )
+    )
+    with bind(services):
+        assert RunCoordinator(services)._spawn_one(run, "validate")
+    rec = services.state.get_stage(run.run_id, "validate")
+    assert rec.status == "running"
+    # The timeout sweep measures from entered_at; a stale value would cancel
+    # the replacement worker before it ever entered the stage.
+    assert rec.entered_at is None
+    assert rec.error_code is None
+    with bind(services):
+        RunCoordinator(services)._sweep_timeouts(run)
+    assert services.state.get_stage(run.run_id, "validate").status == "running"
+
+
+def test_success_reads_the_runs_profile_verdict(tmp_path: Path) -> None:
+    """Finding: coordinator.py:571 — success always read verdict.C.json."""
+    services, _uri, _digest = build_services(tmp_path, "memory")
+    key = "captures/cap_test/runs/run_prof/verdicts/verdict.A.json"
+    verdict = {"site": {"decision": "manual_review"}, "views_needed": ["door-2"]}
+    services.storage.put_bytes(key, json.dumps(verdict).encode(), "application/json")
+    run = _run(services, "run_prof", profile_default="A", artifacts={"verdicts/verdict.A.json": key})
+    services.state.put_run(run)
+    # One final event already exists, so the close does not rewrite the status.
+    emit_failed(
+        services.state,
+        "cap_test",
+        run.run_id,
+        "seed",
+        "seed",
+        "manual_review",
+        clock=Clock(),
+    )
+    with bind(services):
+        RunCoordinator(services)._apply_success(run)
+    # A-profile verdict asks for more views; verdict.C.json is absent, so the
+    # old hard-coded read fell through to manual_review and blocked amendments.
+    assert services.state.get_capture("cap_test").status == "needs_views"