diff --git a/backend/server/api/app.py b/backend/server/api/app.py
index 47da13d1e..eedaf3aa9 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/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/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 ff886dc8a..6a04c1613 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
@@ -76,8 +78,56 @@ 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(finalize_module._slots[capture_id].inflight)
+        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
+
+    # 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.
     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..b6ad39ea4 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] = {}
@@ -335,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)
@@ -346,7 +352,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..8f5990498 100644
--- a/backend/server/orchestrator/api.py
+++ b/backend/server/orchestrator/api.py
@@ -148,6 +148,54 @@ 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"
+    # 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
+    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 +231,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 668c42a88..09e810f70 100644
--- a/backend/server/orchestrator/artifacts.py
+++ b/backend/server/orchestrator/artifacts.py
@@ -304,6 +304,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:
@@ -311,18 +322,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/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/<key>` instead."""
 
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
new file mode 100644
index 000000000..662083ca9
--- /dev/null
+++ b/backend/tests/test_runtime_join_journey.py
@@ -0,0 +1,633 @@
+"""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_after: list[int] = []
+    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
+        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:
+                time.sleep(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()
+    threads[2].join(timeout=15)
+    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, final_cursor + 1))
+    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")
+
+
+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"]
+    ]
+
+
+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"