[obvious-fleet] Backend burst8 integration: one command replays the supported capture packet end to end

backend/scripts/capture_placement_cases.py

@@ -24,9 +24,12 @@
 - The ``objects_3d`` stage boundary gets an explicit, successful object list for
   every case: an empty list where the source scene already carries the hazards,
   and one nonempty observed tap hazard case whose single hazard the coordinator's
-  one-time merge must place into the merged scene exactly once. The default
-  objects stubs never run on this path: a capture with no registered injection
-  fails loudly instead of fabricating a hazard.
+  one-time merge must place into the merged scene exactly once — delivered only
+  when the case's source does not already carry that hazard id, because the merge
+  contract replaces only detector/VLM twins and a same-id re-delivery of a
+  tap-sourced hazard would duplicate the id and be refused. The default objects
+  stubs never run on this path: a capture with no registered injection fails
+  loudly instead of fabricating a hazard.
 - Every other unavailable vision stage stays explicitly stubbed (prewarm off,
   external vision calls off). ``placement`` is ``stubs_exempt``: the real
   conversion and solver run behind the door when a case reaches them.
@@ -129,6 +132,20 @@
 # The one case whose objects_3d boundary carries a nonempty observed tap hazard.
 TAP_CASE_ID = "hazard-right-ground-gap"
 
+# Expectations pinned on the wire scene, evaluated against the journey's real
+# converted-path result: the wire's six-decimal rounding (declared by the packet as
+# 5e-7 ft) propagates through the real conversion and solver into derived feet
+# values, so those compare within the packet's declared 1e-4 ft derived-output
+# tolerance. Scene geometry itself is still compared exactly, field by field (see
+# _check_converted_geometry), and the case-suite evaluator stays exact by default
+# (tol_ft defaults to 0.0).
+_WIRE_DERIVED_TOLERANCE_FT = 1e-4
+
+# The supported case packet is the default replay target; the legacy W7 cases
+# manifest stays available explicitly via --manifest and keeps its own pinned
+# refusal state, reported as blockers rather than mistaken for acceptance.
+DEFAULT_MANIFEST = Path(__file__).resolve().parent.parent / "server/placement/tests/data/capture_cases_v1/manifest.json"
+
 # An observed tap hazard, hand-authored for this runner: a wall_rect vent on the
 # case's wall, away from the case's pinned decision geometry. It exists to prove
 # the coordinator's one-time merge, never to fabricate a conflicting hazard.
@@ -204,6 +221,22 @@ def _scene_bytes(document: dict[str, Any]) -> bytes:
     return json.dumps(document, sort_keys=True).encode()
 
 
+def _objects_3d_for(record: case_suite.CaseRecord) -> list[dict[str, Any]]:
+    """The objects_3d list this case's boundary delivers.
+
+    The tap hazard is delivered only where the upstream source does not already
+    carry it: the merge contract replaces only detector/VLM twins, so a same-id
+    re-delivery of a source-carried (tap-sourced) hazard is never replaced — it
+    duplicates the id and the merged scene is refused as inconsistent. A
+    source-carried hazard still reaches the one-time tap checks through the real
+    merge and real conversion.
+    """
+    if record.id != TAP_CASE_ID:
+        return []
+    source_object_ids = {obj.get("id") for obj in json.loads(record.source_path.read_text()).get("objects") or []}
+    return [] if TAP_HAZARD["id"] in source_object_ids else [copy.deepcopy(TAP_HAZARD)]
+
+
 def _injected_scene_run(ctx: StageContext) -> StageResult:
     """The scene stage boundary: this capture's stamped source scene, or a loud
     refusal. The default stub scene never runs here."""
@@ -368,7 +401,7 @@ def _walk_front_door(
         "record": record,
         "packet_id": packet_id,
         "packet_sha256": journey["packetSha256"],
-        "objects_3d": [copy.deepcopy(TAP_HAZARD)] if record.id == TAP_CASE_ID else [],
+        "objects_3d": _objects_3d_for(record),
     }
 
     files = []
@@ -757,6 +790,9 @@ def _collect_case(harness: _Harness, record: case_suite.CaseRecord, journey: dic
         case["convertedSha256"] = _sha256(converted_bytes) if converted_bytes else None
         case["resultSha256"] = _sha256(deps.storage.get_bytes(placement_artifact_key))
         case["solverPolicy"] = placement.get("policy")
+        # The solver's own decision, kept distinct from the device's served result
+        # kind (deviceOutcome) so the receipt shows both sides of the door.
+        case["solverDecision"] = placement.get("decision")
 
         # Both solver responses must bind their own input hash to their own exact
         # request bytes. Forging the same wrong hash in both replies leaves them in
@@ -805,7 +841,9 @@ def _collect_case(harness: _Harness, record: case_suite.CaseRecord, journey: dic
                         "direct POST /v1/placements on the captured bytes disagrees with the stage response"
                     )
         if record.wire_path is not None:
-            wire_mismatches = case_suite._evaluate_expectations(record.expectations, placement)
+            wire_mismatches = case_suite._evaluate_expectations(
+                record.expectations, placement, tol_ft=_WIRE_DERIVED_TOLERANCE_FT
+            )
             case["wireSemanticComparison"] = "matched" if not wire_mismatches else "mismatched"
             case["wireSemanticMismatches"] = wire_mismatches
             if wire_mismatches:
@@ -845,6 +883,7 @@ def _collect_case(harness: _Harness, record: case_suite.CaseRecord, journey: dic
         case["convertedSha256"] = None
         case["resultSha256"] = None
         case["solverPolicy"] = None
+        case["solverDecision"] = None
         case["directComparison"] = "not_applicable"
         case["wireSemanticComparison"] = "not_applicable"
     else:
@@ -1012,8 +1051,12 @@ def run_cases(
             "generatedAt": datetime.now(UTC).isoformat(),
             "codeSha": code_sha,
             "producers": {
-                "caseSuiteBranch": "obv/burst7-bs-placement-cases@ffd476ddb3a4c6f0b21683f8c353d4280f05dca5",
-                "sceneStageBranch": "obv/burst7-bs-capture-scene-stage@38a5735abf6c48982f8131b6dc6814afb90e28f3",
+                # The integrated producer SHAs on this branch (base + imports), so the
+                # receipt names exactly which producer code ran, not just branches.
+                "base": "obv/burst7-bs-capture-placement-runner@bdfbb3fc649df8187675b880ba8df83a803fd119",
+                "caseIntegrityFix": "115b6ce313e51cede71496f267b65b8cd78aa118 (obv/burst7-bs-placement-cases, PR #271)",
+                "replayAcceptance": "5f9e29576556f635757eea0605d5a23d29ce1cc5 (obv/burst8-bs-replay-acceptance, PR #275)",
+                "supportedCasePairs": "e7ad2abc15d94691305bdb558a752ee535cf374e (obv/burst8-bs-supported-case-pairs, PR #276)",
             },
             "devSettings": dict(TEMPLATE_ENV),
             "stubbedStages": {
@@ -1055,7 +1098,7 @@ def _redaction_check(receipt: dict[str, Any]) -> str | None:
 
 def main(argv: list[str] | None = None) -> int:
     parser = argparse.ArgumentParser(description="Replay the placement case suite through the real capture front door.")
-    parser.add_argument("--manifest", type=Path, default=case_suite.DEFAULT_MANIFEST, help="Path to the case manifest.")
+    parser.add_argument("--manifest", type=Path, default=DEFAULT_MANIFEST, help="Path to the case manifest.")
     parser.add_argument("--output", type=Path, required=True, help="Where to write the JSON receipt.")
     parser.add_argument("--workdir", type=Path, help="run directory (default: a fresh temp dir)")
     parser.add_argument("--cases", help="comma-separated case ids to run (default: every case)")

backend/server/common/backends/state_memory.py

@@ -50,6 +50,9 @@ def __init__(self) -> None:
         self._open_by_owner: dict[str, list[str]] = {}
         self._claim_lock = threading.Lock()
         self._event_lock = threading.Lock()
+        # Dedicated lock for stage-record creation (the append_event lesson: never
+        # share one lock across unrelated write paths — lock-ordering deadlock).
+        self._stage_lock = threading.Lock()
 
     def get_capture(self, capture_id: str) -> CaptureRecord | None:
         rec = self._captures.get(capture_id)
@@ -183,6 +186,19 @@ def list_runs(self, *, status: set[StageStatus], started_before: datetime | None
     def upsert_stage(self, rec: StageRecord) -> None:
         self._stages[(rec.run_id, rec.stage)] = rec.model_copy(deep=True)
 
+    def create_stage_if_absent(self, rec: StageRecord) -> StageRecord:
+        """Atomic insert-if-absent: the deciding read and the insert share one lock,
+        so a record another coordinator created and started inside the race window
+        is returned untouched instead of being reset to pending. Returns the record
+        now stored."""
+        key = (rec.run_id, rec.stage)
+        with self._stage_lock:
+            existing = self._stages.get(key)
+            if existing is not None:
+                return existing.model_copy(deep=True)
+            self._stages[key] = rec.model_copy(deep=True)
+            return rec.model_copy(deep=True)
+
     def get_stage(self, run_id: str, stage: StageName) -> StageRecord | None:
         rec = self._stages.get((run_id, stage))
         return None if rec is None else rec.model_copy(deep=True)

backend/server/common/backends/state_modal_dict.py

@@ -29,6 +29,13 @@
 _APPEND_GUARD = threading.Lock()
 _APPEND_LOCKS: dict[int, threading.Lock] = {}
 
+# Serializes create_stage_if_absent on the plain-dict fallback, where put_if_absent
+# is an in-process check-then-set; a real Modal Dict claims atomically on its own.
+# One lock, not one per (run, stage): a per-key registry kept a lock for every run
+# a long-lived worker ever touched. Claims only happen for records missing from
+# the coordinator's read, so the shared lock serializes only that recovery path.
+_STAGE_CLAIM_LOCK = threading.Lock()
+
 
 def _append_lock(mapping: object) -> threading.Lock:
     """One lock per backing map, so two states on the same dict share it."""
@@ -503,6 +510,19 @@ def list_runs(self, *, status: set[StageStatus], started_before: datetime | None
     def upsert_stage(self, rec: StageRecord) -> None:
         self._kv[f"stage:{rec.run_id}:{rec.stage}"] = rec.model_dump(mode="json")
 
+    def create_stage_if_absent(self, rec: StageRecord) -> StageRecord:
+        """Insert-if-absent across processes: Modal Dict ``skip_if_exists`` is the
+        atomic claim, so two coordinators resuming the same old run cannot both
+        create the record — the loser reads back the winner's record instead of
+        resetting a running stage to pending. The module lock covers the plain-dict
+        fallback, where the check-then-set is only in-process."""
+
+        key = f"stage:{rec.run_id}:{rec.stage}"
+        with _STAGE_CLAIM_LOCK:
+            self.put_if_absent(key, rec.model_dump(mode="json"))
+        stored = self._model(key, StageRecord)
+        return stored if stored is not None else rec.model_copy(deep=True)
+
     def get_stage(self, run_id: str, stage: StageName) -> StageRecord | None:
         return self._model(f"stage:{run_id}:{stage}", StageRecord)
 

backend/server/common/state.py

@@ -68,6 +68,10 @@ def list_runs(self, *, status: set[StageStatus], started_before: datetime | None
 
     def upsert_stage(self, rec: StageRecord) -> None: ...
 
+    def create_stage_if_absent(self, rec: StageRecord) -> StageRecord:
+        """Atomic insert-if-absent: recovery must never reset a record another
+        coordinator materialized and started first. Returns the record now stored."""
+
     def get_stage(self, run_id: str, stage: StageName) -> StageRecord | None: ...
 
     def stages(self, run_id: str) -> list[StageRecord]: ...

backend/server/common/tests/test_stage_record_claim.py

@@ -0,0 +1,94 @@
+"""Stage-record recovery claims are atomic: a record another coordinator
+materialized and started inside the race window is returned untouched, never
+reset to pending with its attempt and worker identity stripped."""
+
+from __future__ import annotations
+
+import threading
+
+from server.common.backends.state_memory import MemoryState
+from server.common.backends.state_modal_dict import ModalDictState
+from server.contracts.jobs import StageRecord
+
+
+def _rec(run_id: str, stage: str, worker: str | None = None) -> StageRecord:
+    return StageRecord(
+        run_id=run_id,
+        stage=stage,
+        status="pending",
+        attempt=1,
+        heartbeat="2026-10-16T00:00:00Z",
+        worker_call_id=worker,
+    )
+
+
+def test_memory_claim_returns_the_existing_record_untouched() -> None:
+    state = MemoryState()
+    first = state.create_stage_if_absent(_rec("run", "poses", worker="left"))
+    assert first.worker_call_id == "left"
+
+    second = state.create_stage_if_absent(_rec("run", "poses", worker="right"))
+
+    assert second.worker_call_id == "left", "the loser must read back the winner"
+    stored = state.get_stage("run", "poses")
+    assert stored.worker_call_id == "left" and stored.attempt == 1
+
+
+def test_memory_claim_inserts_when_absent() -> None:
+    state = MemoryState()
+    created = state.create_stage_if_absent(_rec("run", "poses"))
+    assert created.status == "pending" and created.attempt == 1
+    assert state.get_stage("run", "poses") == created
+
+
+def test_modal_claim_two_states_share_one_map() -> None:
+    mapping: dict[str, object] = {}
+    left, right = ModalDictState(mapping), ModalDictState(mapping)
+    winner = left.create_stage_if_absent(_rec("run", "poses", worker="left"))
+    loser = right.create_stage_if_absent(_rec("run", "poses", worker="right"))
+    assert winner.worker_call_id == "left"
+    assert loser.worker_call_id == "left", "the loser must read back the winner"
+    assert right.get_stage("run", "poses").worker_call_id == "left"
+
+
+def test_modal_claim_many_threads_one_winner() -> None:
+    state = ModalDictState()
+    results: list[str] = []
+    guard = threading.Lock()
+
+    def claim(worker: str) -> None:
+        rec = state.create_stage_if_absent(_rec("run", "poses", worker=worker))
+        with guard:
+            results.append(rec.worker_call_id)
+
+    threads = [threading.Thread(target=claim, args=(f"w{i}",)) for i in range(8)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    assert len(set(results)) == 1, f"every claimant must read back the same winner: {results}"
+    stored = state.get_stage("run", "poses")
+    assert stored.worker_call_id == results[0] and stored.attempt == 1
+
+
+def test_modal_claims_leave_no_per_run_state_in_the_module() -> None:
+    """A long-lived worker claims records for many runs. Module-level lock state
+    must stay bounded: a lock kept per (run, stage) key grows without limit."""
+    from server.common.backends import state_modal_dict
+
+    state = ModalDictState()
+    state.create_stage_if_absent(_rec("warm", "poses"))
+
+    def sizes() -> dict[str, int]:
+        return {
+            name: len(value)
+            for name, value in vars(state_modal_dict).items()
+            if isinstance(value, dict) and not name.startswith("__")
+        }
+
+    before = sizes()
+    for index in range(200):
+        state.create_stage_if_absent(_rec(f"run-{index}", "poses"))
+
+    assert sizes() == before

backend/server/orchestrator/api.py

@@ -168,6 +168,20 @@ def _recompose_after_correction(capture_id: str, run: RunRecord) -> None:
     services = current()
     if run.status != "done":
         return
+    present = {rec.stage for rec in services.state.stages(run.run_id)}  # type: ignore[attr-defined]
+    if not present.issuperset(STAGE_NAMES):
+        # A finished run without a record for every stage predates per-stage
+        # records. Reopening it would let the coordinator's missing-record
+        # recovery re-run its upstream stages, and the re-run reads stage would
+        # overwrite the correction just persisted. Keep the run closed instead.
+        log_event(
+            "confirmation.recompose_skipped",
+            level="warning",
+            msg="finished run has no record for every stage",
+            capture_id=capture_id,
+            run_id=run.run_id,
+        )
+        return
     now = services.clock.iso()
     for stage in ("criteria", "result"):
         rec = services.state.get_stage(run.run_id, stage)  # type: ignore[attr-defined]

backend/server/orchestrator/coordinator.py

@@ -239,6 +239,7 @@ def tick(self, run_id: str) -> StageStatus:
 
     def _spawn_ready(self, run: RunRecord, control: RunControl) -> list[str]:
         spawned: list[str] = []
+        self._materialize_missing_records(run)
         status = self._status_map(run.run_id)
         names = ready_stages(status)
         if "criteria" in names and not control.merge_done:
@@ -756,6 +757,34 @@ def _status_map(self, run_id: str) -> dict[StageName, StageStatus]:
         found = {rec.stage: rec.status for rec in self._rows(run_id)}
         return {name: found.get(name, "pending") for name in STAGE_NAMES}
 
+    def _materialize_missing_records(self, run: RunRecord) -> None:
+        """Recovery for runs written by older builds: a run can lack stage records
+        entirely (records materialize as the API schedules stages). The status map
+        treats a missing record as pending, but _spawn_one refuses to start a stage
+        with no record — so such a run hangs forever between ticks. Idempotently
+        create only the missing pending records: existing records keep their
+        attempts, failures and artifacts, and terminal runs never reach here
+        because tick returns before spawning.
+
+        Only stages absent from the rows this tick already read are claimed: with
+        ModalDictState each claim is a conditional remote write plus a read-back,
+        so claiming every stage cost two round-trips per stage on every tick.
+        """
+        present = {rec.stage for rec in self._rows(run.run_id)}
+        missing = [name for name in STAGE_NAMES if name not in present]
+        if not missing:
+            return
+        now = self.s.clock.iso()
+        for name in missing:
+            # Atomic insert-if-absent: two coordinators can resume the same old run
+            # (sweeper, _resume_if_stale, worker tick) without an exclusive claim. A
+            # plain check-then-upsert would reset a record the other coordinator
+            # created and a worker started inside the race window.
+            stored = self.s.state.create_stage_if_absent(  # type: ignore[attr-defined]
+                StageRecord(run_id=run.run_id, stage=name, status="pending", attempt=1, heartbeat=now)
+            )
+            self._remember_stage(stored)
+
     def _snapshot(self, run_id: str) -> str:
         parts = [f"{rec.stage}:{rec.status}:{rec.attempt}:{rec.failures}:{rec.respawns}" for rec in self._rows(run_id)]
         run = self._run(run_id)

backend/server/orchestrator/tests/test_greptile_followup.py

@@ -142,6 +142,10 @@ def test_confirmation_updates_the_reads_record(tmp_path: Path) -> None:
     assert stored.meter.status == "auto_fill"
     assert stored.confirm is not None
     assert stored.confirm.answer == "12345"
+    # This done run has no stage records, so the correction must not reopen it:
+    # missing-record recovery would re-run reads and overwrite the answer.
+    assert services.state.stages("run_reads") == []
+    assert services.state.get_run("run_reads").status == "done"
 
 
 def test_new_attempt_clears_entered_at(tmp_path: Path) -> None: