From d5189055356fbe0e08997dbb7929fc96262747c5 Mon Sep 17 00:00:00 2001 From: Obvious Date: Thu, 15 Oct 2026 09:52:21 +0000 Subject: [PATCH 01/12] 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) --- 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 a5a07e6afec4f7c96c8853ba64c851ee20f22e88 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/12] 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) --- 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 2ce3533d6604ebdc0f69f5b125ead3969c7655c2 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/12] 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 9935ec1c96fee8e3ba4a5f47efce19935ce418ff Mon Sep 17 00:00:00 2001 From: Obvious Date: Fri, 16 Oct 2026 01:02:36 +0000 Subject: [PATCH 04/12] 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 --- 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 020d729fe60d402db50477d9eb40380aa1da10a2 Mon Sep 17 00:00:00 2001 From: Obvious Date: Fri, 16 Oct 2026 01:02:36 +0000 Subject: [PATCH 05/12] 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 --- 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 607697fffcc12d2c4a47db379752afc71a6bcee1 Mon Sep 17 00:00:00 2001 From: Obvious Date: Fri, 16 Oct 2026 05:54:44 +0000 Subject: [PATCH 06/12] 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 --- 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 201a17549c60e8c48cdeed0bfdfd3d0723d2a53b Mon Sep 17 00:00:00 2001 From: Obvious Date: Fri, 16 Oct 2026 06:09:01 +0000 Subject: [PATCH 07/12] 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 --- 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 86ad280b15d255f77d7de2b5936279a5a90c2419 Mon Sep 17 00:00:00 2001 From: SamGu-NRX Date: Fri, 16 Oct 2026 02:47:06 -0500 Subject: [PATCH 08/12] test(backend): failing regressions for the #278 review findings - finalize: one storage exception publishes one failure while the worker still holds the slot (inflight, not merely present in _slots) - result stage: a recomposition after an accepted confirmation publishes a new final event; a retry of either composition does not - commit route: a busy index lease commits nothing, and files committed before the lease went busy are still announced - journey: a correction after the run's original deadline still recomposes - journey: the cursor poller drains to the producers' final cursor under a deadline instead of stopping when the producers stop Co-Authored-By: Claude Opus 5.5 --- backend/server/api/tests/test_commit_lease.py | 107 ++++++++++++++++++ .../api/tests/test_finalize_background.py | 15 +-- backend/server/result/tests/test_compose.py | 71 ++++++++++++ backend/tests/test_runtime_join_journey.py | 43 ++++++- 4 files changed, 223 insertions(+), 13 deletions(-) create mode 100644 backend/server/api/tests/test_commit_lease.py diff --git a/backend/server/api/tests/test_commit_lease.py b/backend/server/api/tests/test_commit_lease.py new file mode 100644 index 000000000..33a82b74b --- /dev/null +++ b/backend/server/api/tests/test_commit_lease.py @@ -0,0 +1,107 @@ +"""A busy file-index lease must not strand committed files unannounced. + +ModalDictState merges the per-capture file index under a strict lease; a lease +that stays busy raises LeaseBusy and the commit route answers a retryable 503. +Whatever the failed request already committed must still reach +``files_committed`` and ``on_files_committed``, and whatever it did not commit +must stay pending so the retry commits and announces it. +""" + +from __future__ import annotations + +import time +from urllib.parse import urlparse + +import pytest + +from server.api.tests.conftest import sha256_hex +from server.api.tests.test_ingest import _create, _file +from server.common.backends import state_modal_dict +from server.common.backends.state_modal_dict import ModalDictState +from server.common.lease import LeaseBusy + +_BLOBS = {"stills/s1.jpg": b"jpeg-one", "stills/s2.jpg": b"jpeg-two"} + + +@pytest.fixture +def shared(client, monkeypatch): + deps = client.app_state + deps.state = ModalDictState() + monkeypatch.setattr(state_modal_dict, "_INDEX_LEASE_WAIT_S", 0.05) + announced: list[list[str]] = [] + monkeypatch.setattr( + deps.hooks, "on_files_committed", lambda _capture_id, files: announced.append([f.path for f in files]) + ) + return deps, announced + + +def _upload_all(client, capture_id: str) -> None: + files = [_file(path, data, meta={"still": f"s{index + 1}"}) for index, (path, data) in enumerate(_BLOBS.items())] + registered = client.post(f"/v1/captures/{capture_id}/files", json={"files": files}) + assert registered.status_code == 200, registered.text + for item, data in zip(registered.json()["files"], _BLOBS.values(), strict=True): + target = item["upload"] + put = client.put(urlparse(target["url"]).path, content=data, headers=target["headers"]) + assert put.status_code == 200, put.text + + +def _commit(client, capture_id: str): + return client.post( + f"/v1/captures/{capture_id}/files:commit", + json={"files": [{"path": path, "sha256": sha256_hex(data)} for path, data in _BLOBS.items()]}, + ) + + +def _committed_events(client, capture_id: str) -> list[str]: + events = client.get(f"/v1/captures/{capture_id}/events").json()["events"] + return sorted(path for e in events if e["type"] == "files_committed" for path in e["data"]["paths"]) + + +def test_a_busy_lease_saves_nothing_and_the_retry_announces(client, shared): + deps, announced = shared + capture_id = _create(client, "packet-lease-busy") + _upload_all(client, capture_id) + # Another container holds a fresh lease on this capture's file index. + deps.state.put_if_absent(f"lease:files:{capture_id}", {"id": "other", "at": time.time()}) + + busy = _commit(client, capture_id) + assert busy.status_code == 503, busy.text + assert busy.json()["errors"][0]["code"] == "lease_busy" + for path in _BLOBS: + assert deps.state.get_file(capture_id, path).state == "pending", "nothing may commit without the lease" + + deps.state.delete_kv(f"lease:files:{capture_id}") + retried = _commit(client, capture_id) + assert retried.status_code == 200, retried.text + assert _committed_events(client, capture_id) == sorted(_BLOBS) + assert sorted(path for batch in announced for path in batch) == sorted(_BLOBS) + assert sorted(rec.path for rec in deps.state.files(capture_id)) == sorted(_BLOBS) + + +def test_files_committed_before_a_busy_lease_are_still_announced(client, shared, monkeypatch): + deps, announced = shared + capture_id = _create(client, "packet-lease-partial") + _upload_all(client, capture_id) + + real_with_lease = state_modal_dict.with_lease + calls = {"n": 0} + + def second_commit_busy(state, name, fn, **kwargs): + calls["n"] += 1 + if calls["n"] == 2: + raise LeaseBusy(f"lease {name!r} stayed busy") + return real_with_lease(state, name, fn, **kwargs) + + monkeypatch.setattr(state_modal_dict, "with_lease", second_commit_busy) + busy = _commit(client, capture_id) + assert busy.status_code == 503, busy.text + # The first file committed before the lease went busy: it is announced now, + # because the retry will read it as already committed. + assert _committed_events(client, capture_id) == ["stills/s1.jpg"] + assert deps.state.get_file(capture_id, "stills/s2.jpg").state == "pending" + + monkeypatch.setattr(state_modal_dict, "with_lease", real_with_lease) + retried = _commit(client, capture_id) + assert retried.status_code == 200, retried.text + assert _committed_events(client, capture_id) == sorted(_BLOBS) + assert sorted(path for batch in announced for path in batch) == sorted(_BLOBS) diff --git a/backend/server/api/tests/test_finalize_background.py b/backend/server/api/tests/test_finalize_background.py index 86ec0fa64..6a04c1613 100644 --- a/backend/server/api/tests/test_finalize_background.py +++ b/backend/server/api/tests/test_finalize_background.py @@ -102,7 +102,7 @@ def offline(_key): original_failure_path = finalize_module._on_background_failure def probing_failure_path(*args, **kwargs): - publication_slot_state.append(capture_id in finalize_module._slots) + publication_slot_state.append(finalize_module._slots[capture_id].inflight) return original_failure_path(*args, **kwargs) monkeypatch.setattr(deps.storage, "get_bytes", offline) @@ -116,12 +116,13 @@ def probing_failure_path(*args, **kwargs): 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. 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) + # One storage exception publishes one failure, while this worker still holds + # 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 finalize_module._slots[capture_id].inflight is False + events = client.get(f"/v1/captures/{capture_id}/events").json()["events"] + assert len([e for e in events if e["type"] == "failed" and e["data"].get("code") == "finalize_failed"]) == 1 assert deps.state.get_capture(capture_id).status == "uploading" # And the healed retry still lands one healthy run. diff --git a/backend/server/result/tests/test_compose.py b/backend/server/result/tests/test_compose.py index 657874caa..5d6b7c1f4 100644 --- a/backend/server/result/tests/test_compose.py +++ b/backend/server/result/tests/test_compose.py @@ -616,3 +616,74 @@ def test_route_skips_the_duplicate_point_when_the_spot_starts_at_the_meter(): assert pts is not None and len(pts) >= 2 for index in range(len(pts) - 1): assert pts[index] != pts[index + 1] + + +def test_a_correction_republishes_the_final_event_and_a_retry_does_not(tmp_path, monkeypatch): + """A late correction re-runs the result stage under the same run id. A retry of + the stage must not publish a second final event, but a recomposition after an + accepted confirmation must, or clients following the cursor never see the + corrected outcome.""" + for name in ("WEBHOOK_HMAC_SECRET", "BASE_WEBHOOK_URL", "SHOW_REJECT_TO_HOMEOWNER", "HSS_RESULT_PROMPTS_PATH"): + monkeypatch.delenv(name, raising=False) + capture, run_rec, epochs = _capture_run() + state = MemoryState() + state.put_capture(capture) + state.put_run(run_rec) + + class _Settings: + RULES_PROFILE_DEFAULT = "C" + SHOW_REJECT_TO_HOMEOWNER = False + HSS_RESULT_MAX_VIEWS = 6 + HSS_RESULT_SIGNED_URL_TTL_S = 900 + HSS_RESULT_PREVIEW_MAX_POINTS = 200_000 + HSS_RESULT_METER_AUTOFILL = "barcode" + HSS_RESULT_REVIEW_THREEJS_URL = "https://unpkg.com/three@0.160.0/build/three.module.js" + HSS_RESULT_PROMPTS_PATH = str(DATA / "prompts.en-US.json") + BASE_WEBHOOK_URL = None + + ctx = StageContext( + capture_id=capture.capture_id, + run_id=run_rec.run_id, + stage="result", + attempt=1, + branch="lidar", + flow="guided", + tier="arkit", + work_dir=tmp_path, + inputs={ + "scene": str(DATA / "synthetic"), + "verdicts": str(DATA / "synthetic" / "verdicts"), + "reads": str(DATA / "reads.auto_fill.json"), + "out": str(tmp_path / "out"), + "epochs": ",".join(epochs), + }, + deadline_at="2026-09-27T00:00:00Z", + heartbeat=lambda _step: None, + vlm_budget_usd=0, + storage=None, + state=state, + settings=_Settings(), + ) + + def finals() -> list[int]: + return [ + ev.seq + for ev in state.events_after(capture.capture_id, 0) + if ev.type in {"verdict_ready", "retake_request"} and ev.run_id == run_rec.run_id + ] + + run(ctx) + run(ctx) # a retry of the same composition + first = finals() + assert len(first) == 1 + + state.append_event( + capture.capture_id, + "confirmation", + {"confirmationId": "cfm_1", "fieldId": "meter_number", "imageId": "s1", "kind": "corrected", "by": "device"}, + ) + run(ctx) # the recomposition the correction triggers + run(ctx) # and a retry of that recomposition + second = finals() + assert len(second) == 2, "the correction must publish its own final event, once" + assert second[0] == first[0] diff --git a/backend/tests/test_runtime_join_journey.py b/backend/tests/test_runtime_join_journey.py index 9aaf82f5d..662083ca9 100644 --- a/backend/tests/test_runtime_join_journey.py +++ b/backend/tests/test_runtime_join_journey.py @@ -289,17 +289,21 @@ def produce(slot: int) -> None: poller_pages: list[list[dict]] = [] poller_after: list[int] = [] - poller_stop = threading.Event() + final_cursor = base_count + 60 def poll() -> None: + # Poll until the cursor reaches the producers' last event, not until the + # producers stop: a poller that stops with them can exit before draining + # the tail, and the assertion below would blame the cursor for it. start.wait() after = 0 - while not poller_stop.is_set(): + give_up = time.monotonic() + 10.0 + while after < final_cursor and time.monotonic() < give_up: 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) + time.sleep(0.02) else: after = page["next"] @@ -312,8 +316,7 @@ def poll() -> None: thread.start() for thread in threads[:2]: thread.join() - poller_stop.set() - threads[2].join(timeout=10) + threads[2].join(timeout=15) assert not threads[2].is_alive() # Each producer drew a distinct slice of the sequence — never the same seq. @@ -323,7 +326,7 @@ def poll() -> None: 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)) + expected = list(range(1, final_cursor + 1)) assert seqs == expected, "the cursor skipped an event a producer appended" # The store itself holds the complete, contiguous sequence. @@ -600,3 +603,31 @@ def test_correction_after_the_result_recomposes_the_served_payload(journey) -> N assert [e["data"]["confirmationId"] for e in events_after if e["type"] == "confirmation"] == [ corrected.json()["confirmationId"] ] + + +def test_a_correction_after_the_original_deadline_gets_its_own_budget(journey) -> None: + """The run's deadline measured its first pass. A correction accepted after that + deadline re-opens the run; without a fresh budget the next tick hard-stops it + and the capture falls to manual review with the old result still served.""" + root, packet, client, app = journey + deps = app.state.deps + capture_id, raw = _open_front_door(root, packet, client, app, "join-confirm-expired") + run_id = _finalize(client, capture_id, raw, "join-confirm-expired")["runId"] + drain_finalize(deps) + run = deps.state.get_run(run_id) + assert run.status == "done" + failure_before = run.failure_code + # The original 20-minute deadline has passed by the time the correction lands. + run.deadline_at = "2000-01-01T00:00:00+00:00" + deps.state.put_run(run) + status_before = deps.state.get_capture(capture_id).status + + image_id = packet["stills"][0]["id"] + corrected = _confirm(client, capture_id, "meter_number", "corrected", "4242", image_id, "2026-10-15T00:02:00Z") + assert corrected.status_code == 201, corrected.text + + rerun = deps.state.get_run(run_id) + assert rerun.status == "done", (rerun.status, rerun.failure_code) + assert rerun.failure_code == failure_before + assert deps.state.get_capture(capture_id).status == status_before + assert client.get(f"/v1/captures/{capture_id}/result").json()["meter"]["value"] == "4242" From d392105556a9957ca37d2eae7027d8d8db4100e9 Mon Sep 17 00:00:00 2001 From: SamGu-NRX Date: Fri, 16 Oct 2026 02:47:30 -0500 Subject: [PATCH 09/12] fix(backend): publish an escaped finalize failure once, under the slot The merge kept both copies of _guard's failure path. The first holds the slot and finish_sync across the run_started check and the write; the second ran after _clear_slot without the lock, so one storage exception published two failures and the unlocked copy could still overwrite a healed retry. Drop the second copy. Co-Authored-By: Claude Opus 5.5 --- backend/server/api/finalize.py | 12 ------------ 1 file changed, 12 deletions(-) diff --git a/backend/server/api/finalize.py b/backend/server/api/finalize.py index 54fb368f1..c0ea67811 100644 --- a/backend/server/api/finalize.py +++ b/backend/server/api/finalize.py @@ -519,18 +519,6 @@ def _guard(deps: Deps, capture_id: str, future: Future[None]) -> None: ) _on_background_failure(deps, capture_id, failure) _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) From 5920a2c5e14309aa6c6cab7b996eb9714c2fa149 Mon Sep 17 00:00:00 2001 From: SamGu-NRX Date: Fri, 16 Oct 2026 02:47:51 -0500 Subject: [PATCH 10/12] fix(backend): a late correction re-runs on a fresh deadline Reopening a finished run for a correction kept the run's original deadline_at. Once that had passed (20 minutes by default) the next tick hard-stopped the run and the capture fell to manual review with the old result. The correction now gets the HARD_STOP_MIN budget a new run gets, counted from when it is accepted. Runs that are not corrected keep their deadline. Co-Authored-By: Claude Opus 5.5 --- backend/server/orchestrator/api.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/backend/server/orchestrator/api.py b/backend/server/orchestrator/api.py index d54a16887..8f5990498 100644 --- a/backend/server/orchestrator/api.py +++ b/backend/server/orchestrator/api.py @@ -181,6 +181,10 @@ def _recompose_after_correction(capture_id: str, run: RunRecord) -> None: ) ) run.status = "running" + # The run's deadline measured its first pass, so a correction accepted after + # it would hard-stop on the next tick. The re-run gets the same HARD_STOP_MIN + # budget start_run gives a new run, counted from the correction. + run.deadline_at = deadline_from(services.clock, services.settings.HARD_STOP_MIN) 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 From eb145718cdbc2157f6a8763797482a91e77c387c Mon Sep 17 00:00:00 2001 From: SamGu-NRX Date: Fri, 16 Oct 2026 02:48:18 -0500 Subject: [PATCH 11/12] fix(backend): a corrected result publishes its new final event A late correction recomposes the result under the same run id, and emit_events skipped publishing whenever the run already had a final event, so clients following the cursor never saw the corrected outcome. A final event older than the capture's latest confirmation is now stale: the recomposition publishes its own, while a retry of either composition still returns the seqs already published. The check reads every event page, since events_after returns at most one page. Co-Authored-By: Claude Opus 5.5 --- backend/server/result/compose.py | 24 +++++++++++++++++++++--- 1 file changed, 21 insertions(+), 3 deletions(-) diff --git a/backend/server/result/compose.py b/backend/server/result/compose.py index 268830662..7b403263e 100644 --- a/backend/server/result/compose.py +++ b/backend/server/result/compose.py @@ -12,7 +12,7 @@ import numpy as np from pydantic import ValidationError -from server.contracts.jobs import CaptureRecord, RunRecord, StageContext, StageResult +from server.contracts.jobs import CaptureRecord, EventRecord, RunRecord, StageContext, StageResult from server.contracts.result import ( BoxArkitWorld, BoxSceneM, @@ -254,9 +254,17 @@ def compose_result( def emit_events(result: ComposeResult, state) -> list[int]: - prior = state.events_after(result.capture_id, 0) + """Publish the run's final event once per composition. + + A retry of the result stage finds its final event and returns those seqs. A + late correction re-runs the stage under the same run id, and its confirmation + event lands after the old final: finals older than the latest confirmation + are stale, so the recomposed result publishes its own. + """ + prior = _all_events(state, result.capture_id) finals = {"verdict_ready", "retake_request"} - already = [ev for ev in prior if ev.run_id == result.run_id and ev.type in finals] + corrected = max((ev.seq for ev in prior if ev.type == "confirmation"), default=0) + already = [ev for ev in prior if ev.run_id == result.run_id and ev.type in finals and ev.seq > corrected] if already: return [ev.seq for ev in already] seqs = [] @@ -266,6 +274,16 @@ def emit_events(result: ComposeResult, state) -> list[int]: return seqs +def _all_events(state, capture_id: str) -> list[EventRecord]: + """Every event on the capture; events_after returns one page at a time.""" + events: list[EventRecord] = [] + while True: + page = state.events_after(capture_id, events[-1].seq if events else 0) + if not page: + return events + events.extend(page) + + def _stage_input(ctx: StageContext, short: str, key: str) -> Path | None: """The short input key wins; orchestrated runs hydrate `work_dir/` instead.""" From 6f57242fca2437ced31ecd685d643ed5d4f07887 Mon Sep 17 00:00:00 2001 From: SamGu-NRX Date: Fri, 16 Oct 2026 02:49:15 -0500 Subject: [PATCH 12/12] fix(backend): a busy index lease no longer strands committed files upsert_file saved the file record as committed before taking the strict index lease, so LeaseBusy left a committed record outside the index and the commit route exited before files_committed and on_files_committed; the retry read the file as committed and never announced it. The record is now written inside the lease, so a busy lease saves nothing. The commit route also announces whatever the request did commit before a later file failed (busy lease or storage head error), since the retry skips those. Co-Authored-By: Claude Opus 5.5 --- backend/server/api/routes/files.py | 113 ++++++++++-------- .../common/backends/state_modal_dict.py | 5 +- 2 files changed, 65 insertions(+), 53 deletions(-) diff --git a/backend/server/api/routes/files.py b/backend/server/api/routes/files.py index 10037feec..1dbb69a05 100644 --- a/backend/server/api/routes/files.py +++ b/backend/server/api/routes/files.py @@ -289,58 +289,48 @@ async def commit_files(capture_id: str, body: CommitRequest, request: Request): newly: list[FileRecord] = [] extra = extras(deps, capture_id) size_only = set(extra.get("size_only_paths") or []) - for item in body.files: - rec = deps.state.get_file(capture_id, item.path) - if rec is None: - not_found.append(item.path) - continue - if rec.state == "committed": - committed.append(item.path) - continue - spec = (extra.get("digests") or {}).get(item.path) - if spec and (rec.sha256 != spec["sha256"] or rec.bytes != spec["bytes"]): - mismatch.append(item.path) - continue - try: - meta = deps.storage.head(upload_key(capture_id, item.path)) - except FileNotFoundError: - not_found.append(item.path) - continue - except Exception as exc: - log_event("storage.error", op="head", capture_id=capture_id, level="error") - raise fail(503, "storage_unavailable", "/", "storage head failed", **{"Retry-After": "5"}) from exc - md5_bad = item.path not in size_only and meta.md5_b64 not in (None, rec.md5) - if meta.bytes != rec.bytes or md5_bad: - rec.state = "mismatch" + # A failure partway through the batch (a busy index lease, a storage + # head error) must still announce the files already committed: the + # retry reads them as committed and would never announce them. + try: + for item in body.files: + rec = deps.state.get_file(capture_id, item.path) + if rec is None: + not_found.append(item.path) + continue + if rec.state == "committed": + committed.append(item.path) + continue + spec = (extra.get("digests") or {}).get(item.path) + if spec and (rec.sha256 != spec["sha256"] or rec.bytes != spec["bytes"]): + mismatch.append(item.path) + continue + try: + meta = deps.storage.head(upload_key(capture_id, item.path)) + except FileNotFoundError: + not_found.append(item.path) + continue + except Exception as exc: + log_event("storage.error", op="head", capture_id=capture_id, level="error") + raise fail( + 503, "storage_unavailable", "/", "storage head failed", **{"Retry-After": "5"} + ) from exc + md5_bad = item.path not in size_only and meta.md5_b64 not in (None, rec.md5) + if meta.bytes != rec.bytes or md5_bad: + rec.state = "mismatch" + deps.state.upsert_file(rec) + mismatch.append(item.path) + continue + rec.state = "committed" + rec.generation = meta.generation + rec.key = upload_key(capture_id, item.path) + rec.committed_at = isoformat(deps.now()) deps.state.upsert_file(rec) - mismatch.append(item.path) - continue - rec.state = "committed" - rec.generation = meta.generation - rec.key = upload_key(capture_id, item.path) - rec.committed_at = isoformat(deps.now()) - deps.state.upsert_file(rec) - committed.append(item.path) - newly.append(rec) - if newly: - listed = (extras(deps, capture_id).get("accepted") or {}).get("listed") - payload = FilesCommittedData( - paths=[rec.path for rec in newly], committed=len(newly), listed=listed - ).model_dump(mode="json") - emit(deps, capture_id, "files_committed", payload) - hooked = [ - CommittedFile( - path=rec.path, - role=rec.role, - sha256=rec.sha256, - bytes=rec.bytes, - generation=rec.generation or "", - key=rec.key or upload_key(capture_id, rec.path), - meta=rec.meta, - ) - for rec in newly - ] - safe_hook("on_files_committed", deps.hooks.on_files_committed, capture_id, hooked) + committed.append(item.path) + newly.append(rec) + finally: + if newly: + _announce_committed(deps, capture_id, newly) maybe_finish_after_commit(deps, deps.state.get_capture(capture_id) or record) log_event( "files.committed", n=len(newly), mismatch=len(mismatch), not_found=len(not_found), capture_id=capture_id @@ -356,3 +346,24 @@ async def complete_multipart(capture_id: str, body: CompleteMultipartRequest, re async with deps.locks.acquire(capture_id): load_owned(deps, capture_id, principal, write=True) raise fail(501, "multipart_not_implemented", "/", "multipart upload is not implemented in this PR") + + +def _announce_committed(deps, capture_id: str, newly: list[FileRecord]) -> None: + listed = (extras(deps, capture_id).get("accepted") or {}).get("listed") + payload = FilesCommittedData(paths=[rec.path for rec in newly], committed=len(newly), listed=listed).model_dump( + mode="json" + ) + emit(deps, capture_id, "files_committed", payload) + hooked = [ + CommittedFile( + path=rec.path, + role=rec.role, + sha256=rec.sha256, + bytes=rec.bytes, + generation=rec.generation or "", + key=rec.key or upload_key(capture_id, rec.path), + meta=rec.meta, + ) + for rec in newly + ] + safe_hook("on_files_committed", deps.hooks.on_files_committed, capture_id, hooked) diff --git a/backend/server/common/backends/state_modal_dict.py b/backend/server/common/backends/state_modal_dict.py index ca3733b78..b6ad39ea4 100644 --- a/backend/server/common/backends/state_modal_dict.py +++ b/backend/server/common/backends/state_modal_dict.py @@ -340,9 +340,10 @@ def delete_capture(self, capture_id: str) -> None: self._kv[tomb] = {"capture_id": capture_id} def upsert_file(self, rec: FileRecord) -> None: - self._kv[f"file:{rec.capture_id}:{rec.path}"] = rec.model_dump(mode="json") - def merge() -> None: + # The record is written inside the lease too: a busy lease then saves + # nothing, so the caller's retry still sees the file in its old state. + 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)