diff --git a/backend/scripts/capture_placement_cases.py b/backend/scripts/capture_placement_cases.py
index 1b0bff964..d035879c2 100644
--- a/backend/scripts/capture_placement_cases.py
+++ b/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)")
diff --git a/backend/server/common/backends/state_memory.py b/backend/server/common/backends/state_memory.py
index e8d95e925..e4f5b8389 100644
--- a/backend/server/common/backends/state_memory.py
+++ b/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)
diff --git a/backend/server/common/backends/state_modal_dict.py b/backend/server/common/backends/state_modal_dict.py
index b6ad39ea4..a770ba747 100644
--- a/backend/server/common/backends/state_modal_dict.py
+++ b/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)
 
diff --git a/backend/server/common/state.py b/backend/server/common/state.py
index ce0d2a152..75fcc6632 100644
--- a/backend/server/common/state.py
+++ b/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]: ...
diff --git a/backend/server/common/tests/test_stage_record_claim.py b/backend/server/common/tests/test_stage_record_claim.py
new file mode 100644
index 000000000..e631f025d
--- /dev/null
+++ b/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
diff --git a/backend/server/orchestrator/api.py b/backend/server/orchestrator/api.py
index 8f5990498..0a382e53b 100644
--- a/backend/server/orchestrator/api.py
+++ b/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]
diff --git a/backend/server/orchestrator/coordinator.py b/backend/server/orchestrator/coordinator.py
index 6c96a6bef..4fe3a5f3b 100644
--- a/backend/server/orchestrator/coordinator.py
+++ b/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)
diff --git a/backend/server/orchestrator/tests/test_greptile_followup.py b/backend/server/orchestrator/tests/test_greptile_followup.py
index 039a3a4bd..d74b700dc 100644
--- a/backend/server/orchestrator/tests/test_greptile_followup.py
+++ b/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:
diff --git a/backend/server/orchestrator/tests/test_handoff.py b/backend/server/orchestrator/tests/test_handoff.py
index 00cd0acb1..3c4216d5f 100644
--- a/backend/server/orchestrator/tests/test_handoff.py
+++ b/backend/server/orchestrator/tests/test_handoff.py
@@ -4,6 +4,7 @@
 
 from datetime import timedelta
 
+from server.common.backends.state_memory import MemoryState
 from server.contracts.jobs import RunRecord, StageRecord, StageResult
 from server.orchestrator.coordinator import RunControl, RunCoordinator
 from server.orchestrator.graph import STAGE_NAMES
@@ -346,3 +347,104 @@ def spawn(run_id: str) -> None:
     control = get_json(services.state, f"run_control:{run.run_id}")
     assert control["coordinator_respawns"] == 1
     assert services.state.stages(run.run_id) == []
+
+
+def test_a_run_without_stage_records_still_progresses(tmp_path) -> None:
+    """An older unfinished run can lack stage records entirely: the status map
+    treats a missing record as pending, but spawning refused to start a stage
+    with no record, so the run hung forever between ticks. Recovery must
+    materialize the missing pending records idempotently — existing records
+    keep their attempt and status — and the run must reach a terminal state."""
+    services, _uri, digest = build_services(tmp_path, "memory")
+    deadline = (services.clock.now() + timedelta(hours=1)).isoformat()
+    run = _run(services, digest, deadline=deadline)
+    # The run exists but no stage record was ever written for it.
+    assert services.state.stages(run.run_id) == []
+    validate = StageRecord(
+        run_id=run.run_id, stage="validate", status="done", attempt=2, heartbeat=services.clock.iso()
+    )
+    services.state.upsert_stage(validate)
+
+    RunCoordinator(services).tick(run.run_id)
+
+    rows = {rec.stage: rec for rec in services.state.stages(run.run_id)}
+    assert {rec.stage for rec in rows.values()} == set(STAGE_NAMES)
+    assert rows["validate"].status == "done" and rows["validate"].attempt == 2, "existing record preserved"
+    status = "running"
+    for _ in range(40):
+        status = RunCoordinator(services).tick(run.run_id)
+        if status not in {"pending", "running"}:
+            break
+    assert status in {"done", "failed"}, f"run must terminate, not hang: {status}"
+
+
+def test_materialize_yields_to_a_record_that_raced_in(tmp_path, monkeypatch) -> None:
+    """Two coordinators can resume the same old run without an exclusive claim
+    (the sweeper, _resume_if_stale, and the worker tick all recover). If another
+    coordinator creates a record and a worker starts it inside the recovery race
+    window, the atomic claim must return that record untouched — never reset a
+    running stage to pending, attempt 1, no worker."""
+    services, _uri, digest = build_services(tmp_path, "memory")
+    deadline = (services.clock.now() + timedelta(hours=1)).isoformat()
+    run = _run(services, digest, deadline=deadline)
+    assert services.state.stages(run.run_id) == []
+
+    real_create = MemoryState.create_stage_if_absent
+
+    def racing_create(self, rec):
+        if rec.stage == "poses" and self.get_stage(rec.run_id, rec.stage) is None:
+            # The other coordinator wins the race: it creates the record and a
+            # worker starts it before our claim lands.
+            self.upsert_stage(
+                StageRecord(
+                    run_id=rec.run_id,
+                    stage="poses",
+                    status="running",
+                    attempt=2,
+                    heartbeat=rec.heartbeat,
+                    worker_call_id="other-coord",
+                )
+            )
+        return real_create(self, rec)
+
+    monkeypatch.setattr(MemoryState, "create_stage_if_absent", racing_create)
+
+    RunCoordinator(services).tick(run.run_id)
+
+    poses = services.state.get_stage(run.run_id, "poses")
+    assert poses is not None
+    assert poses.status == "running", poses
+    assert poses.attempt == 2 and poses.worker_call_id == "other-coord", poses
+
+
+def test_materialize_claims_only_the_missing_records(tmp_path, monkeypatch) -> None:
+    """With ModalDictState each claim is a conditional remote write plus a read-back,
+    so claiming all stages on every tick costs two round-trips per stage even when
+    every record exists. The tick already read the run's records; only the stages
+    missing from that read may be claimed."""
+    services, _uri, digest = build_services(tmp_path, "memory")
+    deadline = (services.clock.now() + timedelta(hours=1)).isoformat()
+    run = _run(services, digest, deadline=deadline)
+    now = services.clock.iso()
+    for name in STAGE_NAMES:
+        if name != "poses":
+            services.state.upsert_stage(
+                StageRecord(run_id=run.run_id, stage=name, status="pending", attempt=1, heartbeat=now)
+            )
+
+    claimed: list[str] = []
+    real_create = MemoryState.create_stage_if_absent
+
+    def counting_create(self, rec):
+        claimed.append(rec.stage)
+        return real_create(self, rec)
+
+    monkeypatch.setattr(MemoryState, "create_stage_if_absent", counting_create)
+
+    RunCoordinator(services).tick(run.run_id)
+    assert claimed == ["poses"], claimed
+    assert services.state.get_stage(run.run_id, "poses") is not None
+
+    claimed.clear()
+    RunCoordinator(services).tick(run.run_id)
+    assert claimed == [], "a tick whose run has every record must not claim any"
diff --git a/backend/server/placement/adapter.py b/backend/server/placement/adapter.py
index 6c2f09075..c6defdace 100644
--- a/backend/server/placement/adapter.py
+++ b/backend/server/placement/adapter.py
@@ -111,6 +111,18 @@ def __init__(self, message: str) -> None:
         self.code = "placement_objects_3d_unavailable"
 
 
+class PlacementObjectsFormatError(PlacementObjectsError):
+    """The run's objects_3d artifact does not carry an explicit objects list.
+
+    Not retryable: a malformed manifest is a refusal, not an empty hazard set —
+    an empty list is a valid "no hazards", a missing or null list is not.
+    """
+
+    def __init__(self, message: str) -> None:
+        super().__init__(message)
+        self.code = "placement_objects_3d_invalid"
+
+
 def _top_level_server_dir() -> Path:
     return _REPO_ROOT / "server"
 
@@ -236,6 +248,14 @@ def _objects_3d_evidence(ctx: StageContext) -> None:
     fmt = document.get("format") if isinstance(document, dict) else None
     if fmt != "hss-objects-3d/v0":
         raise PlacementObjectsError(f'{_OBJECTS_3D} must carry format "hss-objects-3d/v0", got {fmt!r}')
+    # An empty list is a valid "no hazards"; a missing or null list is a malformed
+    # manifest and must refuse, not decode as "no hazards".
+    objects = document.get("objects") if isinstance(document, dict) else None
+    if not isinstance(objects, list):
+        raise PlacementObjectsFormatError(
+            f'{_OBJECTS_3D} must carry an explicit "objects" list (an empty list is valid), got '
+            f"{'null' if objects is None else type(objects).__name__!r}"
+        )
 
 
 def _source_scene_bytes(ctx: StageContext) -> bytes:
diff --git a/backend/server/placement/case_suite.py b/backend/server/placement/case_suite.py
index 525fd768e..ebd7af6dc 100644
--- a/backend/server/placement/case_suite.py
+++ b/backend/server/placement/case_suite.py
@@ -269,7 +269,32 @@ def as_authored(path: Path) -> str:
     }
 
 
-def _evaluate_expectations(expectations: dict[str, Any], body: dict[str, Any]) -> list[str]:
+def _numbers_match(actual: Any, expected: Any, tol_ft: float) -> bool:
+    """Equality with a declared tolerance for derived feet values: floats and
+    float lists compare within ``tol_ft``, everything else exactly."""
+    if isinstance(expected, list):
+        return (
+            isinstance(actual, list)
+            and len(actual) == len(expected)
+            and all(_numbers_match(a, e, tol_ft) for a, e in zip(actual, expected, strict=True))
+        )
+    if (
+        isinstance(expected, (int, float))
+        and not isinstance(expected, bool)
+        and isinstance(actual, (int, float))
+        and not isinstance(actual, bool)
+    ):
+        return abs(float(actual) - float(expected)) <= tol_ft
+    return actual == expected
+
+
+def _evaluate_expectations(expectations: dict[str, Any], body: dict[str, Any], tol_ft: float = 0.0) -> list[str]:
+    """Evaluate declared expectations against a placement body.
+
+    ``tol_ft`` widens only numeric feet comparisons (``spot`` values,
+    ``missing_evidence`` entries, ``sweep_failing_runs`` start points) to the
+    declared derived-output tolerance; the default 0.0 keeps the suite exact.
+    """
     mismatches: list[str] = []
     expected_decision = expectations.get("decision")
     if expected_decision is not None and body.get("decision") != expected_decision:
@@ -281,7 +306,7 @@ def _evaluate_expectations(expectations: dict[str, Any], body: dict[str, Any]) -
             mismatches.append(f"spot is None but expected {expected_spot!r}")
         else:
             for key, value in expected_spot.items():
-                if spot.get(key) != value:
+                if not _numbers_match(spot.get(key), value, tol_ft):
                     mismatches.append(f"spot.{key} {spot.get(key)!r} != expected {value!r}")
     checks_by_id = {check["id"]: check for check in body.get("checks", [])}
     for check_id, expected in (expectations.get("checks") or {}).items():
@@ -299,7 +324,7 @@ def _evaluate_expectations(expectations: dict[str, Any], body: dict[str, Any]) -
             )
     for expected in expectations.get("missing_evidence") or []:
         found = any(
-            all(actual.get(key) == value for key, value in expected.items())
+            all(_numbers_match(actual.get(key), value, tol_ft) for key, value in expected.items())
             for actual in body.get("missing_evidence", [])
         )
         if not found:
@@ -307,7 +332,7 @@ def _evaluate_expectations(expectations: dict[str, Any], body: dict[str, Any]) -
     sweep = body.get("sweep", [])
     for expected in expectations.get("sweep_failing_runs") or []:
         found = any(
-            run.get("start_ft") == expected.get("start_ft")
+            _numbers_match(run.get("start_ft"), expected.get("start_ft"), tol_ft)
             and set(expected.get("failing") or []).issubset(run.get("failing") or [])
             for run in sweep
         )
diff --git a/backend/server/placement/scene_bridge.py b/backend/server/placement/scene_bridge.py
index 4056fab70..aada91a63 100644
--- a/backend/server/placement/scene_bridge.py
+++ b/backend/server/placement/scene_bridge.py
@@ -54,9 +54,11 @@
 - `overheads` and `facing` — observed entries on the wall, converted with the lengths.
   `object_id` cross-references have no wire field and are dropped.
 - `coverage` — the chain ends (`limit`/`unexplored` with their notes) and the explicit
-  bands from `coverage.bands` and every `coverage.sensitivity[].bands` entry: band name,
-  `span_ft` and `out_ft` convert with the lengths; `views` and `best_incidence_deg` have
-  no wire field and are dropped. A coverage without bands converts to ends only, which
+  primary bands from `coverage.bands`: band name, `span_ft` and `out_ft` convert with
+  the lengths; `views` and `best_incidence_deg` have no wire field and are dropped.
+  Sensitivity bands (`coverage.sensitivity[].bands`) are validated the same way but are
+  not observations — they are what-if variants, not the ground-truth coverage the
+  device saw. A coverage without bands converts to ends only, which
   the wire parser reads as nothing observed — an absent band stays unobserved.
 
 Refusals — every refusal raises `SceneBridgeError`, a `ValueError` carrying a `code` and
@@ -583,7 +585,7 @@ def end_out(end, side: str) -> dict[str, Any]:
 
     observed: list[dict[str, Any]] = []
 
-    def band_out(band, pointer: str) -> None:
+    def band_out(band, pointer: str) -> dict[str, Any]:
         if band.wall_id is not None and band.wall_id != wall_id:
             _invalid(f"the band names unknown wall {band.wall_id!r}", f"{pointer}/wall_id")
         record: dict[str, Any] = {
@@ -592,12 +594,16 @@ def band_out(band, pointer: str) -> None:
         }
         if band.out_m is not None:
             record["out_ft"] = _ft(band.out_m)
-        observed.append(record)
+        return record
 
     for index, band in enumerate(scene.coverage.bands or []):
-        band_out(band, f"/coverage/bands/{index}")
+        observed.append(band_out(band, f"/coverage/bands/{index}"))
     for sensitivity_index, sensitivity in enumerate(scene.coverage.sensitivity or []):
         for band_index, band in enumerate(sensitivity.bands):
+            # Sensitivity variants are validated but are not observations: the wire's
+            # observed list is the ground-truth coverage the device saw, and a what-if
+            # sensitivity band is not that. A sensitivity-only scene still converts to
+            # ends only — no observation is invented.
             band_out(band, f"/coverage/sensitivity/{sensitivity_index}/bands/{band_index}")
 
     if observed:
diff --git a/backend/server/placement/tests/test_scene_bridge.py b/backend/server/placement/tests/test_scene_bridge.py
index 22320b9d3..21904e1e2 100644
--- a/backend/server/placement/tests/test_scene_bridge.py
+++ b/backend/server/placement/tests/test_scene_bridge.py
@@ -122,6 +122,7 @@ def base_scene() -> dict[str, Any]:
         ],
         "coverage": {
             "method": "scene-bridge-test",
+            "bands": [{"band": "wall", "s_m": [-1.0, 1.0]}],
             "sensitivity": [
                 {
                     "surface": "wall",
@@ -220,7 +221,9 @@ def test_happy_path_fields_in_independently_computed_feet() -> None:
         "right": {"kind": "limit", "note": "fence line"},
     }
     assert converted["coverage"]["observed"] == [
-        {"band": "wall", "span_ft": [-2.0 / 0.3048, 4.0 / 0.3048], "out_ft": 2.13 / 0.3048}
+        # The primary band only: the sensitivity variant below is validated but is
+        # not an observation, and must not appear in the converted coverage.
+        {"band": "wall", "span_ft": [-1.0 / 0.3048, 1.0 / 0.3048]}
     ]
 
 
@@ -336,6 +339,43 @@ def test_meter_wall_rect_geometry_places_the_meter_at_its_span_centre_mid_height
     assert parsed.point_at(0.0)[0] == pytest.approx(0.0, abs=1e-6)
 
 
+def test_sensitivity_only_coverage_converts_to_ends_only() -> None:
+    """A sensitivity band is a what-if variant, not a device observation: a scene
+    whose coverage carries only sensitivity bands must convert to ends only, with
+    no invented observed list — that is a coverage gap, not a covered wall."""
+    document = base_scene()
+    document["coverage"] = {
+        "method": "scene-bridge-test",
+        "sensitivity": [
+            {
+                "surface": "wall",
+                "max_incidence_deg": 75,
+                "bands": [{"band": "wall", "s_m": [-2.0, 4.0], "out_m": 2.13}],
+            }
+        ],
+    }
+
+    converted = convert_scene(document)
+
+    assert "observed" not in converted["coverage"]
+    assert converted["coverage"]["ends"] == {
+        "left": {"kind": "unexplored"},
+        "right": {"kind": "limit", "note": "fence line"},
+    }
+
+
+def test_a_malformed_sensitivity_band_still_refuses() -> None:
+    """Not being an observation does not mean unvalidated: a reversed sensitivity
+    span still refuses, and the pointer names the sensitivity band."""
+    document = base_scene()
+    document["coverage"]["sensitivity"][0]["bands"][0]["s_m"] = [4.0, -2.0]
+
+    with pytest.raises(SceneBridgeError) as excinfo:
+        convert_scene(document)
+
+    assert excinfo.value.pointer == "/coverage/sensitivity/0/bands/0/s_m"
+
+
 def test_absent_coverage_band_stays_unobserved_and_unexplored_ends_are_kept() -> None:
     document = base_scene()
     document["coverage"] = {"method": "scene-bridge-test"}
diff --git a/backend/tests/test_capture_placement_cases.py b/backend/tests/test_capture_placement_cases.py
index ce56b6888..7f5c0ab63 100644
--- a/backend/tests/test_capture_placement_cases.py
+++ b/backend/tests/test_capture_placement_cases.py
@@ -26,7 +26,8 @@
 import capture_placement_cases as runner  # noqa: E402
 
 from server.eval.sceneio import validate  # noqa: E402
-from server.orchestrator.stubs._common import write_files  # noqa: E402
+from server.orchestrator.stubs._common import json_bytes, write_files  # noqa: E402
+from server.scene.merge import merge_object_lists  # noqa: E402
 
 MANIFEST = BACKEND / "server" / "placement" / "tests" / "data" / "cases" / "manifest.json"
 
@@ -146,6 +147,7 @@ def _supported_source_a() -> dict:
         ],
         "coverage": {
             "method": "capture-placement-cases-test",
+            "bands": [{"band": "wall", "s_m": [-2.0, 4.0], "out_m": 2.13}],
             "sensitivity": [
                 {
                     "surface": "wall",
@@ -252,6 +254,7 @@ def _supported_source_b() -> dict:
         ],
         "coverage": {
             "method": "capture-placement-cases-test",
+            "bands": [{"band": "wall", "s_m": [-4.0, 2.0], "out_m": 2.5}],
             "sensitivity": [
                 {
                     "surface": "wall",
@@ -528,6 +531,10 @@ def _write_twin_manifest(tmp_path: Path) -> Path:
             "kind": "positive",
             "source": source_ref,
             "wire": wire_ref,
+            # The hardened loader (case-integrity fix) rejects decision-only
+            # positives, so the twins pin source A's full geometric expectations
+            # (the tap hazard does not move them) — the cross-capture refusal must
+            # still come from geometry, not the loader or an expectation mismatch.
             "expectations": _expected_a(),
         },
         {
@@ -587,11 +594,14 @@ def test_every_case_walks_the_real_front_door_with_case_true_boundaries(receipt)
     # The one nonempty objects_3d case: the observed tap hazard, merged exactly once.
     assert _case(receipt, runner.TAP_CASE_ID)["tapHazardMergeCount"] == 1
 
-    # Both positives refused: no merged scenes, so the cross-capture comparison
-    # cannot run and must say so instead of passing vacuously.
+    # Both positives refused: the normalized-geometry hash comes only from a
+    # converted scene, so the cross-capture comparison cannot run and must say so
+    # instead of passing vacuously. The refused positives block the run instead.
     cross = receipt["crossCapture"]
     assert cross["status"] == "not_applicable"
+    assert cross["reason"]
     assert cross["distinctScenes"] is None and cross["distinctNormalizedGeometry"] is None
+    assert not any(blocker.startswith("crossCapture") for blocker in receipt["blockers"])
     assert cross["captureIds"] == [
         _case(receipt, "hazard-right-ground-gap")["captureId"],
         _case(receipt, "hazard-left-ground-gap")["captureId"],
@@ -701,7 +711,18 @@ def test_cli_runs_from_a_temporary_directory_and_exits_nonzero_while_blocked(tmp
     """The one-command entry point: receipt on disk, nonzero exit while the
     committed fixtures stay bridge-incompatible, no secrets in the file."""
     output = tmp_path / "receipt.json"
-    exit_code = runner.main(["--output", str(output), "--workdir", str(tmp_path / "run")])
+    # The committed legacy manifest, passed explicitly since the CLI's default
+    # moved to the supported packet (its refusal state stays pinned here).
+    exit_code = runner.main(
+        [
+            "--manifest",
+            str(runner.case_suite.DEFAULT_MANIFEST),
+            "--output",
+            str(output),
+            "--workdir",
+            str(tmp_path / "run"),
+        ]
+    )
     assert exit_code == 1
     receipt = json.loads(output.read_text())
     assert receipt["schemaVersion"] == runner.RECEIPT_SCHEMA_VERSION
@@ -791,6 +812,8 @@ def test_cli_on_the_committed_manifest_exits_nonzero_in_a_subprocess(tmp_path):
         [
             sys.executable,
             str(BACKEND / "scripts" / "capture_placement_cases.py"),
+            "--manifest",
+            str(runner.case_suite.DEFAULT_MANIFEST),
             "--output",
             str(output),
             "--workdir",
@@ -926,7 +949,10 @@ def test_one_geometry_under_two_identities_fails_cross_capture(tmp_path):
     assert cross["status"] == "evaluated"
     assert cross["distinctScenes"] is True, "identity stamps alone differ"
     assert cross["distinctNormalizedGeometry"] is False, "the geometry is byte-identical"
-    assert any("crossCapture" in blocker for blocker in receipt["blockers"])
+    # Both twins solve to their full expectations, so cross-capture is the only
+    # thing that blocks the run.
+    assert [(case["status"], case["mismatches"]) for case in receipt["cases"]] == [("solved", [])] * 2
+    assert [blocker.split(":")[0] for blocker in receipt["blockers"]] == ["crossCapture"]
     assert receipt["ok"] is False
 
 
@@ -986,3 +1012,52 @@ def test_run_cases_restores_the_process_environment(tmp_path):
     before = dict(os.environ)
     runner.run_cases(MANIFEST, tmp_path / "run", case_ids=["multiwall-source-unsupported"])
     assert dict(os.environ) == before
+
+
+@pytest.mark.parametrize(
+    "manifest",
+    [
+        pytest.param(MANIFEST, id="legacy-cases"),
+        pytest.param(runner.DEFAULT_MANIFEST, id="capture-cases-v1"),
+    ],
+)
+def test_the_tap_case_merges_exactly_one_vent_on_every_committed_manifest(manifest):
+    """The source fixture and the runner's injected objects_3d must agree on who
+    supplies the observed tap vent. The merge keeps a tap-sourced source object
+    and appends the incoming one, so a fixture that already carries the vent plus
+    an injected copy merges two vents and the supported positive stops solving.
+    Whichever side supplies it, the merged scene holds exactly the runner's
+    TAP_HAZARD once, so the converted-span check measures the right object."""
+    record = next(record for record in runner.case_suite.load_cases(manifest) if record.id == runner.TAP_CASE_ID)
+    source_objects = json.loads(record.source_path.read_text())["objects"]
+    merged = merge_object_lists({"objects": source_objects}, runner._objects_3d_for(record))
+    vents = [obj for obj in merged["objects"] if obj["id"] == runner.TAP_HAZARD["id"]]
+    assert vents == [runner.TAP_HAZARD]
+
+
+@pytest.mark.parametrize(
+    "objects_value", [pytest.param(None, id="null-objects"), pytest.param("omit", id="missing-objects")]
+)
+def test_a_format_correct_objects_3d_without_an_objects_list_refuses(tmp_path, monkeypatch, objects_value):
+    """Through the actual capture stage: a format-correct objects_3d artifact whose
+    objects list is missing or null is a malformed manifest and must refuse
+    placement with placement_objects_3d_invalid — never decode as an empty,
+    hazard-free scene — and no placement artifact may exist (the refusal is not a
+    fallback success)."""
+    body = {"format": "hss-objects-3d/v0"}
+    if objects_value != "omit":
+        body["objects"] = None
+
+    def malformed_objects_3d(ctx):
+        return write_files(ctx, {"objects/objects_3d.json": json_bytes(body)})
+
+    monkeypatch.setattr(runner, "_injected_objects_3d_run", malformed_objects_3d)
+    receipt = runner.run_cases(_write_supported_manifest(tmp_path), tmp_path / "run", case_ids=[runner.TAP_CASE_ID])
+    case = receipt["cases"][0]
+
+    assert case["status"] == "refused", case
+    assert case["refusal"]["code"] == "placement_objects_3d_invalid", case["refusal"]
+    assert case["hasPlacementArtifact"] is False, "a refused placement must not publish an artifact"
+    assert case["solverDecision"] is None
+    assert receipt["summary"]["solved"] == 0
+    assert receipt["ok"] is False, "an undeclared positive-case refusal is a blocker"
diff --git a/backend/tests/test_capture_replay_cli_acceptance.py b/backend/tests/test_capture_replay_cli_acceptance.py
new file mode 100644
index 000000000..1705c30aa
--- /dev/null
+++ b/backend/tests/test_capture_replay_cli_acceptance.py
@@ -0,0 +1,173 @@
+"""End-to-end acceptance of the replay CLI against the supported packet.
+
+Runs the actual ``scripts/capture_placement_cases.py`` CLI as a subprocess from a
+temporary cwd, with a fresh work directory and state, over the default supported
+``capture_cases_v1`` manifest: both positive captures solve via the real
+conversion, merge and placement path, every declared negative refuses with no
+placement artifact, and the receipt's cross-checks all hold. A deliberately
+corrupted fixture and an unexpected positive refusal both exit nonzero; the
+legacy W7 cases manifest stays available explicitly and keeps its own pinned
+refusal state reported as blockers, never mistaken for acceptance.
+"""
+
+import hashlib
+import json
+import shutil
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+BACKEND_ROOT = Path(__file__).resolve().parents[1]
+RUNNER = BACKEND_ROOT / "scripts" / "capture_placement_cases.py"
+PACKET_DIR = BACKEND_ROOT / "server/placement/tests/data/capture_cases_v1"
+LEGACY_MANIFEST = BACKEND_ROOT / "server/placement/tests/data/cases/manifest.json"
+
+POSITIVES = ("hazard-right-ground-gap", "hazard-left-ground-gap")
+NEGATIVES = (
+    "multiwall-source-unsupported",
+    "generated-object-evidence-unsupported",
+    "standoff-meter-anchor-unsupported",
+    "footprint-hazard-unsupported",
+    "generated-regions-unsupported",
+)
+
+
+def _run(root: Path, *extra: str) -> subprocess.CompletedProcess:
+    """The real CLI, as a subprocess, from a temporary cwd with fresh state."""
+    return subprocess.run(
+        [
+            sys.executable,
+            str(RUNNER),
+            "--output",
+            str(root / "receipt.json"),
+            "--workdir",
+            str(root / "work"),
+            *extra,
+        ],
+        cwd=root,
+        capture_output=True,
+        text=True,
+        timeout=600,
+    )
+
+
+def _receipt(root: Path) -> dict:
+    receipt = json.loads((root / "receipt.json").read_text())
+    assert receipt["ok"] is True, receipt["blockers"]
+    return receipt
+
+
+def test_the_supported_packet_solves_end_to_end_through_the_real_cli(tmp_path: Path) -> None:
+    result = _run(tmp_path)
+    assert result.returncode == 0, (result.stdout + result.stderr)[-3000:]
+    receipt = _receipt(tmp_path)
+
+    summary = receipt["summary"]
+    assert summary == {"cases": 7, "solved": 2, "refused": 5, "refusedDeclared": 5, "mismatched": 0, "errored": 0}
+    assert len(receipt["codeSha"]) == 40
+
+    by_id = {case["id"]: case for case in receipt["cases"]}
+    for positive in POSITIVES:
+        case = by_id[positive]
+        assert case["status"] == "solved"
+        assert case["mismatches"] == []
+        # Real conversion against the independently authored wire, field by field.
+        assert case["geometryComparison"] == "matched"
+        # Input-hash binding on both solver paths, direct endpoint parity.
+        assert case["stageInputShaMatchesCapturedBytes"] is True
+        assert case["directInputShaMatchesPostedBytes"] is True
+        assert case["directComparison"] == "matched"
+        assert case["wireSemanticComparison"] == "matched"
+        # Device and solver decisions, in distinct fields, both conservative.
+        assert case["solverDecision"] == "manual_review"
+        assert case["solverPolicy"]["id"] == "demo"
+        assert case["solverPolicy"]["auto_approve"] is True
+        assert case["deviceOutcome"]["captureStatus"] == "manual_review"
+        assert case["deviceOutcome"]["servedResultKind"] == "manual_review"
+        assert case["deviceOutcome"]["servedResultRunId"]
+        assert case["terminalEventCount"] == 1
+        assert case["hasPlacementArtifact"] is True
+        for key in ("sourceSha256", "convertedSha256", "resultSha256"):
+            assert len(case[key]) == 64
+    # One-time tap replacement: the tap hazard survives the real merge and the
+    # real conversion exactly once.
+    tap = by_id["hazard-right-ground-gap"]
+    assert tap["tapHazardMergeCount"] == 1
+    assert tap["tapHazardConvertedCount"] == 1
+
+    for negative in NEGATIVES:
+        case = by_id[negative]
+        assert case["status"] == "refused"
+        assert case["hasPlacementArtifact"] is False
+        refusal = case["refusal"]
+        assert refusal["declared"] is True
+        assert refusal["codeMatchesDeclaration"] is True
+        assert refusal["pointerMatchesDeclaration"] is True
+        assert case["deviceOutcome"]["servedResultKind"] == "manual_review"
+        assert case["terminalEventCount"] == 1
+
+    cross = receipt["crossCapture"]
+    assert cross["distinctScenes"] is True
+    assert cross["distinctNormalizedGeometry"] is True
+    assert receipt["finalizeReplay"]["ok"] is True
+
+    producers = receipt["producers"]
+    for producer in ("base", "caseIntegrityFix", "replayAcceptance", "supportedCasePairs"):
+        assert producers[producer]
+
+
+def test_the_replay_cli_exits_nonzero_after_deliberate_fixture_corruption(tmp_path: Path) -> None:
+    packet = tmp_path / "packet"
+    shutil.copytree(PACKET_DIR, packet)
+    source = packet / "hazard-left-ground-gap.source.json"
+    document = json.loads(source.read_text())
+    document["objects"][0]["plus_minus_m"] = 0.5
+    source.write_text(json.dumps(document, indent=2) + "\n")
+    # The manifest's pinned sha256 is deliberately left behind: the drift is the
+    # corruption, and the run must refuse to build on drifted evidence.
+
+    result = _run(tmp_path, "--manifest", str(packet / "manifest.json"))
+    assert result.returncode != 0
+    output = result.stdout + result.stderr
+    assert "drifted fixture" in output and "!= pinned" in output
+
+
+def test_the_replay_cli_exits_nonzero_when_a_positive_unexpectedly_refuses(tmp_path: Path) -> None:
+    packet = tmp_path / "packet"
+    shutil.copytree(PACKET_DIR, packet)
+    source = packet / "hazard-left-ground-gap.source.json"
+    document = json.loads(source.read_text())
+    # The W7 refusal shape: a stand-off meter anchor the bridge does not support.
+    document["objects"][0]["geometry"]["standoff_m"] = 0.1
+    source.write_text(json.dumps(document, indent=2) + "\n")
+    manifest_path = packet / "manifest.json"
+    manifest = json.loads(manifest_path.read_text())
+    pinned_sha = hashlib.sha256(source.read_bytes()).hexdigest()
+    for case in manifest["cases"]:
+        if case["id"] == "hazard-left-ground-gap":
+            case["source"]["sha256"] = pinned_sha
+    manifest_path.write_text(json.dumps(manifest, indent=2) + "\n")
+
+    result = _run(tmp_path, "--manifest", str(manifest_path))
+    assert result.returncode != 0
+    receipt = json.loads((tmp_path / "receipt.json").read_text())
+    assert receipt["ok"] is False
+    assert any("hazard-left-ground-gap" in blocker and "refused" in blocker for blocker in receipt["blockers"])
+
+
+def test_the_legacy_cases_manifest_stays_available_explicitly(tmp_path: Path) -> None:
+    result = _run(tmp_path, "--manifest", str(LEGACY_MANIFEST))
+    assert result.returncode != 0
+    receipt = json.loads((tmp_path / "receipt.json").read_text())
+    # The legacy manifest's pinned refusal state is reported as blockers — the
+    # original limitation, not an acceptance criterion.
+    assert receipt["ok"] is False
+    assert receipt["summary"]["solved"] == 0
+    assert receipt["blockers"]
+
+
+@pytest.mark.parametrize("path", [RUNNER, PACKET_DIR / "manifest.json", LEGACY_MANIFEST])
+def test_the_replay_targets_exist(path: Path) -> None:
+    assert path.exists()