diff --git a/backend/server/contracts/jobs.py b/backend/server/contracts/jobs.py index 758b41d93..b2ec804dd 100644 --- a/backend/server/contracts/jobs.py +++ b/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 diff --git a/backend/server/contracts/schemas/jobs.v0.schema.json b/backend/server/contracts/schemas/jobs.v0.schema.json index d2e235ffe..6710d6b29 100644 --- a/backend/server/contracts/schemas/jobs.v0.schema.json +++ b/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": [ { diff --git a/backend/server/orchestrator/api.py b/backend/server/orchestrator/api.py index 02d54af49..6de2cd234 100644 --- a/backend/server/orchestrator/api.py +++ b/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: diff --git a/backend/server/orchestrator/artifacts.py b/backend/server/orchestrator/artifacts.py index 0bb57e18e..c3b104e61 100644 --- a/backend/server/orchestrator/artifacts.py +++ b/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 diff --git a/backend/server/orchestrator/coordinator.py b/backend/server/orchestrator/coordinator.py index b66e3b4d8..6c96a6bef 100644 --- a/backend/server/orchestrator/coordinator.py +++ b/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 []) diff --git a/backend/server/orchestrator/services.py b/backend/server/orchestrator/services.py index aeb4baf20..a6fcbab45 100644 --- a/backend/server/orchestrator/services.py +++ b/backend/server/orchestrator/services.py @@ -49,6 +49,7 @@ 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, @@ -56,6 +57,7 @@ def stage_context(self, run: RunRecord, rec: StageRecord, call_id: str) -> Stage 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. diff --git a/backend/server/orchestrator/tests/test_greptile_followup.py b/backend/server/orchestrator/tests/test_greptile_followup.py new file mode 100644 index 000000000..b47b58fc3 --- /dev/null +++ b/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"