From 6c4322acd8ba8480c568b9fdd1d8c4d39c45d8f1 Mon Sep 17 00:00:00 2001 From: Obvious Date: Thu, 15 Oct 2026 09:52:21 +0000 Subject: [PATCH 01/20] fix(backend): address Greptile review findings on the orchestrator Four findings from the apex review of PR #247, each pinned by a regression test in test_greptile_followup.py: - Amendment runs hydrated the base packet: RunRecord now carries the accepted packet_uri and hydrate_packet resolves manifest paths under the packet's own prefix (artifacts.py:202). - on_confirmation passed a capture_id string where confirm() expects a Reads record, so device corrections were never applied. It now loads the newest run's reads artifact, applies the answer, and persists the updated record (api.py:173). - A retry inherited the previous attempt's entered_at, so the timeout sweep cancelled the replacement worker before it started. _spawn_one clears the timestamp when a new attempt begins (coordinator.py:266). - Final success always read verdicts/verdict.C.json; it now reads the run's selected profile verdict and falls back to C (coordinator.py:571). The committed jobs.v0.schema.json is regenerated for the new RunRecord.packet_uri field. (cherry picked from commit 13f9f4a6286794369c61130879d6df5c8820b634) Co-authored-by: obvious-autobuild[bot] <262744130+obvious-autobuild[bot]@users.noreply.github.com> --- backend/server/contracts/jobs.py | 3 + .../contracts/schemas/jobs.v0.schema.json | 12 ++ backend/server/orchestrator/api.py | 30 +++- backend/server/orchestrator/artifacts.py | 16 +- backend/server/orchestrator/coordinator.py | 17 +- backend/server/orchestrator/services.py | 2 + .../tests/test_greptile_followup.py | 150 ++++++++++++++++++ 7 files changed, 221 insertions(+), 9 deletions(-) create mode 100644 backend/server/orchestrator/tests/test_greptile_followup.py diff --git a/backend/server/contracts/jobs.py b/backend/server/contracts/jobs.py index 01165242b..b9beefe19 100644 --- a/backend/server/contracts/jobs.py +++ b/backend/server/contracts/jobs.py @@ -122,6 +122,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 e409f9529..02cbb3aa3 100644 --- a/backend/server/contracts/schemas/jobs.v0.schema.json +++ b/backend/server/contracts/schemas/jobs.v0.schema.json @@ -516,6 +516,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 be23d9020..92084cd82 100644 --- a/backend/server/orchestrator/artifacts.py +++ b/backend/server/orchestrator/artifacts.py @@ -190,21 +190,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) @@ -244,7 +247,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(): @@ -254,6 +257,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) @@ -273,7 +279,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" From 223b32a6b61f520e2476d050fa962b5905edc187 Mon Sep 17 00:00:00 2001 From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com> Date: Thu, 15 Oct 2026 10:01:55 +0000 Subject: [PATCH 02/20] fix(backend): retryable finalize handoff and lease-guarded file-index merge - finish_ready: a failing on_finalized hook now raises ApiFailure(503) so the handoff stays retryable; run_started is only recorded after the hook succeeds. - _guard: exceptions escaping background finalize work flow through the normal failure-event and retry path instead of vanishing; the original exception is still set on the future for test callers. - ModalDictState.upsert_file: merge the per-capture path index under the shared state-store lease (claimed via the backing Dict's atomic skip_if_exists put) so concurrent containers cannot overwrite each other's index entries. - Regression tests for all three: failed run-start stays retryable, background exceptions emit failure events and leave retryable state, and two concurrent writers keep every index entry (the fake modal.Dict pairs concurrent readers onto the same snapshot to force the stale-read interleaving deterministically). (cherry picked from commit 64dfb7a96aeaa8b971f4eb7458e339b1c6f4d6a2) Co-authored-by: obvious-autobuild[bot] <262744130+obvious-autobuild[bot]@users.noreply.github.com> --- backend/server/api/finalize.py | 22 +++- .../api/tests/test_finalize_background.py | 83 ++++++++++++ .../common/backends/state_modal_dict.py | 14 +- .../common/tests/test_modal_dict_files.py | 124 ++++++++++++++++++ 4 files changed, 238 insertions(+), 5 deletions(-) create mode 100644 backend/server/api/tests/test_finalize_background.py create mode 100644 backend/server/common/tests/test_modal_dict_files.py diff --git a/backend/server/api/finalize.py b/backend/server/api/finalize.py index 7b4f0c45d..42ecadd7d 100644 --- a/backend/server/api/finalize.py +++ b/backend/server/api/finalize.py @@ -15,7 +15,6 @@ from server.api.errors import ApiFailure, fail, pointer_from_loc from server.api.events import emit from server.api.finalize_checks import manifest_errors -from server.api.hooks import safe_hook from server.api.ids import require_contract_id from server.api.keys import packet_key from server.api.promotion import promote @@ -345,7 +344,14 @@ def finish_ready( if record.status == "failed": raise fail(409, "invalid_state", "/", "promotion failed") if extra.get("started_run_id") != run_id: - safe_hook("on_finalized", deps.hooks.on_finalized, record.capture_id, digest, key, parent_run_id=parent_run_id) + try: + deps.hooks.on_finalized(record.capture_id, digest, key, parent_run_id=parent_run_id) + except Exception as exc: + log_event("hook.error", hook="on_finalized", error_code=type(exc).__name__, level="error") + # The run did not start. Leaving run_started unset keeps the handoff + # retryable: the background failure path restores the previous status, + # and a later finalize request schedules the run again. + raise fail(503, "run_start_failed", "/", "run start failed; finalize again") from exc log_event("run.started", run_id=run_id, capture_id=record.capture_id) extra = extras(deps, record.capture_id) extra["run_started"] = True @@ -494,6 +500,18 @@ def _guard(deps: Deps, capture_id: str, future: Future[None]) -> None: _finish_loop(deps, capture_id) except Exception as exc: _clear_slot(capture_id) + # The 202 has already returned and the capture reads as processing, so a + # failure stored only on this Future would strand the client: no failure + # event, no retry, no run for the sweepers to recover. Route the exception + # through the normal failure path instead. Once the run has started, the + # handoff succeeded; later errors must not undo it. + if not extras(deps, capture_id).get("run_started"): + failure = ( + exc + if isinstance(exc, ApiFailure) + else fail(503, "finalize_failed", "/", f"finalize background work failed ({type(exc).__name__})") + ) + _on_background_failure(deps, capture_id, failure) future.set_exception(exc) return future.set_result(None) diff --git a/backend/server/api/tests/test_finalize_background.py b/backend/server/api/tests/test_finalize_background.py new file mode 100644 index 000000000..ff886dc8a --- /dev/null +++ b/backend/server/api/tests/test_finalize_background.py @@ -0,0 +1,83 @@ +"""Escaped background finalize failures stay visible and retryable.""" + +from __future__ import annotations + +import pytest + +from server.api.deps import drain_finalize, extras +from server.api.tests.conftest import photo_packet +from server.api.tests.test_ingest import _create, _finalize, _put_commit + + +def _ready_capture(client, packet_id: str): + """A capture with every packet file committed, ready for a full finalize.""" + capture_id = _create(client, packet_id) + blob = b"jpeg" + _put_commit(client, capture_id, "stills/s1.jpg", blob, meta={"still": "s1"}) + packet = photo_packet(packet_id, [("stills/s1.jpg", blob)]) + return capture_id, packet + + +def test_failed_run_start_stays_retryable(client, monkeypatch): + capture_id, packet = _ready_capture(client, "packet-hook-fail") + deps = client.app_state + + def boom(*args, **kwargs): + raise RuntimeError("modal start_run failed") + + monkeypatch.setattr(deps.hooks, "on_finalized", boom) + client.hold_finalize = True + try: + response, _raw = _finalize(client, capture_id, packet) + assert response.status_code == 202, response.text + # finish_ready raises ApiFailure(503); _work_once handles it through the + # normal background failure path, so the drain itself completes. + drain_finalize(deps) + finally: + client.hold_finalize = False + + # The handoff failed, so the run must not be marked started. + assert not extras(deps, capture_id).get("run_started") + assert extras(deps, capture_id).get("started_run_id") is None + events = client.get(f"/v1/captures/{capture_id}/events").json()["events"] + failures = [e for e in events if e["type"] == "failed" and e["data"].get("code") == "run_start_failed"] + assert failures and failures[0]["data"]["next"] == "retry_finalize" + assert deps.state.get_capture(capture_id).status == "uploading" + + # With the hook healthy again, a later finalize starts the run. + monkeypatch.undo() + second, _raw = _finalize(client, capture_id, packet) + assert second.status_code == 202, second.text + assert second.json()["status"] == "processing" + assert extras(deps, capture_id).get("run_started") is True + + +def test_background_exception_fails_retryably(client, monkeypatch): + capture_id, packet = _ready_capture(client, "packet-bg-fail") + deps = client.app_state + + def offline(_key): + raise RuntimeError("volume offline") + + monkeypatch.setattr(deps.storage, "get_bytes", offline) + client.hold_finalize = True + try: + response, _raw = _finalize(client, capture_id, packet) + assert response.status_code == 202, response.text + with pytest.raises(RuntimeError): + drain_finalize(deps) + finally: + client.hold_finalize = False + + # The storage error must surface as a failure event and a retryable status, + # not as a capture stuck on processing with the exception on a dead Future. + events = client.get(f"/v1/captures/{capture_id}/events").json()["events"] + failures = [e for e in events if e["type"] == "failed" and e["data"].get("code") == "finalize_failed"] + assert failures and failures[0]["data"]["next"] == "retry_finalize" + assert deps.state.get_capture(capture_id).status == "uploading" + + # Storage healthy again: a later finalize completes the handoff. + monkeypatch.undo() + second, _raw = _finalize(client, capture_id, packet) + assert second.status_code == 202, second.text + assert second.json()["status"] == "processing" diff --git a/backend/server/common/backends/state_modal_dict.py b/backend/server/common/backends/state_modal_dict.py index c6e002f43..8633e0a61 100644 --- a/backend/server/common/backends/state_modal_dict.py +++ b/backend/server/common/backends/state_modal_dict.py @@ -336,9 +336,17 @@ def delete_capture(self, capture_id: str) -> None: def upsert_file(self, rec: FileRecord) -> None: self._kv[f"file:{rec.capture_id}:{rec.path}"] = rec.model_dump(mode="json") - paths = set(self._kv.get(f"files:{rec.capture_id}") or []) - paths.add(rec.path) - self._kv[f"files:{rec.capture_id}"] = sorted(paths) + + def merge() -> None: + paths = set(self._kv.get(f"files:{rec.capture_id}") or []) + paths.add(rec.path) + self._kv[f"files:{rec.capture_id}"] = sorted(paths) + + # Two containers can register files in parallel; the route's capture lock + # only guards one process. Merge the index under the shared Dict lease so + # a stale read in one container cannot drop the other's entry. The lease + # claims its slot with the Dict's atomic skip_if_exists put. + with_lease(self, f"files:{rec.capture_id}", merge) def get_file(self, capture_id: str, path: str) -> FileRecord | None: return self._model(f"file:{capture_id}:{path}", FileRecord) diff --git a/backend/server/common/tests/test_modal_dict_files.py b/backend/server/common/tests/test_modal_dict_files.py new file mode 100644 index 000000000..8f5ee2762 --- /dev/null +++ b/backend/server/common/tests/test_modal_dict_files.py @@ -0,0 +1,124 @@ +"""Concurrent upsert_file writers keep every index entry.""" + +from __future__ import annotations + +import threading +import time + +from server.common.backends.state_modal_dict import ModalDictState +from server.contracts.jobs import FileRecord + + +class _ReadGate: + """Hold each concurrent pair of readers until both have entered. + + A thread arriving alone waits briefly and proceeds (the leased, serialized + path); two threads arriving together are released with the same snapshot, + which is exactly the stale-read interleaving the index merge must survive. + """ + + def __init__(self) -> None: + self._cond = threading.Condition() + self._waiting = 0 + self.enabled = True + + def hold(self) -> None: + if not self.enabled: + return + with self._cond: + self._waiting += 1 + self._cond.notify_all() + if self._waiting >= 2: + self._cond.notify_all() + return + deadline = time.monotonic() + 0.05 + while self._waiting < 2 and time.monotonic() < deadline: + self._cond.wait(deadline - time.monotonic()) + self._waiting = max(0, self._waiting - 1) + + +class _FakeModalDict: + """modal.Dict stand-in: point get, write-through setitem, atomic skip_if_exists put.""" + + def __init__(self) -> None: + self.store: dict = {} + self._put_lock = threading.Lock() + self.gate = _ReadGate() + + def get(self, key): + # Capture the snapshot BEFORE the GIL is released by the gate hold: the + # unfixed read-modify-write has no GIL release between read and write, so + # a hold placed after the read would never produce a stale snapshot. + value = self.store.get(key) + if key.startswith("files:"): + self.gate.hold() + return value + + def __contains__(self, key) -> bool: + return key in self.store + + def __setitem__(self, key, value) -> None: + self.store[key] = value + + def __delitem__(self, key) -> None: + del self.store[key] + + def pop(self, key, default=None): + if key not in self.store: + return default + value = self.store[key] + del self.store[key] + return value + + def put(self, key, value, skip_if_exists: bool = False) -> bool: + with self._put_lock: + if skip_if_exists: + if key in self.store: + return False + self.store[key] = value + return True + self.store[key] = value + return True + + def keys(self): + return list(self.store.keys()) + + +def _file(capture_id: str, path: str) -> FileRecord: + return FileRecord( + capture_id=capture_id, + path=path, + role="still", + bytes=3, + sha256="a" * 64, + md5="b", + content_type="image/jpeg", + state="committed", + ) + + +def test_two_concurrent_writers_keep_every_index_entry() -> None: + mapping = _FakeModalDict() + left = ModalDictState(mapping) + right = ModalDictState(mapping) + start = threading.Barrier(2) + + def register(state: ModalDictState, tag: str) -> None: + start.wait() + for index in range(20): + state.upsert_file(_file("cap_1", f"{tag}/s{index}.jpg")) + + threads = [ + threading.Thread(target=register, args=(left, "left")), + threading.Thread(target=register, args=(right, "right")), + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + mapping.gate.enabled = False + paths = [rec.path for rec in left.files("cap_1")] + expected = {f"{tag}/s{index}.jpg" for tag in ("left", "right") for index in range(20)} + assert set(paths) == expected + assert len(paths) == 40 From 9f07c90fa7a6b02ade7bea9e87cdf3868312be59 Mon Sep 17 00:00:00 2001 From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com> Date: Thu, 15 Oct 2026 15:47:23 +0000 Subject: [PATCH 03/20] fix(backend): serialize MemoryState event seq assignment Two concurrent append_event callers could both read the same last event and append duplicate seq values, breaking the strictly increasing per-capture sequence and the events_after cursor. Hold a dedicated per-instance event lock across read-tail, seq assignment and append; release it before the returned deep copy. events_after keeps its cursor and deep-copy behavior. The regression test gates both producers at event construction while the event lock is free: 10 of 10 pre-fix runs produce duplicate seqs, 10 of 10 post-fix runs stay contiguous and complete. Greptile P1 on PR #247, comment r4281331039. Co-authored-by: obvious-autobuild[bot] <262744130+obvious-autobuild[bot]@users.noreply.github.com> (cherry picked from commit ca7448e23ad8b0e6b889e920e3ebdb6a7eebcb5c) --- .../server/common/backends/state_memory.py | 28 ++++--- .../common/tests/test_state_memory_events.py | 78 +++++++++++++++++++ 2 files changed, 94 insertions(+), 12 deletions(-) create mode 100644 backend/server/common/tests/test_state_memory_events.py diff --git a/backend/server/common/backends/state_memory.py b/backend/server/common/backends/state_memory.py index 4cc6ef244..e8d95e925 100644 --- a/backend/server/common/backends/state_memory.py +++ b/backend/server/common/backends/state_memory.py @@ -49,6 +49,7 @@ def __init__(self) -> None: self._spend: dict[str, float] = {} self._open_by_owner: dict[str, list[str]] = {} self._claim_lock = threading.Lock() + self._event_lock = threading.Lock() def get_capture(self, capture_id: str) -> CaptureRecord | None: rec = self._captures.get(capture_id) @@ -137,18 +138,21 @@ def append_event( run_id: str | None = None, stage: StageName | None = None, ) -> EventRecord: - prior = self._events.get(capture_id, []) - seq = (prior[-1].seq + 1) if prior else 1 - rec = EventRecord( - capture_id=capture_id, - seq=seq, - type=type, - at=datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ"), - data=dict(data), - run_id=run_id, - stage=stage, - ) - self._events.setdefault(capture_id, []).append(rec) + with self._event_lock: + # Read the tail, assign the seq and append as one atomic step, so two + # producers never observe the same last event. + prior = self._events.get(capture_id, []) + seq = (prior[-1].seq + 1) if prior else 1 + rec = EventRecord( + capture_id=capture_id, + seq=seq, + type=type, + at=datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ"), + data=dict(data), + run_id=run_id, + stage=stage, + ) + self._events.setdefault(capture_id, []).append(rec) return rec.model_copy(deep=True) def events_after(self, capture_id: str, seq: int, limit: int = 100) -> list[EventRecord]: diff --git a/backend/server/common/tests/test_state_memory_events.py b/backend/server/common/tests/test_state_memory_events.py new file mode 100644 index 000000000..f7359548e --- /dev/null +++ b/backend/server/common/tests/test_state_memory_events.py @@ -0,0 +1,78 @@ +"""Concurrent appends on MemoryState keep a gap-free sequence.""" + +from __future__ import annotations + +import threading + +import pytest + +from server.common.backends import state_memory +from server.common.backends.state_memory import MemoryState + +_APPENDS_PER_PRODUCER = 40 +_ROUNDS = 5 + + +def _emit( + state: MemoryState, + capture_id: str, + seqs: list[int], + errors: list[Exception], + guard: threading.Lock, +) -> None: + try: + for _ in range(_APPENDS_PER_PRODUCER): + rec = state.append_event(capture_id, "hint", {"code": "frames_blurry", "message": "m"}) + rec.data["code"] = "mutated" + with guard: + seqs.append(rec.seq) + except Exception as exc: + with guard: + errors.append(exc) + + +def _run_round( + monkeypatch: pytest.MonkeyPatch, + real_event_record: type, + capture_id: str, +) -> tuple[list[int], list[Exception], MemoryState]: + state = MemoryState() + barrier = threading.Barrier(2) + + def gated_event_record(**kwargs: object): + # Pre-fix, both producers reach event construction unlocked, meet at the + # barrier and read the same tail. With the event lock held, the gate is + # a no-op and the producers never meet. + lock = getattr(state, "_event_lock", None) + if lock is None or not lock.locked(): + barrier.wait(timeout=30) + return real_event_record(**kwargs) + + monkeypatch.setattr(state_memory, "EventRecord", gated_event_record) + seqs: list[int] = [] + errors: list[Exception] = [] + guard = threading.Lock() + threads = [threading.Thread(target=_emit, args=(state, capture_id, seqs, errors, guard)) for _ in range(2)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + monkeypatch.undo() + return seqs, errors, state + + +def test_append_event_two_producers_stay_contiguous_unique_and_ordered(monkeypatch: pytest.MonkeyPatch) -> None: + capture_id = "cap_seq2" + total = 2 * _APPENDS_PER_PRODUCER + real_event_record = state_memory.EventRecord + for _round in range(_ROUNDS): + seqs, errors, state = _run_round(monkeypatch, real_event_record, capture_id) + assert errors == [] + ordered = sorted(seqs) + assert ordered == list(range(1, total + 1)) + assert len(set(ordered)) == total + assert [rec.seq for rec in state.events_after(capture_id, 0, limit=total + 10)] == list(range(1, total + 1)) + assert [rec.seq for rec in state.events_after(capture_id, 40)] == list(range(41, total + 1)) + assert [rec.seq for rec in state.events_after(capture_id, 40, limit=5)] == [41, 42, 43, 44, 45] + stored = state.events_after(capture_id, 0, limit=1)[0] + assert stored.data["code"] == "frames_blurry" From e805a26095a29fd3376b1c4e95305bf3771816e1 Mon Sep 17 00:00:00 2001 From: Obvious Date: Fri, 16 Oct 2026 01:02:36 +0000 Subject: [PATCH 04/20] test(backend): add runtime-join journeys over the capture front door Three API-level journeys prove the imported fixes interact over the real documented dev settings: retryable dispatch with exactly one run and one final event after the retry, concurrent event producers fully visible through cursor polling, and amendment runs hydrating their own packet bytes and files. Mutation-checked: the dispatch journey fails when the 503 restore path is dropped; the amendment journey fails when the amendment hydration prefix is dropped; the seq-lock race is caught by the imported deterministic gate. Co-authored-by: Sam Gu Co-authored-by: obvious-autobuild[bot] <262744130+obvious-autobuild[bot]@users.noreply.github.com> --- backend/tests/test_runtime_join_journey.py | 468 +++++++++++++++++++++ 1 file changed, 468 insertions(+) create mode 100644 backend/tests/test_runtime_join_journey.py diff --git a/backend/tests/test_runtime_join_journey.py b/backend/tests/test_runtime_join_journey.py new file mode 100644 index 000000000..b4106adeb --- /dev/null +++ b/backend/tests/test_runtime_join_journey.py @@ -0,0 +1,468 @@ +"""The joined runtime, end to end: dispatch faults, event visibility, amendments. + +Three sibling fixes landed on their own branches and are imported here as scoped +diffs, so their regressions pass on the joined tree but nothing proves the fixes +*interact* over the real API: + +- #251 (`13f9f4a6`, cut from `2d0911f8`): amendment runs hydrate their own packet, + device confirmations apply, retries get a fresh timeout budget, and the final + status reads the run's own rules profile. +- #252 (`64dfb7a9`, cut from `2d0911f8`): a failed ``on_finalized`` dispatch stays + retryable, background exceptions surface as retryable failures, and concurrent + file-index merges run under the state lease. +- #266 (`ca7448e2`, cut from `57f57ff8`): ``MemoryState.append_event`` assigns the + per-capture ``seq`` under a dedicated lock, so the ``events_after`` cursor the + client polls never skips or repeats an event. + +This file is that proof: one API-level journey per interaction, over the +documented dev settings (SD-73) — memory state, local volume storage, static test +auth, an HMAC upload secret, stage stubs with prewarm and external vision calls +off. No Modal, no GPU, no network, no credentials. + +The journeys reuse the committed ``mini-arkit`` packet fixture and the capture +front door exactly as ``test_capture_placement_frontdoor.py`` walks it, so the +existing capture journey — through the real, ``stubs_exempt`` placement stage — +still completes here; the three tests below pin what the joins add on top of it. +""" + +from __future__ import annotations + +import base64 +import hashlib +import json +import shutil +import threading +import time +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +from server.api.app import create_app +from server.api.deps import drain_finalize, extras +from server.common.config import Settings +from server.contracts.jobs import run_id_for_capture + +REPO_ROOT = Path(__file__).resolve().parents[2] +PACKET_FIXTURE = REPO_ROOT / "backend" / "data" / "fixtures" / "packets" / "mini-arkit" + +TERMINAL_OUTCOMES = {"verdict_ready", "retake_request"} + +# The documented dev setting, as in the front-door test: the real hook chain with +# stubbed stages (placement is stubs_exempt and runs the real solver). +TEMPLATE_ENV = { + "HSS_ORCH_STAGE_STUBS": "1", + "HSS_ORCH_PREWARM": "0", + "VISION_EXTERNAL_CALLS": "off", + "STATE_BACKEND": "memory", + "STORAGE_BACKEND": "volume", + "SMP_ENVIRONMENT": "dev", +} + + +@pytest.fixture +def journey(tmp_path, monkeypatch): + """The app behind the capture front door, plus a private copy of the fixture.""" + for key, value in TEMPLATE_ENV.items(): + monkeypatch.setenv(key, value) + monkeypatch.setenv("VOLUME_ROOT", str(tmp_path / "vol")) + root = tmp_path / "packet" + shutil.copytree(PACKET_FIXTURE, root) + packet = json.loads((root / "packet.json").read_text()) + settings = Settings( + SMP_ENVIRONMENT="dev", + HSS_AUTH_MODE="static", + # Synthetic, explicitly static test auth — no real principal. + CAPTURE_API_TOKENS=f"device:{'d' * 40}", + UPLOAD_URL_HMAC_SECRET="h" * 32, + STORAGE_BACKEND="volume", + STATE_BACKEND="memory", + VOLUME_ROOT=str(tmp_path / "vol"), + CAPTURE_API_PUBLIC_URL="http://testserver", + HSS_INGEST_HOOKS="auto", + VISION_EXTERNAL_CALLS="off", + GIT_SHA="0123456789abcdef", + SMP_SERVICE_NAME="capture-api", + ) + app = create_app(settings=settings) + client = TestClient(app) + client.headers["Authorization"] = f"Bearer {'d' * 40}" + return root, packet, client, app + + +def _content_type(role: str) -> str: + if role == "stream": + return "application/gzip" + if role in {"features", "depth", "confidence", "mesh"}: + return "application/octet-stream" + return "image/jpeg" + + +def _poll_events(client: TestClient, capture_id: str) -> list[list[dict]]: + """Walk the events cursor to quiescence, keeping every page as delivered.""" + pages: list[list[dict]] = [] + after = 0 + for _ in range(60): + page = client.get(f"/v1/captures/{capture_id}/events", params={"after": after}).json() + assert page["next"] >= after, "the events cursor moved backwards" + pages.append(page["events"]) + if page["next"] == after: + return pages + after = page["next"] + pytest.fail("event feed never stopped moving") + + +def _flat(client: TestClient, capture_id: str) -> list[dict]: + return [event for page in _poll_events(client, capture_id) for event in page] + + +def _register_and_commit_fixture(root: Path, packet: dict, client: TestClient, app, capture_id: str) -> None: + """Register, upload and commit every fixture file the manifest lists.""" + registered_paths = [] + for entry in packet["files"]: + data = (root / entry["path"]).read_bytes() + item = { + "path": entry["path"], + "bytes": len(data), + "sha256": hashlib.sha256(data).hexdigest(), + "md5": base64.b64encode(hashlib.md5(data).digest()).decode(), + "role": entry.get("role", "still"), + "contentType": _content_type(entry.get("role", "still")), + } + registered = client.post(f"/v1/captures/{capture_id}/files", json={"files": [item]}) + assert registered.status_code == 200, registered.text + upload = registered.json()["files"][0]["upload"] + # The signed URL is its own capability: no bearer header on the PUT. + put = TestClient(app).put(upload["url"], content=data, headers=upload.get("headers") or {}) + assert put.status_code == 200, put.text + registered_paths.append({"path": item["path"], "sha256": item["sha256"]}) + commit = client.post(f"/v1/captures/{capture_id}/files:commit", json={"files": registered_paths}) + assert commit.status_code == 200, commit.text + + +def _finalize( + client: TestClient, + capture_id: str, + raw: bytes, + packet_id: str, + *, + amendment_id: str | None = None, +) -> dict: + suffix = f"/amendments/{amendment_id}/finalize" if amendment_id else "/finalize" + response = client.post( + f"/v1/captures/{capture_id}{suffix}", + content=raw, + headers={ + "Idempotency-Key": amendment_id or packet_id, + "Content-Type": "application/json", + "X-Packet-Sha256": hashlib.sha256(raw).hexdigest(), + }, + ) + assert response.status_code == 202, response.text + return response.json() + + +def _open_front_door(root: Path, packet: dict, client: TestClient, app, packet_id: str) -> tuple[str, bytes]: + """A capture with every packet file committed; returns (capture_id, packet bytes).""" + document = dict(packet) + document["packetId"] = packet_id + raw = (json.dumps(document, indent=2) + "\n").encode() + (root / "packet.json").write_bytes(raw) + + session = packet["session"] + created = client.post( + "/v1/captures", + json={ + "packetId": packet_id, + "formatVersion": packet["formatVersion"], + "tier": session["tier"], + "flow": "guided", + "device": { + "model": session["deviceModel"], + "systemVersion": session["systemVersion"], + "appVersion": session["appVersion"], + }, + }, + headers={"Idempotency-Key": packet_id}, + ) + assert created.status_code == 201, created.text + capture_id = created.json()["captureId"] + _register_and_commit_fixture(root, packet, client, app, capture_id) + return capture_id, raw + + +def test_failed_dispatch_stays_retryable_and_the_retry_lands_one_run_one_final(journey) -> None: + """A broken ``on_finalized`` dispatch returns the retryable error and fabricates + nothing; once the hook heals, one finalize retry establishes one run with one + composed final event — not a second run, not a duplicate final.""" + root, packet, client, app = journey + deps = app.state.deps + + # The sibling journey stays healthy while the fault capture is under test. + sibling, _ = _open_front_door(root, packet, client, app, "join-dispatch-ok") + accepted = _finalize( + client, + sibling, + (json.dumps({**packet, "packetId": "join-dispatch-ok"}, indent=2) + "\n").encode(), + "join-dispatch-ok", + ) + drain_finalize(deps) + assert deps.state.get_run(accepted["runId"]).status == "done" + + def boom(*args, **kwargs): + raise RuntimeError("orchestrator start_run failed") + + original = deps.hooks.on_finalized + deps.hooks.on_finalized = boom + try: + fault_id, raw = _open_front_door(root, packet, client, app, "join-dispatch-fault") + response = _finalize(client, fault_id, raw, "join-dispatch-fault") + assert response["status"] == "processing" + drain_finalize(deps) + finally: + deps.hooks.on_finalized = original + + # The dispatch failed retryably — the existing 503 path, not a dead end. + extra = extras(deps, fault_id) + assert not extra.get("run_started") + assert extra.get("started_run_id") is None + assert deps.state.runs_for_capture(fault_id) == [] + assert deps.state.get_capture(fault_id).status == "uploading" + events = _flat(client, fault_id) + failures = [e for e in events if e["type"] == "failed" and e["data"].get("code") == "run_start_failed"] + assert failures and failures[0]["data"]["next"] == "retry_finalize" + # A controlled fault interrupts dispatch; it cannot fabricate a success artifact. + assert not [e for e in events if e["type"] in TERMINAL_OUTCOMES] + assert deps.state.get_capture(sibling).status == "manual_review" + + # The retry — same bytes, same deterministic run id — starts exactly one run. + second = _finalize(client, fault_id, raw, "join-dispatch-fault") + assert second["status"] == "processing" + for _ in range(600): + if extras(deps, fault_id).get("run_started"): + break + time.sleep(0.01) + assert extras(deps, fault_id).get("run_started") is True + drain_finalize(deps) + + runs = deps.state.runs_for_capture(fault_id) + assert len(runs) == 1 and runs[0].status == "done" + assert runs[0].run_id == run_id_for_capture(fault_id, raw) + events = _flat(client, fault_id) + finals = [e for e in events if e["type"] in TERMINAL_OUTCOMES] + assert [e["type"] for e in finals] == ["verdict_ready"] + # The wire Event carries no run_id field — the run id lives in the payload. + assert finals[0]["data"]["runId"] == runs[0].run_id + seqs = [e["seq"] for e in events] + assert seqs == sorted(set(seqs)) and seqs == list(range(1, len(seqs) + 1)) + # The retried journey still completes through the placement stage. + assert any( + e["type"] == "stage" and e["data"].get("stage") == "placement" and e["data"].get("status") == "done" + for e in events + ) + assert len(failures) == 1 + + +def test_concurrent_event_producers_fully_visible_through_cursor_polling(journey) -> None: + """Two producers appending to one capture — the API request path and the + background finalize worker — never lose an event to the cursor: a poller + walking ``after`` pages sees every seq exactly once, contiguously.""" + root, packet, client, app = journey + deps = app.state.deps + capture_id, _ = _open_front_door(root, packet, client, app, "join-events") + _finalize( + client, capture_id, (json.dumps({**packet, "packetId": "join-events"}, indent=2) + "\n").encode(), "join-events" + ) + drain_finalize(deps) + base_events = _flat(client, capture_id) + base_count = len(base_events) + assert base_count > 0 + + producers_seen: list[list[int]] = [[], []] + start = threading.Barrier(3) + + def produce(slot: int) -> None: + start.wait() + for index in range(30): + rec = deps.state.append_event(capture_id, "capture_check", {"join": f"p{slot}-{index}"}) + producers_seen[slot].append(rec.seq) + + poller_pages: list[list[dict]] = [] + poller_stop = threading.Event() + + def poll() -> None: + start.wait() + after = 0 + while not poller_stop.is_set(): + page = client.get(f"/v1/captures/{capture_id}/events", params={"after": after}).json() + poller_pages.append(page["events"]) + if page["next"] == after: + poller_stop.wait(0.02) + else: + after = page["next"] + + threads = [ + threading.Thread(target=produce, args=(0,)), + threading.Thread(target=produce, args=(1,)), + threading.Thread(target=poll), + ] + for thread in threads: + thread.start() + for thread in threads[:2]: + thread.join() + poller_stop.set() + threads[2].join(timeout=10) + assert not threads[2].is_alive() + + # Each producer drew a distinct slice of the sequence — never the same seq. + assert producers_seen[0] and producers_seen[1] + assert set(producers_seen[0]).isdisjoint(producers_seen[1]) + + seqs = [event["seq"] for page in poller_pages for event in page] + assert seqs == sorted(seqs), "a page arrived behind the cursor" + assert len(seqs) == len(set(seqs)), "the cursor delivered an event twice" + expected = list(range(1, base_count + 61)) + assert seqs == expected, "the cursor skipped an event a producer appended" + + # The store itself holds the complete, contiguous sequence. + assert [e["seq"] for e in _flat(client, capture_id)] == expected + + +def test_amendment_hydrates_its_own_packet_bytes_and_files(journey) -> None: + """An amendment run reads the amendment's packet manifest and files, not the + original capture's — its run record carries its own packet_uri, and its new + file resolves under the amendment prefix.""" + root, packet, client, app = journey + deps = app.state.deps + capture_id, raw = _open_front_door(root, packet, client, app, "join-amend") + accepted = _finalize(client, capture_id, raw, "join-amend") + parent_run = accepted["runId"] + drain_finalize(deps) + assert deps.state.get_run(parent_run).status == "done" + + # The stub result never asks for views, so take the capture to the state the + # device would see (needs_views with one requested view) the way the + # amendment suite seeds it — the parent run itself stays the real journey's. + run = deps.state.get_run(parent_run) + result_key = run.artifacts["result/result.json"] + served = json.loads(deps.storage.get_bytes(result_key)) + served["viewsNeeded"] = [{"id": "wall-left"}] + served["outcome"]["viewsNeeded"] = [{"id": "wall-left"}] + deps.storage.put_bytes(result_key, json.dumps(served).encode(), "application/json") + deps.state.set_status(capture_id, "needs_views") + + opened = client.post( + f"/v1/captures/{capture_id}/amendments", + json={"amendmentId": "amd-join-1", "answers": ["wall-left"]}, + headers={"Idempotency-Key": "amd-join-1"}, + ) + assert opened.status_code == 201, opened.text + assert opened.json()["pathPrefix"] == "amendments/amd-join-1/" + + # The amendment's own still — a path and bytes the original capture never had. + blob = b"amendment-wall-bytes-not-in-the-original" + amend_path = "amendments/amd-join-1/stills/amend-wall.jpg" + registered = client.post( + f"/v1/captures/{capture_id}/files", + json={ + "files": [ + { + "path": amend_path, + "bytes": len(blob), + "sha256": hashlib.sha256(blob).hexdigest(), + "md5": base64.b64encode(hashlib.md5(blob).digest()).decode(), + "role": "still", + "contentType": "image/jpeg", + } + ] + }, + ) + assert registered.status_code == 200, registered.text + upload = registered.json()["files"][0]["upload"] + put = TestClient(app).put(upload["url"], content=blob, headers=upload.get("headers") or {}) + assert put.status_code == 200, put.text + committed = client.post( + f"/v1/captures/{capture_id}/files:commit", + json={"files": [{"path": amend_path, "sha256": hashlib.sha256(blob).hexdigest()}]}, + ) + assert committed.status_code == 200, committed.text + + # The device's amendment packet: its own packetId, the parent's packet id, + # and a manifest of exactly the files it re-captured. + amend_packet = { + "format": "house-capture-packet", + "formatVersion": "0.4", + "packetId": "join-amend-followup", + "parentPacketId": "join-amend", + "createdAt": "2026-10-15T00:00:00.000Z", + "units": {"length": "meters", "angle": "degrees", "time": "seconds", "image": "pixels"}, + "conventions": { + "world": "ARKit gravity, y up", + "camera": "ARKit camera, looks along -z", + "pose": "arkit-column-major", + "intrinsics": "[fx, fy, cx, cy]", + "pixel": "[u, v] top-left, v down", + }, + "keyframes": [], + "source": {"kind": "synthetic"}, + "session": { + "id": "sess-amend", + "startedAt": "2026-10-15T00:00:00.000Z", + "startedAtUptime": 0.0, + "appVersion": "1", + "deviceModel": "iPhone", + "systemVersion": "18", + "tier": "photo_only", + "lidarAvailable": False, + "sceneDepthEnabled": False, + }, + "stills": [ + { + "id": "s1", + "purpose": "house_overview", + "img": "stills/amend-wall.jpg", + "w": 2, + "h": 2, + "timestamp": 0, + "orientation": 1, + } + ], + "files": [ + { + "path": "stills/amend-wall.jpg", + "bytes": len(blob), + "sha256": hashlib.sha256(blob).hexdigest(), + "role": "still", + } + ], + } + amend_raw = json.dumps(amend_packet).encode() + finalize = _finalize(client, capture_id, amend_raw, "join-amend-followup", amendment_id="amd-join-1") + assert finalize["status"] == "processing" + drain_finalize(deps) + + runs = deps.state.runs_for_capture(capture_id) + assert len(runs) == 2 + amend_run = max(runs, key=lambda item: item.created_at) + assert amend_run.run_id == run_id_for_capture(capture_id, amend_raw) + # The run's identity is the amendment's packet, not the original capture's. + assert amend_run.packet_uri == f"captures/{capture_id}/amendments/amd-join-1/packet.json" + assert amend_run.parent_run_id == parent_run + assert amend_run.status == "done" + assert amend_run.packet_sha256 == hashlib.sha256(amend_raw).hexdigest() + + events = _flat(client, capture_id) + assert [e["type"] for e in events if e["type"] in TERMINAL_OUTCOMES] == ["verdict_ready", "verdict_ready"] + # The wire Event carries no run_id field — the run id lives in the payload. + assert any(e["data"].get("runId") == amend_run.run_id for e in events if e["type"] == "verdict_ready") + seqs = [e["seq"] for e in events] + assert seqs == sorted(set(seqs)) and seqs == list(range(1, len(seqs) + 1)) + + # The amendment's file landed under the amendment prefix, and the run + # completed — hydration resolved the manifest under the amendment prefix. + # Under the pre-fix derivation the manifest-listed file would be fetched from + # captures/{capture_id}/stills/amend-wall.jpg, which does not exist, and the + # run would fail during stage hydration. + assert deps.storage.get_bytes(f"captures/{capture_id}/amendments/amd-join-1/stills/amend-wall.jpg") == blob + with pytest.raises(OSError): + deps.storage.get_bytes(f"captures/{capture_id}/stills/amend-wall.jpg") From 54b439cd489cad473e1264b2cd525a82fca87086 Mon Sep 17 00:00:00 2001 From: Obvious Date: Fri, 16 Oct 2026 01:02:36 +0000 Subject: [PATCH 05/20] test(backend): read the fake start_run failure as a local contract Rename the raised error from the Modal-flavored string to the orchestrator contract and note that Modal is the production transport while this suite pins the retryable handoff locally. Co-authored-by: Sam Gu Co-authored-by: obvious-autobuild[bot] <262744130+obvious-autobuild[bot]@users.noreply.github.com> --- backend/server/api/tests/test_finalize_background.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/backend/server/api/tests/test_finalize_background.py b/backend/server/api/tests/test_finalize_background.py index ff886dc8a..c8269798f 100644 --- a/backend/server/api/tests/test_finalize_background.py +++ b/backend/server/api/tests/test_finalize_background.py @@ -22,8 +22,10 @@ def test_failed_run_start_stays_retryable(client, monkeypatch): capture_id, packet = _ready_capture(client, "packet-hook-fail") deps = client.app_state + # A local fake of any orchestrator start_run failure — Modal is the + # production transport; this suite pins the retryable handoff contract. def boom(*args, **kwargs): - raise RuntimeError("modal start_run failed") + raise RuntimeError("orchestrator start_run failed") monkeypatch.setattr(deps.hooks, "on_finalized", boom) client.hold_finalize = True From 0d19b3c1f0f5a240ab274070563804f15ffbcbf0 Mon Sep 17 00:00:00 2001 From: Obvious Date: Fri, 16 Oct 2026 05:54:44 +0000 Subject: [PATCH 06/20] fix(backend): close amendment, confirmation, and retry defects on the runtime join - Lease file-index merge now claims strictly: with_lease(strict=True) raises LeaseBusy on exhaustion and the API maps it to a retryable 503 with Retry-After, so two concurrent register+commit writers can no longer silently merge under one side's stale lease. - Packet hydration resolves the selected packet's manifest before committed file records, so an amendment reusing a base still name hydrates its own bytes instead of the base capture's. - Finalize background-failure publication holds finish_sync across the run_started check and the failure event, releasing the worker slot only after publication; a successful retry can no longer interleave. - Accepted confirmations that arrive before reads exist are persisted per field and applied when the reads stage creates its record (real and stub paths); the route keeps its authorization and replay gates. - Late confirmations on a done run recompose through the normal coordinator lifecycle (criteria+result reset, same run id), and the stub result now derives the meter from hydrated reads/reads.json so a correction is observable on GET /result. - Event-polling journey drains to the expected final cursor instead of an unbounded loop. Co-authored-by: Sam Gu Co-authored-by: obvious-autobuild[bot] <262744130+obvious-autobuild[bot]@users.noreply.github.com> --- backend/server/api/app.py | 8 ++ backend/server/api/finalize.py | 23 +-- .../api/tests/test_finalize_background.py | 47 +++++- .../common/backends/state_modal_dict.py | 11 +- backend/server/common/lease.py | 27 +++- backend/server/orchestrator/api.py | 51 +++++++ backend/server/orchestrator/artifacts.py | 21 +-- backend/server/orchestrator/stubs/reads.py | 33 ++++- backend/server/orchestrator/stubs/result.py | 50 ++++++- .../tests/test_greptile_followup.py | 55 ++++++- backend/server/reads/__init__.py | 8 ++ backend/server/reads/pending.py | 65 +++++++++ backend/tests/test_runtime_join_journey.py | 134 ++++++++++++++++++ 13 files changed, 509 insertions(+), 24 deletions(-) create mode 100644 backend/server/reads/pending.py diff --git a/backend/server/api/app.py b/backend/server/api/app.py index 8534c2c8c..a3586c0e0 100644 --- a/backend/server/api/app.py +++ b/backend/server/api/app.py @@ -31,6 +31,7 @@ uploads, ) from server.common.config import Settings, load, resolve_state, resolve_storage +from server.common.lease import LeaseBusy from server.common.logging import log_event, request_fields _MB = 1024 * 1024 @@ -202,6 +203,13 @@ async def send_now(response: JSONResponse): async def api_failure(_request: Request, failure: ApiFailure): return error_response(failure) + @app.exception_handler(LeaseBusy) + async def lease_busy(_request: Request, exc: LeaseBusy): + # The file-index merge never ran, so the registration is safe to retry. + return error_response( + fail(503, "lease_busy", "/", "file index is busy; retry the registration", **{"Retry-After": "1"}) + ) + @app.exception_handler(RequestValidationError) async def validation_error(_request: Request, exc: RequestValidationError): items = [] diff --git a/backend/server/api/finalize.py b/backend/server/api/finalize.py index 42ecadd7d..c0ea67811 100644 --- a/backend/server/api/finalize.py +++ b/backend/server/api/finalize.py @@ -499,19 +499,26 @@ def _guard(deps: Deps, capture_id: str, future: Future[None]) -> None: try: _finish_loop(deps, capture_id) except Exception as exc: - _clear_slot(capture_id) # The 202 has already returned and the capture reads as processing, so a # failure stored only on this Future would strand the client: no failure # event, no retry, no run for the sweepers to recover. Route the exception # through the normal failure path instead. Once the run has started, the # handoff succeeded; later errors must not undo it. - if not extras(deps, capture_id).get("run_started"): - failure = ( - exc - if isinstance(exc, ApiFailure) - else fail(503, "finalize_failed", "/", f"finalize background work failed ({type(exc).__name__})") - ) - _on_background_failure(deps, capture_id, failure) + # + # The worker slot stays occupied and the publication lock is held across + # the run_started check and the failure write: a concurrent finalize retry + # publishes its success under the same lock, so this older worker can + # never reset a healed retry or emit a contradictory terminal event after + # it. + with deps.locks.finish_sync(capture_id): + if not extras(deps, capture_id).get("run_started"): + failure = ( + exc + if isinstance(exc, ApiFailure) + else fail(503, "finalize_failed", "/", f"finalize background work failed ({type(exc).__name__})") + ) + _on_background_failure(deps, capture_id, failure) + _clear_slot(capture_id) future.set_exception(exc) return future.set_result(None) diff --git a/backend/server/api/tests/test_finalize_background.py b/backend/server/api/tests/test_finalize_background.py index c8269798f..b9aea5fd6 100644 --- a/backend/server/api/tests/test_finalize_background.py +++ b/backend/server/api/tests/test_finalize_background.py @@ -78,8 +78,53 @@ def offline(_key): assert failures and failures[0]["data"]["next"] == "retry_finalize" assert deps.state.get_capture(capture_id).status == "uploading" - # Storage healthy again: a later finalize completes the handoff. + +def test_escaped_failure_publishes_before_releasing_the_worker_slot(client, monkeypatch): + """Finding: _guard — the slot was released before the failure was published. + + A finalize retry that lands in the gap sees a free slot, publishes its + success, and is then overwritten by the older worker's failure — the healed + capture resets to uploading behind a failure event for a run that started. + With publication and the run_started check serialized under the finish lock, + the older worker observes the retry's success and defers; either order ends + consistent. This pins the ordering half: at the moment the failure is + published, this worker still owns the slot. + """ + import server.api.finalize as finalize_module + + capture_id, packet = _ready_capture(client, "packet-bg-order") + deps = client.app_state + + def offline(_key): + raise RuntimeError("volume offline") + + publication_slot_state: list[bool] = [] + original_failure_path = finalize_module._on_background_failure + + def probing_failure_path(*args, **kwargs): + publication_slot_state.append(capture_id in finalize_module._slots) + return original_failure_path(*args, **kwargs) + + monkeypatch.setattr(deps.storage, "get_bytes", offline) + monkeypatch.setattr(finalize_module, "_on_background_failure", probing_failure_path) + client.hold_finalize = True + try: + response, _raw = _finalize(client, capture_id, packet) + assert response.status_code == 202, response.text + with pytest.raises(RuntimeError): + drain_finalize(deps) + finally: + client.hold_finalize = False + + # The failure published while this worker still held the slot: a retry that + # starts after the slot frees sees the failure already published (or the + # run_started flag of a healed retry) — never both. + assert publication_slot_state == [True] + assert deps.state.get_capture(capture_id).status == "uploading" + + # And the healed retry still lands one healthy run. monkeypatch.undo() second, _raw = _finalize(client, capture_id, packet) assert second.status_code == 202, second.text + assert extras(deps, capture_id).get("run_started") is True assert second.json()["status"] == "processing" diff --git a/backend/server/common/backends/state_modal_dict.py b/backend/server/common/backends/state_modal_dict.py index 8633e0a61..ca3733b78 100644 --- a/backend/server/common/backends/state_modal_dict.py +++ b/backend/server/common/backends/state_modal_dict.py @@ -20,6 +20,11 @@ StageStatus, ) +# How long upsert_file waits for the per-capture index lease before failing. +# A module constant so tests can shorten it without sleeping through the real +# budget. +_INDEX_LEASE_WAIT_S = 3.0 + _OPEN = frozenset({"uploading", "awaiting_files"}) _APPEND_GUARD = threading.Lock() _APPEND_LOCKS: dict[int, threading.Lock] = {} @@ -346,7 +351,11 @@ def merge() -> None: # only guards one process. Merge the index under the shared Dict lease so # a stale read in one container cannot drop the other's entry. The lease # claims its slot with the Dict's atomic skip_if_exists put. - with_lease(self, f"files:{rec.capture_id}", merge) + # + # The merge runs strictly under the lease: a busy lease raises LeaseBusy + # (the route turns it into a retryable 503) instead of merging unlocked + # while an abandoned-but-valid lease holds the slot. + with_lease(self, f"files:{rec.capture_id}", merge, wait_s=_INDEX_LEASE_WAIT_S, strict=True) def get_file(self, capture_id: str, path: str) -> FileRecord | None: return self._model(f"file:{capture_id}:{path}", FileRecord) diff --git a/backend/server/common/lease.py b/backend/server/common/lease.py index eb7e4db0c..4de6cb93a 100644 --- a/backend/server/common/lease.py +++ b/backend/server/common/lease.py @@ -7,8 +7,21 @@ from collections.abc import Callable -def with_lease(state: object, name: str, fn: Callable[[], None], *, wait_s: float = 3.0, lease_s: float = 10.0) -> None: - """Run ``fn`` while holding ``lease:{name}``. An abandoned lease can be taken after ``lease_s``.""" +def with_lease( + state: object, + name: str, + fn: Callable[[], None], + *, + wait_s: float = 3.0, + lease_s: float = 10.0, + strict: bool = False, +) -> None: + """Run ``fn`` while holding ``lease:{name}``. An abandoned lease can be taken after ``lease_s``. + + With ``strict=True`` the caller refuses to run ``fn`` without ownership: a + lease that stays busy past ``wait_s`` raises :class:`LeaseBusy` instead of + executing the protected section unlocked. + """ key = f"lease:{name}" token = {"id": uuid.uuid4().hex, "at": time.time()} @@ -26,5 +39,15 @@ def with_lease(state: object, name: str, fn: Callable[[], None], *, wait_s: floa state.delete_kv(key) # type: ignore[attr-defined] continue if time.monotonic() >= deadline: + if strict: + raise LeaseBusy(f"lease {name!r} stayed busy for {wait_s:g}s") return fn() time.sleep(0.01) + + +class LeaseBusy(RuntimeError): + """A strict lease could not be acquired within its wait budget. + + Callers turn this into a retryable failure: the protected section never ran, + so retrying the operation later is safe. + """ diff --git a/backend/server/orchestrator/api.py b/backend/server/orchestrator/api.py index 6de2cd234..d54a16887 100644 --- a/backend/server/orchestrator/api.py +++ b/backend/server/orchestrator/api.py @@ -148,6 +148,50 @@ def dispatch_early(capture_id: str, files: list[CommittedFile]) -> None: dispatch(current(), capture_id, files) +def _recompose_after_correction(capture_id: str, run: RunRecord) -> None: + """Re-run criteria and result on the finished run so the served result reflects a late correction. + + A confirmation accepted after reads reached the result cannot retroactively + enter the composed payload. ``start_run`` cannot stage a fresh run here — it + derives the run id from the packet digest, so the same packet joins the + existing run instead of creating one. Instead the finished run's criteria and + result stages go back to pending, the run re-enters the normal coordinator + lifecycle, and the re-executed stages overwrite the same artifact keys: the + recomposed payload is what ``GET /result`` serves next. The acceptance event + and the recomposition's ``verdict_ready`` land on ``GET /events``. A replayed + confirmation still short-circuits upstream, so one correction recomposes once. + + A live run is left alone — its result stage has not consumed the corrected + reads yet, so it recomposes by simply running. + """ + + services = current() + if run.status != "done": + return + now = services.clock.iso() + for stage in ("criteria", "result"): + rec = services.state.get_stage(run.run_id, stage) # type: ignore[attr-defined] + services.state.upsert_stage( # type: ignore[attr-defined] + StageRecord( + run_id=run.run_id, + stage=stage, + status="pending", + attempt=(rec.attempt if rec is not None else 0) + 1, + heartbeat=now, + ) + ) + run.status = "running" + services.state.put_run(run) # type: ignore[attr-defined] + raw = dict(get_json(services.state, f"run_control:{run.run_id}") or {}) # type: ignore[attr-defined] + raw["merge_done"] = False + put_json(services.state, f"run_control:{run.run_id}", raw) # type: ignore[attr-defined] + try: + RunCoordinator(services).drive(run.run_id) + except CoordinatorDied: + pass + log_event("confirmation.recompose", capture_id=capture_id, run_id=run.run_id) + + def on_confirmation( capture_id: str, confirmation: ConfirmationRequest, @@ -183,6 +227,13 @@ def on_confirmation( ) # 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] + # A correction accepted after the result was composed must reach the served + # payload, not only the stored reads. The route replays short-circuit above + # this hook, so one accepted correction recomposes once. + runs = services.state.runs_for_capture(capture_id) # type: ignore[attr-defined] + if runs: + newest = max(runs, key=lambda item: item.created_at) + _recompose_after_correction(capture_id, newest) def _latest_reads_key(services: object, capture_id: str) -> str | None: diff --git a/backend/server/orchestrator/artifacts.py b/backend/server/orchestrator/artifacts.py index 92084cd82..cd8add6b1 100644 --- a/backend/server/orchestrator/artifacts.py +++ b/backend/server/orchestrator/artifacts.py @@ -262,6 +262,17 @@ def _packet_file_jobs(state: object, capture_id: str, packet_dir: Path, packet_k base = packet_key.rsplit("packet.json", 1)[0] or f"captures/{capture_id}/" jobs: list[tuple[Path, str]] = [] seen: set[str] = set() + # The selected packet's manifest resolves first: its paths are authoritative + # for this run. An amendment that reuses a base filename (stills/s1.jpg) with + # different bytes must download its own copy under the amendment prefix — a + # committed base-file record entering `seen` first would silently hydrate the + # base bytes and the real validator would reject the size/hash mismatch. + for rel_text in _packet_paths(packet): + rel = _safe_rel(rel_text) + if rel is None or rel.as_posix() in seen: + continue + jobs.append((packet_dir / rel, f"{base}{rel.as_posix()}")) + seen.add(rel.as_posix()) files = getattr(state, "files", None) records = files(capture_id) if callable(files) else [] for rec in records: @@ -269,18 +280,10 @@ def _packet_file_jobs(state: object, capture_id: str, packet_dir: Path, packet_k continue rel = _safe_rel(str(getattr(rec, "path", "") or "")) key = str(getattr(rec, "key", "") or "") - if rel is None or not key: + if rel is None or not key or rel.as_posix() in seen: continue jobs.append((packet_dir / rel, key)) seen.add(rel.as_posix()) - for rel_text in _packet_paths(packet): - if rel_text in seen: - continue - rel = _safe_rel(rel_text) - if rel is None: - continue - jobs.append((packet_dir / rel, f"{base}{rel.as_posix()}")) - seen.add(rel.as_posix()) return jobs diff --git a/backend/server/orchestrator/stubs/reads.py b/backend/server/orchestrator/stubs/reads.py index 685598eee..cef50127f 100644 --- a/backend/server/orchestrator/stubs/reads.py +++ b/backend/server/orchestrator/stubs/reads.py @@ -13,5 +13,36 @@ def run(ctx: StageContext) -> StageResult: raw = json.loads(_STUB.read_text()) if _STUB.is_file() else {"meter_number": "000000000"} - body = {"format": "base-reads/v0", "capture_id": ctx.capture_id, "facts": [], "meter_number": raw["meter_number"]} + body: dict = { + "format": "base-reads/v0", + "capture_id": ctx.capture_id, + "facts": [], + "meter_number": raw["meter_number"], + } + # The reads stage is the application point for confirmations accepted before + # reads existed; the stub path applies the same persisted slots so an early + # answer reaches the composed result in stub mode too. Without pending slots + # the stub body is byte-identical to the old shape. + try: + from server.reads.pending import pending_slots + from server.reads.types import ConfirmBlock, MeterRead, Reads + + rows = pending_slots(ctx.state, ctx.capture_id) + if rows: + reads = Reads( + capture_id=ctx.capture_id, + status="complete", + meter=MeterRead( + status="auto_fill", + value=str(raw["meter_number"]), + core=str(raw["meter_number"]), + confidence=0.5, + ), + confirm=ConfirmBlock(candidates=[str(raw["meter_number"])]), + ) + from server.reads.pending import apply_pending + + body = Reads.model_validate(apply_pending(ctx.state, ctx.capture_id, reads)).model_dump(mode="json") + except ImportError: + pass return write_files(ctx, {"reads/reads.json": json_bytes(body)}) diff --git a/backend/server/orchestrator/stubs/result.py b/backend/server/orchestrator/stubs/result.py index 24c82b548..856898f18 100644 --- a/backend/server/orchestrator/stubs/result.py +++ b/backend/server/orchestrator/stubs/result.py @@ -3,9 +3,11 @@ from __future__ import annotations import json +from pathlib import Path +from typing import Literal from server.contracts.jobs import Profile, StageContext, StageResult -from server.contracts.result import Outcome, ResultResponse +from server.contracts.result import MeterBlock, Outcome, ResultResponse from server.orchestrator.stubs._common import json_bytes, sha256, write_files _PROFILES: frozenset[str] = frozenset({"A", "B", "C", "S"}) @@ -37,6 +39,7 @@ def run(ctx: StageContext) -> StageResult: viewsNeeded=[], reasons=[], ), + meter=_stub_meter(ctx), ) body = response.model_dump(mode="json") ResultResponse.model_validate(body) @@ -57,3 +60,48 @@ def _profile(ctx: StageContext) -> Profile: if configured in _PROFILES: return configured # type: ignore[return-value] return "C" + + +def _meter_status(raw: str) -> Literal["auto_filled", "prefill_confirm", "unreadable", "not_captured"]: + """The real composer's meter mapping, with autofill enabled.""" + if raw == "auto_fill": + return "auto_filled" + if raw in {"prefill", "choose", "single_reader"}: + return "prefill_confirm" + if raw == "unreadable": + return "unreadable" + return "not_captured" + + +def _stub_meter(ctx: StageContext) -> MeterBlock: + """The reads-derived meter, as the real composer reports it. + + The result stage hydrates ``reads/reads.json``; serving the meter from it is + what makes a recomposition after a late confirmation observable on + ``GET /result`` without editing the verdict. The mapping mirrors + ``server.result.meter.meter_block`` with autofill enabled. + """ + + try: + doc = json.loads((Path(ctx.work_dir) / "reads" / "reads.json").read_text()) + meter = doc.get("meter") or {} + except (OSError, json.JSONDecodeError, AttributeError): + meter = {} + status = _meter_status(str(meter.get("status") or "")) + value = meter.get("value") + shown = str(value) if isinstance(value, str) else None + if status == "not_captured": + shown = None + candidates = [ + str(item.get("value")) + for item in (meter.get("candidates") or []) + if isinstance(item, dict) and item.get("value") + ][:3] + return MeterBlock( + status=status, + value=shown, + candidates=candidates, + barcodeConfirmed=bool(meter.get("barcode_match")), + readersAgree=bool(meter.get("agree")), + clean=False, + ) diff --git a/backend/server/orchestrator/tests/test_greptile_followup.py b/backend/server/orchestrator/tests/test_greptile_followup.py index b47b58fc3..039a3a4bd 100644 --- a/backend/server/orchestrator/tests/test_greptile_followup.py +++ b/backend/server/orchestrator/tests/test_greptile_followup.py @@ -6,18 +6,22 @@ from __future__ import annotations +import base64 +import hashlib 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.contracts.jobs import FileRecord, 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.packet.integrity import check_files +from server.packet.report import Bag from server.reads.types import ConfirmBlock, MeterRead, Reads @@ -61,6 +65,55 @@ def test_amendment_run_hydrates_its_own_packet(tmp_path: Path) -> None: assert (packet_dir / "extra" / "photo.jpg").read_bytes() == b"jpeg" +def test_amendment_same_name_still_hydrates_amendment_bytes(tmp_path: Path) -> None: + """Finding: _packet_file_jobs — a committed base file claimed the relative path first. + + An amendment that re-captures a base filename (stills/s1.jpg) with different + bytes must hydrate its own copy under the amendment prefix. Under the + pre-fix derivation the committed base record entered the seen-set first and + the base bytes landed at packet_dir/stills/s1.jpg, so the real validator + would reject the size/sha256 mismatch against the amendment manifest. + """ + services, _uri, _digest = build_services(tmp_path, "memory") + base_blob = b"base-still-bytes" + amend_blob = b"amendment-still-bytes-different" + services.storage.put_bytes("captures/cap_test/stills/s1.jpg", base_blob, "image/jpeg") + services.state.upsert_file( + FileRecord( + capture_id="cap_test", + path="stills/s1.jpg", + role="still", + bytes=len(base_blob), + sha256=hashlib.sha256(base_blob).hexdigest(), + md5=base64.b64encode(hashlib.md5(base_blob).digest()).decode(), + content_type="image/jpeg", + state="committed", + key="captures/cap_test/stills/s1.jpg", + ) + ) + amend_key = "captures/cap_test/amendments/amd-2/packet.json" + amend_packet = { + "keyframes": [], + "stills": [], + "files": [ + {"path": "stills/s1.jpg", "sha256": hashlib.sha256(amend_blob).hexdigest(), "bytes": len(amend_blob)} + ], + } + services.storage.put_bytes(amend_key, json.dumps(amend_packet).encode(), "application/json") + services.storage.put_bytes("captures/cap_test/amendments/amd-2/stills/s1.jpg", amend_blob, "image/jpeg") + run = _run(services, "run_amend_2", 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's bytes — not the committed base file's — sit at the manifest path. + assert (packet_dir / "stills" / "s1.jpg").read_bytes() == amend_blob + # And the real packet validator agrees with the amendment manifest. + bag = Bag() + check_files(amend_packet, packet_dir, bag, check_hashes=True, known_sha256=None) + assert bag.errors == [] + + 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") diff --git a/backend/server/reads/__init__.py b/backend/server/reads/__init__.py index e41206abe..0e811c385 100644 --- a/backend/server/reads/__init__.py +++ b/backend/server/reads/__init__.py @@ -25,6 +25,14 @@ def run(ctx: StageContext) -> StageResult: ) packet = _packet(ctx) reads = build_reads(ctx.capture_id, ctx.storage, cfg, packet) + # Accepted confirmations that arrived before reads existed apply here, once, + # onto the record this stage creates. + try: + from server.reads.pending import apply_pending + + reads = apply_pending(ctx.state, ctx.capture_id, reads) + except ImportError: + pass rel = write_reads( ctx.capture_id, ctx.run_id, diff --git a/backend/server/reads/pending.py b/backend/server/reads/pending.py new file mode 100644 index 000000000..628731151 --- /dev/null +++ b/backend/server/reads/pending.py @@ -0,0 +1,65 @@ +"""Accepted confirmations the reads record must carry. + +The confirmations route persists every accepted answer as a per-field slot +(``confirm:{capture_id}:{field_id}``, listed in ``confirmix:{capture_id}``) as +soon as it passes the conflict and authorization checks. A confirmation that +arrives before ``reads/reads.json`` exists is applied here, when the reads stage +creates its record, so an early answer reaches the result without the client +retrying. The route's fingerprint check keeps a replayed confirmation from being +applied twice as a new answer; re-applying one accepted slot is idempotent. +""" + +from __future__ import annotations + +from server.common.logging import log_event +from server.reads.policy import confirm +from server.reads.types import Reads + +_BY = frozenset({"device", "reviewer"}) +_FIELDS = frozenset({"meter_number", "panel_brand", "main_breaker_amps"}) + + +def pending_slots(state: object, capture_id: str) -> list[dict]: + """The accepted per-field confirmation slots persisted for this capture.""" + + if state is None: + return [] + listing = state.get_kv(f"confirmix:{capture_id}") # type: ignore[attr-defined] + rows: list[dict] = [] + for field_id in (listing or {}).get("fields") or []: + row = state.get_kv(f"confirm:{capture_id}:{field_id}") # type: ignore[attr-defined] + if isinstance(row, dict) and row.get("kind") in {"confirmed", "corrected", "none"}: + rows.append(row) + return rows + + +def apply_pending(state: object, capture_id: str, reads: Reads) -> Reads: + """Apply each accepted confirmation slot once, preserving answer and authorization.""" + + for row in pending_slots(state, capture_id): + by = row.get("by") if row.get("by") in _BY else "device" + field_id = row.get("fieldId") if row.get("fieldId") in _FIELDS else None + if field_id is None: + continue + kind = row.get("kind") + try: + reads = confirm( + reads, + still_id=str(row.get("imageId") or ""), + choice=row.get("value") if kind == "confirmed" else None, + typed=row.get("value") if kind == "corrected" else None, + by=by, # type: ignore[arg-type] + field_id=field_id, # type: ignore[arg-type] + ) + except Exception as exc: + # A stale slot that no longer fits the record must not fail the stage. + log_event( + "reads.pending_confirmation.skipped", + capture_id=capture_id, + field_id=str(row.get("fieldId")), + error=type(exc).__name__, + ) + return reads + + +__all__ = ["apply_pending", "pending_slots"] diff --git a/backend/tests/test_runtime_join_journey.py b/backend/tests/test_runtime_join_journey.py index b4106adeb..9aaf82f5d 100644 --- a/backend/tests/test_runtime_join_journey.py +++ b/backend/tests/test_runtime_join_journey.py @@ -288,6 +288,7 @@ def produce(slot: int) -> None: producers_seen[slot].append(rec.seq) poller_pages: list[list[dict]] = [] + poller_after: list[int] = [] poller_stop = threading.Event() def poll() -> None: @@ -296,6 +297,7 @@ def poll() -> None: while not poller_stop.is_set(): page = client.get(f"/v1/captures/{capture_id}/events", params={"after": after}).json() poller_pages.append(page["events"]) + poller_after.append(after) if page["next"] == after: poller_stop.wait(0.02) else: @@ -466,3 +468,135 @@ def test_amendment_hydrates_its_own_packet_bytes_and_files(journey) -> None: assert deps.storage.get_bytes(f"captures/{capture_id}/amendments/amd-join-1/stills/amend-wall.jpg") == blob with pytest.raises(OSError): deps.storage.get_bytes(f"captures/{capture_id}/stills/amend-wall.jpg") + + +def _register_extra_still(root: Path, client: TestClient, app, capture_id: str, name: str, image_id: str) -> bytes: + """One still beyond the fixture manifest, carrying the meta the image lookup reads.""" + blob = f"extra-still-{name}-bytes".encode() + item = { + "path": f"stills/{name}.jpg", + "bytes": len(blob), + "sha256": hashlib.sha256(blob).hexdigest(), + "md5": base64.b64encode(hashlib.md5(blob).digest()).decode(), + "role": "still", + "contentType": "image/jpeg", + "meta": { + "still": { + "id": image_id, + "purpose": "meter_close", + "img": f"stills/{name}.jpg", + "w": 2, + "h": 2, + "timestamp": 0, + "orientation": 1, + } + }, + } + registered = client.post(f"/v1/captures/{capture_id}/files", json={"files": [item]}) + assert registered.status_code == 200, registered.text + upload = registered.json()["files"][0]["upload"] + put = TestClient(app).put(upload["url"], content=blob, headers=upload.get("headers") or {}) + assert put.status_code == 200, put.text + committed = client.post( + f"/v1/captures/{capture_id}/files:commit", json={"files": [{"path": item["path"], "sha256": item["sha256"]}]} + ) + assert committed.status_code == 200, committed.text + return blob + + +def _confirm(client: TestClient, capture_id: str, field: str, kind: str, value: str, image_id: str, at: str): + return client.post( + f"/v1/captures/{capture_id}/confirmations", + json={"fieldId": field, "kind": kind, "value": value, "imageId": image_id, "clientAt": at}, + ) + + +def test_confirmation_before_reads_reaches_the_composed_result(journey) -> None: + """A confirmation accepted before reads exist is persisted, applied when the + reads stage creates its record, and served in the composed result — the + client never re-sends an already-accepted answer.""" + root, packet, client, app = journey + deps = app.state.deps + capture_id, raw = _open_front_door(root, packet, client, app, "join-confirm-early") + + # The capture is still open (uploading) when the device confirms the meter. + _register_extra_still(root, client, app, capture_id, "early-meter", "early-1") + accepted = _confirm(client, capture_id, "meter_number", "corrected", "31415", "early-1", "2026-10-15T00:00:00Z") + assert accepted.status_code == 201, accepted.text + first_id = accepted.json()["confirmationId"] + + # The answer persists per field before any run exists. + listed = client.get(f"/v1/captures/{capture_id}/confirmations").json() + fields = [row["fieldId"] for row in listed["confirmations"]] + assert fields == ["meter_number"] + assert listed["confirmations"][0]["value"] == "31415" + + finalized = _finalize(client, capture_id, raw, "join-confirm-early") + drain_finalize(deps) + run = deps.state.get_run(finalized["runId"]) + assert run.status == "done" + + # The composed result carries the early answer. + served = client.get(f"/v1/captures/{capture_id}/result").json() + assert served["meter"]["value"] == "31415" + assert served["meter"]["status"] == "auto_filled" + + # Replaying the same answer is idempotent — same confirmation, one event, no + # new run, no recomposition. + replayed = _confirm(client, capture_id, "meter_number", "corrected", "31415", "early-1", "2026-10-15T00:00:00Z") + assert replayed.status_code == 200, replayed.text + assert replayed.json()["confirmationId"] == first_id + assert len(deps.state.runs_for_capture(capture_id)) == 1 + + events = _flat(client, capture_id) + confirmations = [e for e in events if e["type"] == "confirmation"] + assert [e["data"]["confirmationId"] for e in confirmations] == [first_id] + seqs = [e["seq"] for e in events] + assert seqs == sorted(set(seqs)) and seqs == list(range(1, len(seqs) + 1)) + + +def test_correction_after_the_result_recomposes_the_served_payload(journey) -> None: + """A late correction publishes the acceptance event and the recomposed result: + the run's criteria and result stages re-execute through the normal lifecycle + (one run, the same run id), the served meter carries the corrected value, and + replaying the correction neither runs anything new nor changes the payload.""" + root, packet, client, app = journey + deps = app.state.deps + capture_id, raw = _open_front_door(root, packet, client, app, "join-confirm-late") + accepted = _finalize(client, capture_id, raw, "join-confirm-late") + run_id = accepted["runId"] + drain_finalize(deps) + assert deps.state.get_run(run_id).status == "done" + + baseline = client.get(f"/v1/captures/{capture_id}/result").json() + assert baseline["meter"]["value"] != "777" + base_verdicts = len([e for e in _flat(client, capture_id) if e["type"] == "verdict_ready"]) + + image_id = packet["stills"][0]["id"] + corrected = _confirm(client, capture_id, "meter_number", "corrected", "777", image_id, "2026-10-15T00:01:00Z") + assert corrected.status_code == 201, corrected.text + + # One run, re-driven in place: the recomposed result serves the correction. + runs = deps.state.runs_for_capture(capture_id) + assert len(runs) == 1 and runs[0].run_id == run_id and runs[0].status == "done" + served = client.get(f"/v1/captures/{capture_id}/result").json() + assert served["meter"]["value"] == "777" + assert served["meter"]["status"] == "auto_filled" + + events = _flat(client, capture_id) + confirmations = [e for e in events if e["type"] == "confirmation"] + assert [e["data"]["confirmationId"] for e in confirmations] == [corrected.json()["confirmationId"]] + # The recomposition is visible: the run's result re-published its terminal. + assert len([e for e in events if e["type"] == "verdict_ready"]) == base_verdicts + 1 + seqs = [e["seq"] for e in events] + assert seqs == sorted(set(seqs)) and seqs == list(range(1, len(seqs) + 1)) + + # Replaying the same correction changes nothing further. + replayed = _confirm(client, capture_id, "meter_number", "corrected", "777", image_id, "2026-10-15T00:01:00Z") + assert replayed.status_code == 200, replayed.text + assert client.get(f"/v1/captures/{capture_id}/result").json()["meter"]["value"] == "777" + events_after = _flat(client, capture_id) + assert len([e for e in events_after if e["type"] == "verdict_ready"]) == base_verdicts + 1 + assert [e["data"]["confirmationId"] for e in events_after if e["type"] == "confirmation"] == [ + corrected.json()["confirmationId"] + ] From 926a1a825fe108d19a087dcdfc08a6036f3918f8 Mon Sep 17 00:00:00 2001 From: Obvious Date: Fri, 16 Oct 2026 06:09:01 +0000 Subject: [PATCH 07/20] test(backend): pin finalize failure-publication ordering, not the publication count MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Merged burst7 retries storage-offline background finalize failures (retry_finalize on 503-class), so the failure publishes once per exhausted attempt. The pinned invariant — every publication happens while the worker still owns the slot — is unchanged. Co-authored-by: Sam Gu Co-authored-by: obvious-autobuild[bot] <262744130+obvious-autobuild[bot]@users.noreply.github.com> --- backend/server/api/tests/test_finalize_background.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/backend/server/api/tests/test_finalize_background.py b/backend/server/api/tests/test_finalize_background.py index b9aea5fd6..86ec0fa64 100644 --- a/backend/server/api/tests/test_finalize_background.py +++ b/backend/server/api/tests/test_finalize_background.py @@ -118,8 +118,10 @@ def probing_failure_path(*args, **kwargs): # The failure published while this worker still held the slot: a retry that # starts after the slot frees sees the failure already published (or the - # run_started flag of a healed retry) — never both. - assert publication_slot_state == [True] + # run_started flag of a healed retry) — never both. Storage-offline failures + # are retryable, so the merged finalize republishes per exhausted attempt; + # the pinned invariant is that every publication happens under the slot. + assert publication_slot_state and all(publication_slot_state) assert deps.state.get_capture(capture_id).status == "uploading" # And the healed retry still lands one healthy run. From 64a4803f327e2640572fdf899b39c33548406079 Mon Sep 17 00:00:00 2001 From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com> Date: Fri, 16 Oct 2026 07:35:04 +0000 Subject: [PATCH 08/20] feat(backend): one-command local capture API on loopback (make local-api) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit scripts/local_capture_api.py starts the real app under uvicorn, loopback only, fresh tokens + HMAC secret per run, memory state and volume storage, ingest hooks auto so finalize drives the orchestrator inline; it writes runs/local-api/session.json (0600) with the base URL, tokens, pid, sha. make local-api-smoke drives the mini-arkit packet over real TCP via httpx against the script as a subprocess — no TestClient, no drain_finalize. Co-authored-by: Obvious --- backend/Makefile | 13 + backend/docs/local-capture-api.md | 101 +++++++ backend/scripts/local_capture_api.py | 221 +++++++++++++++ backend/tests/test_local_capture_api.py | 359 ++++++++++++++++++++++++ 4 files changed, 694 insertions(+) create mode 100644 backend/docs/local-capture-api.md create mode 100644 backend/scripts/local_capture_api.py create mode 100644 backend/tests/test_local_capture_api.py diff --git a/backend/Makefile b/backend/Makefile index 330a0e613..55995389d 100644 --- a/backend/Makefile +++ b/backend/Makefile @@ -86,6 +86,19 @@ e2e-dev: PYTHONPATH=. uv run --frozen python scripts/e2e_dev.py --self-check PYTHONPATH=. uv run --frozen python scripts/e2e_dev.py $(if $(PACKET),--packet $(PACKET),) +# Burst11: the one-command local capture API. Serves loopback only (127.0.0.1); +# the iOS simulator shares the Mac's loopback, a physical phone does not. +# Writes backend/runs/local-api/session.json (mode 0600, gitignored) with the +# base URL, app and reviewer tokens, pid and git sha. Stop with Ctrl-C. +local-api: + uv run --frozen --extra api python scripts/local_capture_api.py --host 127.0.0.1 --port $(PORT) + +# Burst11: drives the committed mini-arkit packet against the local capture API +# over a real TCP socket — the launcher script starts as a subprocess on an +# ephemeral port, no TestClient. +local-api-smoke: + uv run --frozen --extra api pytest tests/test_local_capture_api.py -q + orchestrator-e2e: STATE_BACKEND=memory STORAGE_BACKEND=local HSS_ORCH_STAGE_STUBS=1 VISION_EXTERNAL_CALLS=off SMP_ENVIRONMENT=dev \ uv run --frozen pytest server/orchestrator/tests -q diff --git a/backend/docs/local-capture-api.md b/backend/docs/local-capture-api.md new file mode 100644 index 000000000..24255aaeb --- /dev/null +++ b/backend/docs/local-capture-api.md @@ -0,0 +1,101 @@ +# The local capture API: one command, loopback only + +`make local-api` gives a reviewer, the iOS simulator and `tools/hunter-local-ingest` a working +capture API on `127.0.0.1` from a fresh checkout. It replaces the hand-assembled uvicorn command in +`server/api/README.md`, which ran with `HSS_INGEST_HOOKS=null` — so finalize never reached the +orchestrator and no run ever started. + +## Fresh checkout + +```bash +git clone https://github.com/SamGu-NRX/BaseScanning.git && cd BaseScanning +git checkout alpha +cd backend +uv sync --frozen --extra api +make local-api # PORT=8000 default; override with PORT=8123 make local-api +``` + +Wait for the banner: + +``` +local capture API ready: http://127.0.0.1:8000 +session file: .../backend/runs/local-api/session.json +stand-in stages: validate, poses, objects, reads, scale, dense, coverage, scene, objects_3d, criteria, result +real stage: placement +``` + +Stop with Ctrl-C. Every run generates fresh tokens; nothing persists across restarts except the +uploaded bytes under `runs/local-api/volume/`. + +`make local-api-smoke` proves the same launcher end to end: it starts the script as a subprocess on +an ephemeral port and drives the committed `data/fixtures/packets/mini-arkit` packet over real TCP +with httpx — create, register, PUT to the signed URLs, commit, finalize, long-poll events to +terminal `manual_review`, read the result and the reviewer page — plus the auth and error +boundaries. No TestClient, no `drain_finalize`: the orchestrator is driven inline by `start_run`. + +## Environment + +| Variable | Value set by `make local-api` | Why | +|---|---|---| +| `SMP_ENVIRONMENT` | `dev` | dev posture; no cloud credentials needed | +| `HSS_AUTH_MODE` | `static` | bearer tokens from `CAPTURE_API_TOKENS` (explicit, because dev otherwise defaults to `none`) | +| `CAPTURE_API_TOKENS` | `device:,reviewer:local-review:` | generated per run; tokens are ≥32 chars (`server/api/auth.py`) | +| `UPLOAD_URL_HMAC_SECRET` | fresh hex per run | signs the volume upload tokens | +| `STATE_BACKEND` | `memory` | no Modal, no database | +| `STORAGE_BACKEND` | `volume` | signed PUT URLs a client can actually fetch: `{CAPTURE_API_PUBLIC_URL}/v1/uploads/{token}` (`storage_volume.py`), unlike `storage_local.py`'s `file://` URLs | +| `VOLUME_ROOT` | `backend/runs/local-api/volume` | uploaded bytes live under gitignored `runs/` | +| `CAPTURE_API_PUBLIC_URL` | `http://127.0.0.1:` | derived from the bind address, so upload URLs are reachable on this machine | +| `HSS_INGEST_HOOKS` | `auto` | finalize reaches the orchestrator (this is the point) | +| `HSS_ORCH_STAGE_STUBS` | `1` | committed stage stubs instead of GPU/vision work | +| `HSS_ORCH_PREWARM` | `0` | no prewarm work on capture create | +| `VISION_EXTERNAL_CALLS` | `off` | no networked model calls | +| `BASE_WEBHOOK_URL` / `WEBHOOK_HMAC_SECRET` | unset | webhooks off | + +Pass through without interpretation: `HSS_LOCAL_OUTCOME` and `HSS_LOCAL_STAGE_DELAY_S`. The stage +stubs honor them when the burst11-bs-local-outcome-states packet lands; this launcher never reads +or changes them, so `HSS_LOCAL_OUTCOME=pass make local-api` just works. + +## The session file + +`backend/runs/local-api/session.json`, mode `0600` (the directory `runs/` is gitignored): + +```json +{ + "baseUrl": "http://127.0.0.1:8000/v1", + "appToken": "", + "reviewerToken": "", + "pid": 12345, + "gitSha": "", + "stageStubs": true, + "startedAt": "" +} +``` + +The shape is pinned verbatim by `tests/test_local_capture_api.py` because the iOS run and later +packets read it. Two choices to note: the tokens are written **inline** (`appToken`, not +`appTokenFile`) because a single self-contained file is one less path for every consumer to resolve +against an unknown working directory; and `gitSha` is the full head SHA so a captured session always +identifies the code that served it. Tokens appear nowhere else — not in stdout, not in the uvicorn +access log (which keeps upload URLs as `/v1/uploads/{token}` route shapes anyway). + +## What is a stand-in + +`HSS_ORCH_STAGE_STUBS=1` runs every one of the twelve orchestrator stages as its committed stub +**except `placement`**, which is `stubs_exempt` (deterministic CPU work, no GPU/network/credentials) +and solves for real. A terminal `manual_review` verdict therefore proves the whole HTTP + finalize + +orchestrator path, not the quality of the vision stages. + +## The iOS simulator + +The simulator shares the Mac's loopback, so it talks to `http://127.0.0.1:8000/v1` directly — set +the app's capture API base URL and paste the `appToken` from the session file. **A physical phone +does not share the Mac's loopback** and is out of scope: the server refuses any non-loopback bind, +by design. + +## tools/hunter-local-ingest + +Point it at the session: read `backend/runs/local-api/session.json`, use `baseUrl` as the API base, +`appToken` as the bearer on every request, and register uploads from the packet directory exactly as +the smoke test does. The review page needs the `reviewerToken` bearer: +`GET {baseUrl}/captures/{captureId}/review` with `Authorization: Bearer ` returns the +rendered HTML. diff --git a/backend/scripts/local_capture_api.py b/backend/scripts/local_capture_api.py new file mode 100644 index 000000000..baa3e1e10 --- /dev/null +++ b/backend/scripts/local_capture_api.py @@ -0,0 +1,221 @@ +#!/usr/bin/env python3 +"""One-command local capture API for reviewers, the iOS simulator and harness scripts. + +`make local-api` (backend/) starts the real HTTP app `server.api.app:create_app` +under uvicorn, bound to loopback only, with generated tokens and a fresh HMAC +secret, and writes the session the iOS run and later packets read: + + backend/runs/local-api/session.json (mode 0600; runs/ is gitignored) + +The journey is the documented front door: create a capture, register files, +PUT bytes to the signed upload URLs, commit, finalize `packet.json`, long-poll +events, then read the result and the reviewer page — over a real TCP socket, +unlike `scripts/capture_flow_e2e.py`, which drives TestClient in process. + +Loopback only: the server refuses any `--host` other than 127.0.0.1 or ::1, and +`CAPTURE_API_PUBLIC_URL` (which signs the upload URLs) is derived from the same +host, so it can never point a client at a routable address. The iOS simulator +shares the Mac's loopback, so it reaches 127.0.0.1 directly; a physical phone +does not and is out of scope. + +Stand-in stages: `HSS_ORCH_STAGE_STUBS=1` runs every orchestrator stage as its +committed stub except `placement`, which is `stubs_exempt` and solves for real. +Pass-through, without interpretation: `HSS_LOCAL_OUTCOME` and +`HSS_LOCAL_STAGE_DELAY_S` are honored by the stage stubs when the +burst11-bs-local-outcome-states packet lands; this script neither reads nor +changes them. +""" + +from __future__ import annotations + +import argparse +import json +import os +import secrets +import socket +import subprocess +import sys +import threading +import time +import urllib.error +import urllib.request +from datetime import UTC, datetime +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +LOOPBACK_HOSTS = {"127.0.0.1", "::1"} + +# Twelve-stage graph (server/orchestrator/graph.py). With HSS_ORCH_STAGE_STUBS=1 +# every one of these is a committed stub; `placement` alone is stubs_exempt. +STAND_IN_STAGES = ( + "validate", + "poses", + "objects", + "reads", + "scale", + "dense", + "coverage", + "scene", + "objects_3d", + "criteria", + "result", +) +REAL_STAGES = ("placement",) + +SESSION_DEFAULT = ROOT / "runs" / "local-api" + + +def _git_sha() -> str: + env = os.environ.get("GIT_SHA") + if env: + return env + try: + return subprocess.run( + ["git", "rev-parse", "HEAD"], + cwd=ROOT.parent, + capture_output=True, + text=True, + check=True, + ).stdout.strip() + except Exception: + return "unknown" + + +def _free_port() -> int: + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +def _public_url(host: str, port: int) -> str: + host_part = f"[{host}]" if ":" in host else host + return f"http://{host_part}:{port}" + + +def _write_session(path: Path, session: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "w") as handle: + json.dump(session, handle, indent=2) + handle.write("\n") + os.chmod(path, 0o600) + + +def _wait_health(base: str, timeout_s: float) -> bool: + deadline = time.monotonic() + timeout_s + while time.monotonic() < deadline: + try: + with urllib.request.urlopen(f"{base}/v1/healthz", timeout=2) as response: + if response.status == 200: + return True + except (urllib.error.URLError, TimeoutError, OSError): + time.sleep(0.25) + return False + + +def _announce(base: str, session_path: Path, timeout_s: float) -> None: + """Print the banner once /v1/healthz answers. Runs on a side thread while + uvicorn serves on the main thread, so the tokens stay off the terminal.""" + + def _ready() -> None: + if _wait_health(base, timeout_s): + standins = ", ".join(STAND_IN_STAGES) + print(f"\nlocal capture API ready: {base}", flush=True) + print(f"session file: {session_path}", flush=True) + print(f"stand-in stages: {standins}", flush=True) + print(f"real stage: {', '.join(REAL_STAGES)}", flush=True) + print("the app and reviewer tokens are in the session file only, never on stdout", flush=True) + else: + print(f"\nlocal capture API failed its health check within {timeout_s:.0f}s; stopping", flush=True) + os._exit(1) + + threading.Thread(target=_ready, daemon=True).start() + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--host", default="127.0.0.1", help="bind address; loopback only (127.0.0.1 or ::1)") + parser.add_argument("--port", type=int, default=8000, help="port (0 picks a free one)") + parser.add_argument( + "--session-dir", + type=Path, + default=SESSION_DEFAULT, + help="directory for session.json and the volume root (default: backend/runs/local-api)", + ) + parser.add_argument("--health-timeout", type=float, default=60.0, help="seconds to wait for /v1/healthz") + args = parser.parse_args(argv) + + if args.host not in LOOPBACK_HOSTS: + print( + f"refusing to bind {args.host!r}: the local capture API serves loopback only " + f"(127.0.0.1 or ::1). A phone that is not the simulator cannot reach this server anyway.", + file=sys.stderr, + ) + return 2 + + port = args.port if args.port > 0 else _free_port() + base = _public_url(args.host, port) + + # Fresh per-run credentials: a device bearer for the app and harness, an + # explicit reviewer bearer for the review page (server/api/auth.py reads + # `reviewer: