diff --git a/backend/server/api/events.py b/backend/server/api/events.py
index 8926556c..7599a354 100644
--- a/backend/server/api/events.py
+++ b/backend/server/api/events.py
@@ -53,8 +53,16 @@ async def poll_events(deps: Deps, record: CaptureRecord, after: int, wait_s: flo
             waiter.clear()
         rows = deps.state.events_after(record.capture_id, after, limit=100)
         fresh = deps.state.get_capture(record.capture_id) or record
-        events = [Event(seq=row.seq, type=row.type, at=row.at, data=row.data) for row in rows]
-        nxt = events[-1].seq if events else after
+        from server.orchestrator.publication import public_final_event
+
+        events = []
+        for row in rows:
+            shown = public_final_event(deps.state, row)
+            if shown is not None:
+                event_type, data = shown
+                events.append(Event(seq=row.seq, type=event_type, at=row.at, data=data))
+        # Suppressed stale/duplicate events still advance the existing cursor.
+        nxt = rows[-1].seq if rows else after
         return EventsResponse(status=fresh.status, next=nxt, events=events)
     finally:
         deps.polls[record.capture_id] = deps.polls.get(record.capture_id, 1) - 1
diff --git a/backend/server/api/finalize.py b/backend/server/api/finalize.py
index 0616231b..25d0eaad 100644
--- a/backend/server/api/finalize.py
+++ b/backend/server/api/finalize.py
@@ -142,12 +142,15 @@ def _request_again(capture_id: str) -> None:
             slot.again = True
 
 
-def _clear_slot(capture_id: str) -> None:
+def _clear_slot(capture_id: str) -> bool:
     with _slot_guard:
         slot = _slots.get(capture_id)
         if slot is not None:
+            again = slot.again
             slot.inflight = False
             slot.again = False
+            return again
+    return False
 
 
 def accept_packet(
@@ -497,7 +500,7 @@ def _spawn(deps: Deps, capture_id: str) -> None:
 
 def _guard(deps: Deps, capture_id: str, future: Future[None]) -> None:
     try:
-        _finish_loop(deps, capture_id)
+        retry = _finish_loop(deps, capture_id)
     except Exception as exc:
         # 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
@@ -535,13 +538,17 @@ def _guard(deps: Deps, capture_id: str, future: Future[None]) -> None:
                 level="error",
             )
         finally:
-            _clear_slot(capture_id)
+            retry = _clear_slot(capture_id)
             future.set_exception(exc)
+        if retry:
+            _kick(deps, capture_id)
         return
     future.set_result(None)
+    if retry:
+        _kick(deps, capture_id)
 
 
-def _finish_loop(deps: Deps, capture_id: str) -> None:
+def _finish_loop(deps: Deps, capture_id: str) -> bool:
     while True:
         outcome = _work_once(deps, capture_id)
         with _slot_guard:
@@ -551,7 +558,8 @@ def _finish_loop(deps: Deps, capture_id: str) -> None:
             if again and outcome == "incomplete":
                 continue
             slot.inflight = False
-            return
+            slot.again = False
+            return again
 
 
 def _still_missing(deps: Deps, record: CaptureRecord, extra: dict) -> list[str]:
diff --git a/backend/server/api/hooks.py b/backend/server/api/hooks.py
index e162abc7..980fc844 100644
--- a/backend/server/api/hooks.py
+++ b/backend/server/api/hooks.py
@@ -89,9 +89,11 @@ def resolve_hooks(settings: Settings, state=None, storage=None):
     return OrchestratorHooks(settings, state, storage)
 
 
-def safe_hook(name: str, fn, *args, **kwargs):
+def safe_hook(name: str, fn, *args, strict: bool = False, **kwargs):
     try:
         return fn(*args, **kwargs)
     except Exception as exc:
         log_event("hook.error", hook=name, error_code=type(exc).__name__, level="error")
+        if strict:
+            raise
         return None
diff --git a/backend/server/api/routes/confirmations.py b/backend/server/api/routes/confirmations.py
index fa62e3de..92b88b86 100644
--- a/backend/server/api/routes/confirmations.py
+++ b/backend/server/api/routes/confirmations.py
@@ -81,6 +81,21 @@ async def create_confirmation(capture_id: str, body: ConfirmationRequest, reques
             slot = f"confirm:{capture_id}:{body.fieldId}"
             existing = deps.state.get_kv(slot)
             if existing and existing.get("fingerprint") == digest:
+                if existing.get("hook_applied", True):
+                    return JSONResponse(_public(existing), status_code=200)
+                # The confirmation row is durable, but its correction hook was
+                # interrupted. Replay the hook before acknowledging the retry.
+                safe_hook(
+                    "on_confirmation",
+                    deps.hooks.on_confirmation,
+                    capture_id,
+                    body,
+                    existing["confirmationId"],
+                    by=existing["by"],
+                    strict=True,
+                )
+                existing["hook_applied"] = True
+                deps.state.put_kv(slot, existing)
                 return JSONResponse(_public(existing), status_code=200)
             if existing and existing.get("fingerprint") != digest and by == "device":
                 raise fail(409, "confirmation_conflict", "/", "this field already has a different answer")
@@ -103,13 +118,24 @@ async def create_confirmation(capture_id: str, body: ConfirmationRequest, reques
                 "value": body.value,
                 "receivedAt": isoformat(deps.now()),
                 "seq": stored_event.seq,
+                "hook_applied": False,
             }
             deps.state.put_kv(slot, row)
             listing = deps.state.get_kv(f"confirmix:{capture_id}") or {"fields": []}
             if body.fieldId not in listing["fields"]:
                 listing["fields"].append(body.fieldId)
             deps.state.put_kv(f"confirmix:{capture_id}", listing)
-            safe_hook("on_confirmation", deps.hooks.on_confirmation, capture_id, body, confirmation_id, by=by)
+            safe_hook(
+                "on_confirmation",
+                deps.hooks.on_confirmation,
+                capture_id,
+                body,
+                confirmation_id,
+                by=by,
+                strict=True,
+            )
+            row["hook_applied"] = True
+            deps.state.put_kv(slot, row)
             log_event(
                 "confirmation.received",
                 field_id=body.fieldId,
diff --git a/backend/server/api/routes/result.py b/backend/server/api/routes/result.py
index a5154bd8..b328cd9b 100644
--- a/backend/server/api/routes/result.py
+++ b/backend/server/api/routes/result.py
@@ -12,6 +12,7 @@
 from server.api.errors import fail
 from server.common.logging import log_event
 from server.contracts.result import ResultResponse
+from server.orchestrator.corrections import latest_publication
 from server.orchestrator.stateio import get_json
 
 
@@ -40,6 +41,9 @@ def _artifacts(deps, run) -> dict:
 
     if run is None:
         return {}
+    published = latest_publication(deps.state, run.run_id)
+    if published is not None:
+        return dict(published["artifacts"])
     fresh = deps.state.get_run(run.run_id) or run
     artifacts = dict(fresh.artifacts or {})
     extra = get_json(deps.state, f"stage_artifacts:{fresh.run_id}:result") or {}
@@ -79,7 +83,8 @@ def get_result(capture_id: str, request: Request):
         except Exception as exc:
             log_event("storage.error", op="get", capture_id=capture_id, level="error")
             raise fail(503, "storage_unavailable", "/", "could not read the result", **{"Retry-After": "5"}) from exc
-        payload["status"] = record.status
+        if latest_publication(deps.state, run.run_id) is None:
+            payload["status"] = record.status
         payload["runId"] = run.run_id
         profile = deps.settings.RULES_PROFILE_DEFAULT
         verdict_key = artifacts.get(f"verdicts/verdict.{profile}.json")
diff --git a/backend/server/api/tests/test_finalize_background.py b/backend/server/api/tests/test_finalize_background.py
index 28755d28..727b3ac2 100644
--- a/backend/server/api/tests/test_finalize_background.py
+++ b/backend/server/api/tests/test_finalize_background.py
@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from concurrent.futures import Future
+
 import pytest
 
 from server.api.deps import drain_finalize, extras
@@ -205,3 +207,34 @@ def failing_publication(*args, **kwargs):
     assert deps.state.get_capture(capture_id).status == "manual_review"
     run = deps.state.get_run(body["runId"])
     assert run is not None and run.status == "done"
+
+
+def test_failure_publication_error_preserves_a_queued_retry(auto_client, monkeypatch):
+    """A retry that arrived during a failed worker must be kicked after cleanup."""
+
+    import server.api.finalize as finalize_module
+
+    deps = auto_client.app_state
+    capture_id = "cap-queued-retry"
+    slot = finalize_module._slot_locked(capture_id)
+    slot.inflight = True
+    slot.again = True
+    future = Future()
+    kicked: list[str] = []
+
+    def failed_work(*_args, **_kwargs):
+        raise RuntimeError("work failed")
+
+    def failed_publication(*_args, **_kwargs):
+        raise RuntimeError("publication failed")
+
+    monkeypatch.setattr(finalize_module, "_finish_loop", failed_work)
+    monkeypatch.setattr(finalize_module, "_on_background_failure", failed_publication)
+    monkeypatch.setattr(finalize_module, "_kick", lambda _deps, cid: kicked.append(cid))
+
+    finalize_module._guard(deps, capture_id, future)
+
+    assert kicked == [capture_id]
+    assert slot.inflight is False
+    with pytest.raises(RuntimeError, match="work failed"):
+        future.result()
diff --git a/backend/server/api/tests/test_hooks_orchestrator.py b/backend/server/api/tests/test_hooks_orchestrator.py
index 0cdd2dd5..ab9ddb12 100644
--- a/backend/server/api/tests/test_hooks_orchestrator.py
+++ b/backend/server/api/tests/test_hooks_orchestrator.py
@@ -27,6 +27,7 @@
 import pytest
 
 from server.api.tests.conftest import ApiClient
+from server.common.lease import LeaseBusy
 
 FIXTURE = Path(__file__).resolve().parents[3] / "data" / "fixtures" / "packets" / "mini-arkit"
 PACKET_RAW = (FIXTURE / "packet.json").read_bytes()
@@ -203,6 +204,112 @@ def test_orchestrator_stage_failure_is_recorded_not_a_500(
     assert payload["outcome"]["kind"] == "manual_review"
 
 
+def test_confirmation_busy_is_retryable_and_replays_hook(auto_client: ApiClient) -> None:
+    client = auto_client
+    created = client.post(
+        "/v1/captures",
+        json={
+            "packetId": PACKET_ID,
+            "formatVersion": "0.4",
+            "tier": PACKET["session"]["tier"],
+            "device": {"model": "iPhone", "systemVersion": "18", "appVersion": "1"},
+        },
+        headers={"Idempotency-Key": PACKET_ID},
+    )
+    assert created.status_code == 201, created.text
+    capture_id = created.json()["captureId"]
+    _upload_and_commit(client, capture_id)
+    _finalize(client, capture_id)
+    deps = client.app_state
+    original = deps.hooks.on_confirmation
+    calls = {"count": 0}
+
+    def busy_once(*args, **kwargs):
+        calls["count"] += 1
+        if calls["count"] == 1:
+            raise LeaseBusy("correction lease is busy")
+        return original(*args, **kwargs)
+
+    deps.hooks.on_confirmation = busy_once
+    answer = {
+        "fieldId": "meter_number",
+        "kind": "corrected",
+        "value": "12345",
+        "imageId": PACKET["keyframes"][0]["id"],
+        "clientAt": "2026-10-16T00:00:00.000Z",
+    }
+    busy = client.post(f"/v1/captures/{capture_id}/confirmations", json=answer)
+    assert busy.status_code == 503, busy.text
+    slot = deps.state.get_kv(f"confirm:{capture_id}:meter_number")
+    assert slot["hook_applied"] is False
+    replay = client.post(f"/v1/captures/{capture_id}/confirmations", json=answer)
+    assert replay.status_code == 200, replay.text
+    assert deps.state.get_kv(f"confirm:{capture_id}:meter_number")["hook_applied"] is True
+    assert calls["count"] == 2
+
+
+@pytest.mark.parametrize("boundary", ["before_journal", "after_journal", "before_ack"])
+def test_confirmation_storage_fault_never_acknowledges_or_duplicates(
+    auto_client: ApiClient, monkeypatch: pytest.MonkeyPatch, boundary: str
+) -> None:
+    from server.orchestrator.corrections import correction_snapshot, latest_publication
+
+    client = auto_client
+    created = client.post(
+        "/v1/captures",
+        json={
+            "packetId": PACKET_ID,
+            "formatVersion": "0.4",
+            "tier": PACKET["session"]["tier"],
+            "device": {"model": "iPhone", "systemVersion": "18", "appVersion": "1"},
+        },
+        headers={"Idempotency-Key": PACKET_ID},
+    )
+    assert created.status_code == 201
+    capture_id = created.json()["captureId"]
+    _upload_and_commit(client, capture_id)
+    run_id = _finalize(client, capture_id)["runId"]
+    state = client.app_state.state
+    insert = state.put_if_absent
+    store = state.put_kv
+    hit = []
+
+    def journal(key, row):
+        if key.startswith("capture_correction:") and boundary != "before_ack":
+            hit.append(key)
+            if boundary == "after_journal":
+                assert insert(key, row)
+            raise LeaseBusy("injected storage interruption")
+        return insert(key, row)
+
+    def acknowledgment(key, row):
+        if key.startswith("confirm:") and row.get("hook_applied") and boundary == "before_ack":
+            hit.append(key)
+            raise LeaseBusy("injected acknowledgment interruption")
+        return store(key, row)
+
+    answer = {
+        "fieldId": "meter_number",
+        "kind": "corrected",
+        "value": "12345",
+        "imageId": PACKET["keyframes"][0]["id"],
+        "clientAt": "2026-10-16T00:00:00.000Z",
+    }
+    with monkeypatch.context() as patch:
+        patch.setattr(state, "put_if_absent", journal)
+        patch.setattr(state, "put_kv", acknowledgment)
+        interrupted = client.post(f"/v1/captures/{capture_id}/confirmations", json=answer)
+    assert hit, "the fault boundary was never reached"
+    assert interrupted.status_code == 503, interrupted.text
+    retry = client.post(f"/v1/captures/{capture_id}/confirmations", json=answer)
+    assert retry.status_code in {200, 201}, retry.text
+    run = state.get_run(run_id)
+    rows = correction_snapshot(state, run)
+    assert [row["confirmationId"] for row in rows] == [retry.json()["confirmationId"]]
+    assert latest_publication(state, run_id)["correctionIds"] == [retry.json()["confirmationId"]]
+    assert client.get(f"/v1/captures/{capture_id}/result").json()["meter"]["value"] == "12345"
+
+
 def test_confirmation_replay_recomposes_once(auto_client: ApiClient) -> None:
     """A corrected answer POSTed twice recomposes the result exactly once.
 
diff --git a/backend/server/common/backends/state_modal_dict.py b/backend/server/common/backends/state_modal_dict.py
index a770ba74..bfa0e916 100644
--- a/backend/server/common/backends/state_modal_dict.py
+++ b/backend/server/common/backends/state_modal_dict.py
@@ -518,8 +518,7 @@ def create_stage_if_absent(self, rec: StageRecord) -> StageRecord:
         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"))
+        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)
 
@@ -581,10 +580,13 @@ def put_if_absent(self, key: str, value: dict) -> bool:
                 pass
             else:
                 return bool(inserted)
-        if self._kv.get(key) is not None:
-            return False
-        self._kv[key] = payload
-        return True
+        # All fallback insertions need the shared process lock, not only stage
+        # creation: immutable correction/version sequences also claim slots.
+        with _STAGE_CLAIM_LOCK:
+            if self._kv.get(key) is not None:
+                return False
+            self._kv[key] = payload
+            return True
 
     def put_blob(self, key: str, value: bytes) -> None:
         self._kv[f"blob:{key}"] = value
diff --git a/backend/server/common/lease.py b/backend/server/common/lease.py
index 4de6cb93..f5aba6d7 100644
--- a/backend/server/common/lease.py
+++ b/backend/server/common/lease.py
@@ -1,43 +1,109 @@
-"""A short Dict lease so two containers can update one key without dropping a write."""
+"""A renewable Dict lease with advisory ownership checks.
+
+The checks and writes are separate RPCs, not atomic fencing. An expired owner
+can delete its successor's token. Correction safety therefore comes from
+immutable records and result versions, not exclusive ownership of this lease.
+"""
 
 from __future__ import annotations
 
+import threading
 import time
 import uuid
 from collections.abc import Callable
 
 
+class LeaseLost(RuntimeError):
+    """The caller tried to write after another owner took the lease."""
+
+
+class LeaseGuard:
+    """Detects ownership already lost, but cannot fence a subsequent write."""
+
+    def __init__(self, state: object, key: str, token: dict, heartbeat_key: str) -> None:
+        self._state = state
+        self._key = key
+        self._token = token
+        self._heartbeat_key = heartbeat_key
+
+    def assert_held(self) -> None:
+        current = self._state.get_kv(self._key)  # type: ignore[attr-defined]
+        if not isinstance(current, dict) or current.get("id") != self._token["id"]:
+            raise LeaseLost(f"lease {self._key!r} is no longer held")
+
+    def renew(self) -> None:
+        self.assert_held()
+        # Heartbeats are owner-specific. They never overwrite the lease record,
+        # so a late heartbeat cannot extend or replace a successor's lease.
+        self._state.put_kv(self._heartbeat_key, {"at": time.time()})  # type: ignore[attr-defined]
+
+
 def with_lease(
     state: object,
     name: str,
-    fn: Callable[[], None],
+    fn: Callable[..., None],
     *,
     wait_s: float = 3.0,
     lease_s: float = 10.0,
     strict: bool = False,
+    pass_guard: bool = False,
 ) -> None:
-    """Run ``fn`` while holding ``lease:{name}``. An abandoned lease can be taken after ``lease_s``.
+    """Run ``fn`` while holding a renewable lease.
 
-    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.
+    ``pass_guard`` gives callers a :class:`LeaseGuard` to detect lost ownership.
+    Owner-specific heartbeat keys prevent a late heartbeat from replacing the
+    successor's token, but checking ownership does not fence a later write.
     """
 
     key = f"lease:{name}"
     token = {"id": uuid.uuid4().hex, "at": time.time()}
+    heartbeat_key = f"lease_heartbeat:{name}:{token['id']}"
     deadline = time.monotonic() + wait_s
     while True:
         if state.put_if_absent(key, token):  # type: ignore[attr-defined]
+            guard = LeaseGuard(state, key, token, heartbeat_key)
+            stop = threading.Event()
+            lost = threading.Event()
+
+            def heartbeat(*, _stop=stop, _guard=guard, _lost=lost) -> None:
+                interval = max(0.01, min(lease_s / 3.0, 1.0))
+                while not _stop.wait(interval):
+                    try:
+                        _guard.renew()
+                    except LeaseLost:
+                        _lost.set()
+                        return
+
+            # The initial owner timestamp is valid until the first heartbeat.
+            state.put_kv(heartbeat_key, {"at": token["at"]})  # type: ignore[attr-defined]
+            thread = threading.Thread(target=heartbeat, daemon=True)
+            thread.start()
             try:
-                return fn()
+                if pass_guard:
+                    fn(guard)
+                else:
+                    fn()
+                guard.assert_held()
+                if lost.is_set():
+                    raise LeaseLost(f"lease {name!r} was lost while the callback ran")
             finally:
+                stop.set()
+                thread.join(timeout=max(0.1, min(lease_s / 3.0, 1.0)))
                 current = state.get_kv(key)  # type: ignore[attr-defined]
                 if isinstance(current, dict) and current.get("id") == token["id"]:
                     state.delete_kv(key)  # type: ignore[attr-defined]
+                state.delete_kv(heartbeat_key)  # type: ignore[attr-defined]
+            return
         current = state.get_kv(key)  # type: ignore[attr-defined]
-        if isinstance(current, dict) and time.time() - float(current.get("at") or 0) > lease_s:
-            state.delete_kv(key)  # type: ignore[attr-defined]
-            continue
+        if isinstance(current, dict):
+            owner_id = current.get("id")
+            heartbeat = state.get_kv(f"lease_heartbeat:{name}:{owner_id}")  # type: ignore[attr-defined]
+            at = heartbeat.get("at") if isinstance(heartbeat, dict) else current.get("at")
+            if time.time() - float(at or 0) > lease_s:
+                # This delete is not conditional on the token read above.
+                # Callers must tolerate overlapping owners, including on release.
+                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")
@@ -46,8 +112,4 @@ def with_lease(
 
 
 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.
-    """
+    """A strict lease could not be acquired within its wait budget."""
diff --git a/backend/server/common/tests/test_lease.py b/backend/server/common/tests/test_lease.py
new file mode 100644
index 00000000..f0e5bf13
--- /dev/null
+++ b/backend/server/common/tests/test_lease.py
@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+import threading
+
+import pytest
+
+from server.common.backends.state_memory import MemoryState
+from server.common.lease import LeaseBusy, LeaseLost, with_lease
+
+
+def test_lease_renews_while_callback_is_longer_than_expiry() -> None:
+    state = MemoryState()
+    entered = threading.Event()
+    release = threading.Event()
+    second_entered = threading.Event()
+    errors: list[BaseException] = []
+
+    def first() -> None:
+        entered.set()
+        assert release.wait(2)
+
+    def second() -> None:
+        second_entered.set()
+
+    def run_first() -> None:
+        try:
+            with_lease(state, "completion", first, wait_s=0.2, lease_s=0.05, strict=True)
+        except BaseException as exc:
+            errors.append(exc)
+
+    worker = threading.Thread(target=run_first)
+    worker.start()
+    assert entered.wait(1)
+    with pytest.raises(LeaseBusy):
+        with_lease(state, "completion", second, wait_s=0.05, lease_s=0.05, strict=True)
+    assert not second_entered.is_set()
+    release.set()
+    worker.join(2)
+    assert errors == []
+
+
+def test_expired_holder_cannot_publish_over_second_owner() -> None:
+    state = MemoryState()
+    entered = threading.Event()
+    let_first_finish = threading.Event()
+    second_entered = threading.Event()
+    first_errors: list[BaseException] = []
+
+    def first(guard) -> None:
+        entered.set()
+        assert let_first_finish.wait(2)
+        guard.assert_held()
+
+    def run_first() -> None:
+        try:
+            with_lease(state, "completion", first, wait_s=0.2, lease_s=60, strict=True, pass_guard=True)
+        except BaseException as exc:
+            first_errors.append(exc)
+
+    worker = threading.Thread(target=run_first)
+    worker.start()
+    assert entered.wait(1)
+    state.delete_kv("lease:completion")
+    with_lease(
+        state,
+        "completion",
+        lambda: second_entered.set(),
+        wait_s=0.2,
+        lease_s=60,
+        strict=True,
+    )
+    assert second_entered.is_set()
+    let_first_finish.set()
+    worker.join(2)
+    assert len(first_errors) == 1
+    assert isinstance(first_errors[0], LeaseLost)
diff --git a/backend/server/orchestrator/api.py b/backend/server/orchestrator/api.py
index 7ba0d0eb..d8b7bb84 100644
--- a/backend/server/orchestrator/api.py
+++ b/backend/server/orchestrator/api.py
@@ -7,6 +7,7 @@
 import time
 from pathlib import Path
 
+from server.common.lease import LeaseBusy, LeaseGuard, LeaseLost
 from server.common.logging import log_event
 from server.contracts.api import ConfirmationRequest, StageData
 from server.contracts.jobs import (
@@ -16,7 +17,18 @@
 )
 from server.contracts.packet import Tier
 from server.orchestrator.clock import parse_iso
-from server.orchestrator.coordinator import RunCoordinator, pending_correction_key, reset_criteria_and_result
+from server.orchestrator.coordinator import (
+    RunCoordinator,
+    reset_criteria_and_result,
+    with_correction_lease,
+)
+from server.orchestrator.corrections import (
+    claim_available_corrections,
+    newest_correction_run,
+    pending_corrections,
+    record_correction,
+    seed_run_corrections,
+)
 from server.orchestrator.early.capture_check import final_check, flush_pending
 from server.orchestrator.early.dispatch import dispatch
 from server.orchestrator.early.index import early_keys
@@ -88,6 +100,7 @@ def start_run(
     # The run record lands with the claim, before any slow read, so a dead
     # caller cannot leave a claim that blocks the capture forever.
     services.state.put_run(run)  # type: ignore[attr-defined]
+    seed_run_corrections(services.state, run)
     put_json(
         services.state,
         f"run_control:{run_id}",
@@ -149,59 +162,54 @@ def dispatch_early(capture_id: str, files: list[CommittedFile]) -> None:
 
 
 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 finished run without a record for every stage predates per-stage records.
-    Its upstream stages really finished — the run is done — so the rebuild
-    preserves that work instead of skipping it: every stage history never wrote
-    gets a terminal done record (idempotent, the same recovery shape the
-    coordinator's missing-record materialization uses), and only the criteria and
-    result records reset. No succeeded stage reopens, so the re-run reads stage
-    that would overwrite the just-persisted correction never executes. If the
-    stored upstream artifacts cannot support the rebuild — no scene bytes for the
-    coordinator-owned merge, say — the coordinator closes the run with its normal
-    artifact_invalid failure instead of leaving the stale result served as
-    success.
-
-    A live run is left alone — its result stage has not consumed the corrected
-    reads yet, so it recomposes by simply running.
+    """Reopen the finished run without deleting correction evidence.
+
+    Legacy runs materialize upstream stage rows so only criteria/result rerun.
+    The lease avoids duplicate work; immutable publication establishes safety.
     """
 
     services = current()
     if run.status != "done":
         return
-    present = {rec.stage for rec in services.state.stages(run.run_id)}  # type: ignore[attr-defined]
-    missing = [name for name in STAGE_NAMES if name not in present]
-    if missing:
-        now = services.clock.iso()
+    prepared = [False]
+
+    def prepare(guard: LeaseGuard) -> None:
+        nonlocal run
+        stored = services.state.get_run(run.run_id)  # type: ignore[attr-defined]
+        if stored is None or stored.status != "done" or not pending_corrections(services.state, stored):
+            return
+        run = stored
+        present = {rec.stage for rec in services.state.stages(run.run_id)}  # type: ignore[attr-defined]
+        missing = [name for name in STAGE_NAMES if name not in present]
         preserved = 0
-        for name in missing:
-            if name in ("criteria", "result"):
-                continue
-            services.state.create_stage_if_absent(  # type: ignore[attr-defined]
-                StageRecord(run_id=run.run_id, stage=name, status="done", attempt=1, heartbeat=now, finished_at=now)
+        if missing:
+            now = services.clock.iso()
+            for name in missing:
+                if name in ("criteria", "result"):
+                    continue
+                services.state.create_stage_if_absent(  # type: ignore[attr-defined]
+                    StageRecord(
+                        run_id=run.run_id,
+                        stage=name,
+                        status="done",
+                        attempt=1,
+                        heartbeat=now,
+                        finished_at=now,
+                    )
+                )
+                preserved += 1
+            log_event(
+                "confirmation.recompose_legacy",
+                capture_id=capture_id,
+                run_id=run.run_id,
+                preserved_stages=preserved,
             )
-            preserved += 1
-        log_event(
-            "confirmation.recompose_legacy",
-            capture_id=capture_id,
-            run_id=run.run_id,
-            preserved_stages=preserved,
-        )
-    # A stale live-window marker predates this recomposition: the reset below
-    # already re-runs both stages with every accepted answer.
-    services.state.delete_blob(pending_correction_key(run.run_id))  # type: ignore[attr-defined]
-    reset_criteria_and_result(services, run)
+        reset_criteria_and_result(services, run)
+        prepared[0] = True
+
+    with_correction_lease(services, run.run_id, prepare)
+    if not prepared[0]:
+        return
     try:
         RunCoordinator(services).drive(run.run_id)
     except CoordinatorDied:
@@ -217,87 +225,29 @@ def on_confirmation(
     by: str,
 ) -> None:
     services = current()
-    put_json(
-        services.state,  # type: ignore[attr-defined]
-        f"confirmation:{capture_id}:{confirmation_id}",
-        {"by": by, **confirmation.model_dump(mode="json")},
-    )
-    try:
-        from server.reads.policy import confirm
-        from server.reads.types import Reads
-    except ImportError:
-        return
-    key = _latest_reads_key(services, capture_id)
-    if key is None:
-        return
-    try:
-        reads = Reads.model_validate(json.loads(services.storage.get_bytes(key)))  # type: ignore[attr-defined]
-    except (OSError, ValueError):
-        return
-    updated = confirm(
-        reads,
-        still_id=confirmation.imageId,
-        choice=confirmation.value if confirmation.kind == "confirmed" else None,
-        typed=confirmation.value if confirmation.kind == "corrected" else None,
-        by=by,  # type: ignore[arg-type]
-        field_id=confirmation.fieldId,
-    )
-    # Persist the corrected record so later results read the answer.
-    services.storage.put_bytes(key, updated.model_dump_json().encode(), "application/json")  # type: ignore[attr-defined]
-    # 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)
-        if newest.status in {"pending", "running"}:
-            # A live run has already consumed the hydrated inputs of its running
-            # stages — verdict_ready can be decided while result is still
-            # uploading — so the corrected reads would never re-enter that pass.
-            # Record it; the coordinator consumes the marker before closing done.
-            _record_pending_correction(services, newest, capture_id)
-        else:
-            _recompose_after_correction(capture_id, newest)
-
-
-def _record_pending_correction(services: object, run: RunRecord, capture_id: str) -> None:
-    """Remember that a correction landed while the run was still composing.
-
-    A live criteria or result stage hydrates its inputs before it executes, so
-    the corrected reads may never re-enter the composed payload — the run would
-    finish done with the pre-correction answer, and a replayed confirmation
-    short-circuits upstream without ever recomposing. The coordinator consumes
-    this marker before it marks the run done and re-runs the two stages once;
-    an earlier-arriving correction (the reads stage still open) is applied by
-    the reads stage itself and needs no marker. Overwriting an existing marker
-    is the idempotent shape: one reset replays every accepted answer.
-    """
-    live = False
-    for stage in ("criteria", "result"):
-        rec = services.state.get_stage(run.run_id, stage)  # type: ignore[attr-defined]
-        if rec is not None and rec.status == "running":
-            live = True
-    if not live:
-        return
-    put_json(
-        services.state,  # type: ignore[attr-defined]
-        pending_correction_key(run.run_id),
-        {"captureId": capture_id, "recordedAt": services.clock.iso()},  # type: ignore[attr-defined]
+    # This insertion, not a reads write or a lease, is the acceptance boundary.
+    # The capture sequence also survives answers arriving before a run exists.
+    record_correction(
+        services.state,
+        capture_id,
+        {
+            "confirmationId": confirmation_id,
+            "by": by,
+            # The API fingerprint excludes clientAt. A retry may have a fresh
+            # client timestamp without changing the accepted answer's identity.
+            **confirmation.model_dump(mode="json", exclude={"clientAt"}),
+        },
     )
-    log_event("confirmation.pending_correction", capture_id=capture_id, run_id=run.run_id)
-
-
-def _latest_reads_key(services: object, capture_id: str) -> str | None:
-    """The reads artifact of the capture's newest run, or None before reads exist."""
-
-    runs = services.state.runs_for_capture(capture_id)  # type: ignore[attr-defined]
-    if not runs:
-        return None
-    run = max(runs, key=lambda item: item.created_at)
-    from server.orchestrator.artifacts import overlay_stage_artifacts
-
-    key = overlay_stage_artifacts(services.state, run).artifacts.get("reads/reads.json")  # type: ignore[attr-defined]
-    return key or None
+    target = newest_correction_run(services.state, capture_id)
+    if target is not None:
+        claim_available_corrections(services.state, target)
+    if target is not None and target.status == "done":
+        try:
+            _recompose_after_correction(capture_id, target)
+        except (LeaseBusy, LeaseLost):
+            # The answer is durable and uncovered. The sweeper retries even if
+            # completion races this status read or the lease owner disappears.
+            log_event("confirmation.recompose_deferred", capture_id=capture_id, run_id=target.run_id)
 
 
 def cancel_capture(capture_id: str) -> None:
diff --git a/backend/server/orchestrator/artifacts.py b/backend/server/orchestrator/artifacts.py
index 09e810f7..279add79 100644
--- a/backend/server/orchestrator/artifacts.py
+++ b/backend/server/orchestrator/artifacts.py
@@ -58,6 +58,7 @@
         "verdicts/verdict.C.json",
         "verdicts/verdict.S.json",
         "verdicts/criteria_report.json",
+        "verdicts/correction-inputs.json",
         "verdicts/placement.C.json",
         "result/result.json",
         "result/result.internal.json",
@@ -87,6 +88,11 @@ def overlay_stage_artifacts(state: object, run: RunRecord) -> RunRecord:
         extra = get_json(state, f"stage_artifacts:{run.run_id}:{name}")
         if isinstance(extra, dict):
             artifacts.update({str(key): str(value) for key, value in extra.items()})
+    from server.orchestrator.corrections import latest_publication
+
+    published = latest_publication(state, run.run_id)
+    if published is not None:
+        artifacts.update({key: value for key, value in published["artifacts"].items() if key.startswith("result/")})
     return run.model_copy(update={"artifacts": artifacts})
 
 
diff --git a/backend/server/orchestrator/coordinator.py b/backend/server/orchestrator/coordinator.py
index 44a1b0ca..9cb450cd 100644
--- a/backend/server/orchestrator/coordinator.py
+++ b/backend/server/orchestrator/coordinator.py
@@ -9,6 +9,7 @@
 
 from pydantic import BaseModel, ConfigDict
 
+from server.common.lease import LeaseBusy, LeaseGuard, LeaseLost, with_lease
 from server.common.logging import log_event
 from server.contracts.api import StageData
 from server.contracts.jobs import (
@@ -22,6 +23,7 @@
 from server.eval.sceneio import validate as validate_document
 from server.orchestrator.artifacts import overlay_stage_artifacts
 from server.orchestrator.clock import parse_iso
+from server.orchestrator.corrections import latest_publication, pending_corrections
 from server.orchestrator.events import emit_failed, emit_stage
 from server.orchestrator.executor import CoordinatorDied, InlineExecutor
 from server.orchestrator.graph import STAGE_NAMES, STAGES, gpu_for_attempt, ready_stages
@@ -44,8 +46,28 @@
 _FINAL = {"verdict_ready", "retake_request", "failed"}
 
 
-def pending_correction_key(run_id: str) -> str:
-    return f"run_pending_correction:{run_id}"
+def correction_lease_name(run_id: str) -> str:
+    """Shared state lease for the correction publish/complete decision."""
+
+    return f"correction:{run_id}"
+
+
+def with_correction_lease(services: Services, run_id: str, fn) -> None:
+    # The backend cannot compare-and-delete a lease token. Losing ownership can
+    # cause duplicate work, but immutable corrections/publications remain safe.
+    try:
+        with_lease(
+            services.state,
+            correction_lease_name(run_id),
+            fn,
+            strict=True,
+            wait_s=0.0,
+            lease_s=30.0,
+            pass_guard=True,
+        )
+    except (LeaseBusy, LeaseLost):
+        # No three-second request delay: the durable sequence is the retry queue.
+        log_event("confirmation.lease_deferred", run_id=run_id)
 
 
 def reset_criteria_and_result(services: Services, run: RunRecord) -> None:
@@ -54,7 +76,7 @@ def reset_criteria_and_result(services: Services, run: RunRecord) -> None:
     The shared tail of both correction paths (a finished run recomposed by
     ``on_confirmation`` and a live run that recorded a pending correction): the
     two stages return to pending, the run re-enters the normal lifecycle, and
-    the re-executed stages overwrite the same artifact keys. The run's deadline
+    the re-executed stages publish new immutable artifact keys. The run's deadline
     measured its earlier pass, so the re-run gets the same HARD_STOP_MIN budget
     ``start_run`` gives a new run, counted from now.
     """
@@ -62,6 +84,8 @@ def reset_criteria_and_result(services: Services, run: RunRecord) -> None:
     now = services.clock.iso()
     for stage in ("criteria", "result"):
         rec = state.get_stage(run.run_id, stage)
+        if rec is not None and rec.status in {"pending", "running"}:
+            continue
         state.upsert_stage(
             StageRecord(
                 run_id=run.run_id,
@@ -579,41 +603,51 @@ def _merge_if_ready(self, run: RunRecord, control: RunControl) -> None:
         control.merge_done = True
         self._save_control(run.run_id, control)
 
-    def _consume_pending_correction(self, run: RunRecord) -> bool:
-        """A correction accepted while criteria or result was mid-flight may sit
-        already-consumed inside their hydrated inputs. Before the run closes as
-        done, send the two stages back through once so the served result carries
-        every accepted answer. Consumed first, so a repeated finish or a replayed
-        hook call cannot reset twice; the second pass's extra final event is
-        already covered by the ``final_event_seq`` the first close recorded.
-        """
-        key = pending_correction_key(run.run_id)
-        marker = get_json(self.s.state, key)
-        if not marker:
+    def _consume_pending_correction(self, run: RunRecord, guard: LeaseGuard | None = None) -> bool:
+        """An answer stays pending until a published version covers it."""
+        if not pending_corrections(self.s.state, run):
             return False
-        self.s.state.delete_blob(key)
+        self._remember_existing_final_event(run)
         reset_criteria_and_result(self.s, run)
         log_event("confirmation.recompose_live", capture_id=run.capture_id, run_id=run.run_id)
         return True
 
     def _maybe_finish(self, run: RunRecord) -> None:
-        stages = self._rows(run.run_id)
-        if any(rec.status == "running" for rec in stages):
-            return
+        def decide_completion(guard: LeaseGuard) -> None:
+            fresh = self._run(run.run_id)
+            stages = self._rows(run.run_id, refresh=True)
+            if any(rec.status == "running" for rec in stages):
+                return
+            control = self._control(run.run_id)
+            status = self._status_map(run.run_id)
+            pending_ready = ready_stages(status)
+            if "criteria" in pending_ready and not control.merge_done:
+                pending_ready = [name for name in pending_ready if name != "criteria"]
+            if pending_ready:
+                return
+            result = self.s.state.get_stage(run.run_id, "result")
+            if result is None or result.status != "done":
+                return
+            if self._consume_pending_correction(fresh, guard):
+                return
+            self._write_run_files(fresh)
+            self._apply_success(fresh)
+            # Completion may race an acceptance after the first snapshot.
+            self._consume_pending_correction(self._run(run.run_id))
+
+        with_correction_lease(self.s, run.run_id, decide_completion)
+
+    def _remember_existing_final_event(self, run: RunRecord) -> None:
+        """Account for a first-pass terminal event before resetting its stages."""
+
         control = self._control(run.run_id)
-        status = self._status_map(run.run_id)
-        pending_ready = ready_stages(status)
-        if "criteria" in pending_ready and not control.merge_done:
-            pending_ready = [name for name in pending_ready if name != "criteria"]
-        if pending_ready:
-            return
-        result = self.s.state.get_stage(run.run_id, "result")
-        if result is None or result.status != "done":
-            return
-        if self._consume_pending_correction(run):
+        if control.final_event_seq is not None:
             return
-        self._write_run_files(run)
-        self._apply_success(run)
+        events = self.s.state.events_after(run.capture_id, 0, limit=500)
+        finals = [event for event in events if event.type in _FINAL and event.run_id == run.run_id]
+        if finals:
+            control.final_event_seq = min(finals, key=lambda event: event.seq).seq
+            self._save_control(run.run_id, control)
 
     def _apply_success(self, run: RunRecord) -> None:
         composed = self._composed_outcome(run)
@@ -713,6 +747,11 @@ def _apply_decision(self, run: RunRecord, decision: Decision, message: str, *, e
                 self._save_control(run.run_id, control)
 
     def _verify_final_event(self, run: RunRecord) -> None:
+        if latest_publication(self.s.state, run.run_id) is not None:
+            from server.orchestrator.publication import emit_latest_publication
+
+            emit_latest_publication(self.s.state, self.s.storage, run.run_id, self.s.settings)
+            return
         events = self.s.state.events_after(run.capture_id, 0, limit=500)
         finals = [event for event in events if event.type in _FINAL and event.run_id == run.run_id]
         control = self._control(run.run_id)
diff --git a/backend/server/orchestrator/correction_claims.py b/backend/server/orchestrator/correction_claims.py
new file mode 100644
index 00000000..611c938b
--- /dev/null
+++ b/backend/server/orchestrator/correction_claims.py
@@ -0,0 +1,47 @@
+"""Append-only ownership generations for accepted corrections.
+
+Only a failed owner permits another generation. An owner that fails immediately
+after insertion therefore cannot strand the answer or erase a successor claim.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from server.contracts.jobs import RunRecord
+
+
+def latest_claim(state: Any, capture_id: str, identifier: str) -> dict | None:
+    prefix = f"claim:{capture_id}:{identifier}"
+    generation = 1
+    latest = None
+    while (row := state.get_kv(f"{prefix}:{generation}")) is not None:
+        latest = row
+        generation += 1
+    return latest
+
+
+def claim_answer(state: Any, run: RunRecord, identifier: str) -> dict:
+    """One contender wins each generation, even if both saw no current owner."""
+    prefix = f"claim:{run.capture_id}:{identifier}"
+    generation = 1
+    while True:
+        key = f"{prefix}:{generation}"
+        owner = state.get_kv(key)
+        if owner is None:
+            candidate = {"runId": run.run_id, "generation": generation}
+            if state.put_if_absent(key, candidate):
+                return candidate
+            continue
+        previous = state.get_run(owner["runId"])
+        if previous is None or previous.status != "failed":
+            return latest_claim(state, run.capture_id, identifier) or owner
+        generation += 1
+
+
+def claim_is_pending(state: Any, capture_id: str, identifier: str) -> bool:
+    owner = latest_claim(state, capture_id, identifier)
+    if owner is None:
+        return True
+    run = state.get_run(owner["runId"])
+    return run is None or run.status == "failed"
diff --git a/backend/server/orchestrator/corrections.py b/backend/server/orchestrator/corrections.py
new file mode 100644
index 00000000..412023dc
--- /dev/null
+++ b/backend/server/orchestrator/corrections.py
@@ -0,0 +1,243 @@
+"""Immutable correction sequences and result versions.
+
+Each slot contains its entire record, inserted with put_if_absent. There is no
+reserved slot, mutable head pointer, or separately deletable pending marker.
+Readers stop at the first empty slot. Competing writers retry that same slot,
+so a crash cannot leave a hole hiding later records.
+
+Acceptance records have no owner. Runs claim answers through immutable claim
+generations, and a failed owner allows a successor generation. Run sequences
+record the answers each composition incorporated without rewriting old results.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+from server.contracts.jobs import RunRecord
+from server.orchestrator.correction_claims import claim_answer, claim_is_pending, latest_claim
+from server.reads.policy import confirm
+from server.reads.types import Reads
+
+
+def _records(state: Any, prefix: str) -> list[dict]:
+    rows = []
+    while (row := state.get_kv(f"{prefix}:{len(rows) + 1}")) is not None:
+        rows.append(row)
+    return rows
+
+
+def _append_correction(state: Any, prefix: str, correction: dict) -> int:
+    sequence = 1
+    while True:
+        key = f"{prefix}:{sequence}"
+        prior = state.get_kv(key)
+        if prior is None:
+            if state.put_if_absent(key, correction):
+                return sequence
+            continue
+        if prior["confirmationId"] == correction["confirmationId"]:
+            # Ignore metadata written by earlier deployments when replaying an ID.
+            if {k: v for k, v in prior.items() if k not in {"targetRunId", "legacy"}} != {
+                k: v for k, v in correction.items() if k not in {"targetRunId", "legacy"}
+            }:
+                raise ValueError("confirmation ID was reused with different correction data")
+            return sequence
+        sequence += 1
+
+
+def newest_correction_run(state: Any, capture_id: str) -> RunRecord | None:
+    runs = state.runs_for_capture(capture_id)
+    live = [run for run in runs if run.status in {"pending", "running", "done"}]
+    if not live:
+        return None
+    newest = max(live, key=lambda item: (item.created_at, item.run_id))
+    # A failed amendment must not send new answers back to delivered history.
+    if newest.status == "done" and any(
+        (run.created_at, run.run_id) > (newest.created_at, newest.run_id) for run in runs
+    ):
+        return None
+    return newest
+
+
+def record_correction(state: Any, capture_id: str, correction: dict) -> int:
+    """Acceptance journals the answer, never a possibly stale run selection."""
+    return _append_correction(state, f"capture_correction:{capture_id}", correction)
+
+
+def _restore_previous_claims(state: Any, capture_id: str, rows: list[dict]) -> None:
+    # Preserve ownership already recorded by the fixed-target implementation.
+    # Failed owners remain eligible for the same generational takeover as new ones.
+    for row in rows:
+        identifier = row["confirmationId"]
+        previous = row.get("targetRunId")
+        if previous is None:
+            previous = (state.get_kv(f"pre_run_owner:{capture_id}:{identifier}") or {}).get("runId")
+        if previous is not None:
+            state.put_if_absent(f"claim:{capture_id}:{identifier}:1", {"runId": previous, "generation": 1})
+
+
+def claim_available_corrections(state: Any, run: RunRecord) -> None:
+    rows = _records(state, f"capture_correction:{run.capture_id}")
+    _restore_previous_claims(state, run.capture_id, rows)
+    newest = newest_correction_run(state, run.capture_id)
+    if newest is None or newest.run_id != run.run_id:
+        return
+    for row in rows:
+        if claim_is_pending(state, run.capture_id, row["confirmationId"]):
+            claim_answer(state, newest, row["confirmationId"])
+
+
+def seed_run_corrections(state: Any, run: RunRecord) -> None:
+    """Adopt available answers after the pending run record is durable."""
+    import_legacy_corrections(state, run.capture_id)
+    claim_available_corrections(state, run)
+
+
+def pending_answers(state: Any, capture_id: str) -> list[dict]:
+    rows = _records(state, f"capture_correction:{capture_id}")
+    _restore_previous_claims(state, capture_id, rows)
+    return [row for row in rows if claim_is_pending(state, capture_id, row["confirmationId"])]
+
+
+def correction_snapshot(state: Any, run: RunRecord) -> list[dict]:
+    claim_available_corrections(state, run)
+    rows = _records(state, f"capture_correction:{run.capture_id}")
+    order = {row["confirmationId"]: sequence for sequence, row in enumerate(rows, 1)}
+    for row in rows:
+        owner = latest_claim(state, run.capture_id, row["confirmationId"])
+        if owner is not None and owner["runId"] == run.run_id:
+            answer = {key: value for key, value in row.items() if key not in {"targetRunId", "legacy"}}
+            _append_correction(state, f"run_correction:{run.run_id}", answer)
+    # A failed owner's answer may be adopted after a newer answer was already
+    # composed. Coverage remains append-only, while replay follows acceptance order.
+    return [
+        {**row, "acceptedSequence": order[row["confirmationId"]]}
+        for row in _records(state, f"run_correction:{run.run_id}")
+    ]
+
+
+def import_legacy_corrections(state: Any, capture_id: str, run: RunRecord | None = None) -> None:
+    """Preserve answers acknowledged by old API workers before reads existed."""
+    from server.reads.pending import pending_slots
+
+    for row in sorted(pending_slots(state, capture_id), key=lambda row: row.get("seq", 0)):
+        required = ("confirmationId", "fieldId", "imageId", "kind", "by")
+        missing = [key for key in required if key not in row]
+        if missing:
+            raise ValueError(f"legacy confirmation is missing required fields: {', '.join(missing)}")
+        answer = {key: row[key] for key in required}
+        answer["value"] = row.get("value")
+        record_correction(state, capture_id, answer)
+    if run is not None:
+        claim_available_corrections(state, run)
+
+
+def fold_reads(base: Reads, corrections: list[dict]) -> Reads:
+    """Replay accepted answers in sequence; the latest answer wins per field.
+
+    Authorization and device conflict checks belong to acceptance. Replaying an
+    accepted answer must not conflict with locks left by an earlier answer or by
+    a legacy reads artifact that already contains that answer.
+    """
+    reads = base.model_copy(deep=True)
+    ordered = sorted(enumerate(corrections), key=lambda item: item[1].get("acceptedSequence", item[0]))
+    for _, row in ordered:
+        field = row["fieldId"]
+        if reads.confirm is not None:
+            reads.confirm.locks.pop(field, None)
+        reads = confirm(
+            reads,
+            still_id=row["imageId"],
+            choice=row.get("value") if row["kind"] == "confirmed" else None,
+            typed=row.get("value") if row["kind"] == "corrected" else None,
+            by=row["by"],
+            field_id=field,
+        )
+    return reads
+
+
+def prepare_inputs(state: Any, run: RunRecord, stage: str, work: Path) -> Path:
+    """Criteria and result must consume the same immutable reads snapshot."""
+    path = work / "correction-inputs.json"
+    reads_path = work / "reads/reads.json"
+    hydrated = work / "verdicts/correction-inputs.json"
+    if stage == "result" and hydrated.is_file():
+        path.write_bytes(hydrated.read_bytes())
+    if stage == "criteria":
+        import_legacy_corrections(state, run.capture_id, run)
+        rows = correction_snapshot(state, run)
+        raw = json.loads(reads_path.read_text()) if reads_path.is_file() else None
+        if rows:
+            if raw is None:
+                raise RuntimeError("correction composition has no base reads artifact")
+            raw = fold_reads(Reads.model_validate(raw), rows).model_dump(mode="json")
+        snapshot = {"correctionIds": [row["confirmationId"] for row in rows], "reads": raw}
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(json.dumps(snapshot))
+    elif path.is_file():
+        snapshot = json.loads(path.read_text())
+    else:
+        # Legacy criteria did not record a snapshot. Publish coverage zero, then
+        # the coordinator sees uncovered answers and rebuilds both stages.
+        snapshot = {"correctionIds": [], "reads": None}
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(json.dumps(snapshot))
+    if snapshot["reads"] is not None:
+        reads_path.parent.mkdir(parents=True, exist_ok=True)
+        reads_path.write_text(json.dumps(snapshot["reads"]))
+    return path
+
+
+def latest_publication(state: Any, run_id: str) -> dict | None:
+    versions = _records(state, f"result_version:{run_id}")
+    return versions[-1] if versions else None
+
+
+def publish_version(state: Any, run_id: str, publication: dict) -> dict:
+    """Append only if this composition covers every previously published answer.
+
+    All snapshots are prefixes of one immutable sequence. Comparing their IDs
+    rejects a stale composer even if it deletes a successor's advisory lease.
+    Artifact bytes must already exist at this composition's unique storage keys.
+    """
+    sequence = 1
+    ids = publication["correctionIds"]
+    while True:
+        key = f"result_version:{run_id}:{sequence}"
+        prior = state.get_kv(key)
+        if prior is None:
+            # Preserve historical final events, but identify an older emitter
+            # whose append arrives after this publication. This watermark is
+            # internal metadata, not an extra field in the device event.
+            event_floor = 0
+            if publication.get("captureId"):
+                while page := state.events_after(publication["captureId"], event_floor):
+                    event_floor = page[-1].seq
+            candidate = {**publication, "version": sequence, "afterEventSeq": event_floor}
+            if state.put_if_absent(key, candidate):
+                return candidate
+            continue
+        previous = prior["correctionIds"]
+        if ids[: len(previous)] != previous:
+            # Another writer may have appended beyond the slot just inspected.
+            return latest_publication(state, run_id) or prior
+        if all(
+            prior.get(key) == value for key, value in publication.items() if key not in {"version", "afterEventSeq"}
+        ):
+            return latest_publication(state, run_id) or prior
+        # Equal coverage is a superset too. Upstream verdicts can change without
+        # a new correction; their recomposition must still publish a new result.
+        sequence += 1
+
+
+def pending_corrections(state: Any, run: RunRecord) -> bool:
+    fresh = state.get_run(run.run_id)
+    if fresh is not None and fresh.status == "failed":
+        return False
+    corrections = correction_snapshot(state, run)
+    published = latest_publication(state, run.run_id)
+    covered = published["correctionIds"] if published else []
+    return len(corrections) > len(covered)
diff --git a/backend/server/orchestrator/publication.py b/backend/server/orchestrator/publication.py
new file mode 100644
index 00000000..9e817e41
--- /dev/null
+++ b/backend/server/orchestrator/publication.py
@@ -0,0 +1,78 @@
+"""Deliver final events only for immutable published results."""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from server.orchestrator.corrections import latest_publication
+
+
+def emit_latest_publication(state: Any, storage: Any, run_id: str, settings: Any) -> None:
+    publication = latest_publication(state, run_id)
+    if publication is None:
+        return
+    capture_id = publication["captureId"]
+    after = 0
+    while page := state.events_after(capture_id, after):
+        for event in page:
+            version = event.data.get("_resultVersion")
+            if event.run_id == run_id and version is not None:
+                prior = state.get_kv(f"result_version:{run_id}:{version}")
+                # Preserve the existing retry contract: unchanged correction
+                # coverage does not emit another final. A new answer always does.
+                if (
+                    prior is not None
+                    and prior["correctionIds"] == publication["correctionIds"]
+                    and public_final_event(state, event) is not None
+                ):
+                    return
+        after = page[-1].seq
+    # No lease can fence append_event. A duplicate emitter is harmless, and the
+    # event reader filters superseded versions without changing the device data.
+    # A crash before this append is recovered from the publication by the sweep.
+    for event_type, data in publication["events"]:
+        state.append_event(
+            capture_id, event_type, {**data, "_resultVersion": publication["version"]}, run_id=run_id, stage="result"
+        )
+    state.set_status(capture_id, publication["status"])
+    if publication.get("deliverWebhook"):
+        from server.contracts.result import ResultResponse
+        from server.result.compose import ComposeResult, _deliver
+        from server.result.config import load_config
+
+        capture = state.get_capture(capture_id)
+        if capture is not None:
+            artifacts = publication["artifacts"]
+            response = ResultResponse.model_validate_json(storage.get_bytes(artifacts["result/result.json"]))
+            internal = json.loads(storage.get_bytes(artifacts["result/result.internal.json"]))
+            composed = ComposeResult(response, internal, publication["events"], capture_id, run_id)
+            _deliver(composed, capture, load_config(settings))
+
+
+def public_final_event(state: Any, event: Any) -> tuple[str, dict] | None:
+    """Strip internal coverage and suppress stale or duplicate final events."""
+    version = event.data.get("_resultVersion")
+    if version is None:
+        return event.type, event.data
+    latest = latest_publication(state, event.run_id)
+    if latest is None:
+        return None
+    # Keep finals that were already in history before a recomposition. Only
+    # suppress an old emitter that appended beyond a successor's watermark.
+    for number in range(version + 1, latest["version"] + 1):
+        successor = state.get_kv(f"result_version:{event.run_id}:{number}")
+        if successor is not None and event.seq > successor["afterEventSeq"]:
+            return None
+    after = 0
+    while page := state.events_after(event.capture_id, after):
+        if any(
+            prior.seq < event.seq and prior.run_id == event.run_id and prior.data.get("_resultVersion") == version
+            for prior in page
+        ):
+            return None
+        if page[-1].seq >= event.seq:
+            break
+        after = page[-1].seq
+    data = {key: value for key, value in event.data.items() if key != "_resultVersion"}
+    return event.type, data
diff --git a/backend/server/orchestrator/runner.py b/backend/server/orchestrator/runner.py
index f864bc17..33956552 100644
--- a/backend/server/orchestrator/runner.py
+++ b/backend/server/orchestrator/runner.py
@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import importlib
+import json
+import uuid
 from pathlib import Path
 from typing import Any
 
@@ -57,7 +59,37 @@ def run_stage(
         result = fn(ctx)
     if not isinstance(result, StageResult):
         raise RuntimeError(f"{ctx.stage} returned {type(result).__name__}")
+    snapshot_path = ctx.inputs.get("correction_snapshot")
+    if ctx.stage == "criteria" and snapshot_path:
+        result.artifacts["verdicts/correction-inputs.json"] = snapshot_path
     stored = _publish(ctx, result)
+    if ctx.stage == "result" and snapshot_path:
+        from server.orchestrator.corrections import publish_version
+        from server.orchestrator.publication import emit_latest_publication
+
+        if "result/result.json" not in stored:
+            raise ValueError("artifact_missing: result/result.json was not published")
+        snapshot = json.loads(Path(snapshot_path).read_text())
+        event_path = ctx.work_dir / "result/publication.json"
+        if not event_path.is_file():
+            raise RuntimeError("versioned result did not supply its final event")
+        intent = json.loads(event_path.read_text())
+        events = intent.get("events", [])
+        if len(events) != 1 or events[0][0] not in {"verdict_ready", "retake_request"}:
+            raise ValueError("result_no_event: result must supply exactly one final event")
+        # Bytes precede the atomic publication. A crash leaves only unreferenced
+        # objects, never a published version pointing at partially written data.
+        publication = publish_version(
+            ctx.state,
+            ctx.run_id,
+            {
+                "correctionIds": snapshot["correctionIds"],
+                "artifacts": {**json.loads(ctx.inputs.get("criteria_artifacts", "{}")), **stored},
+                **intent,
+            },
+        )
+        stored = {key: value for key, value in publication["artifacts"].items() if key.startswith("result/")}
+        emit_latest_publication(ctx.state, ctx.storage, ctx.run_id, ctx.settings)
     result.artifacts = stored
     return result
 
@@ -65,6 +97,10 @@ def run_stage(
 def _publish(ctx: StageContext, result: StageResult) -> dict[str, str]:
     stored: dict[str, str] = {}
     prefix = f"captures/{ctx.capture_id}/runs/{ctx.run_id}"
+    if ctx.stage in {"criteria", "result"} and ctx.inputs.get("correction_snapshot"):
+        # Ownership checks are separate RPCs, not fencing. Every composer gets
+        # new object keys; no stale worker can overwrite a published version.
+        prefix = f"{prefix}/compositions/{uuid.uuid4().hex}"
     root = ctx.work_dir.resolve()
     for key, rel in result.artifacts.items():
         if _safe_rel(key) is None:
diff --git a/backend/server/orchestrator/services.py b/backend/server/orchestrator/services.py
index a6fcbab4..612e3bd6 100644
--- a/backend/server/orchestrator/services.py
+++ b/backend/server/orchestrator/services.py
@@ -7,6 +7,7 @@
 import hashlib
 import json
 import tempfile
+import uuid
 from collections.abc import Iterator
 from dataclasses import dataclass
 from datetime import timedelta
@@ -45,6 +46,10 @@ class Services:
     def stage_context(self, run: RunRecord, rec: StageRecord, call_id: str) -> StageContext:
         run = overlay_stage_artifacts(self.state, run)
         work = self.work_root / run.run_id / rec.stage / f"attempt-{rec.attempt}"
+        if rec.stage in {"criteria", "result"}:
+            # An expired lease can leave two composers at the same attempt.
+            # Neither may hydrate or overwrite the other's local inputs.
+            work = work / uuid.uuid4().hex
         work.mkdir(parents=True, exist_ok=True)
         packet_dir = work / "packet"
         packet_dir.mkdir(exist_ok=True)
@@ -64,6 +69,16 @@ def stage_context(self, run: RunRecord, rec: StageRecord, call_id: str) -> Stage
         hydrate_artifacts(self.storage, work, select_stage_artifacts(rec.stage, run.artifacts))
         spent = sum(float(run.spend_usd.get(key, 0.0)) for key in ("vlm_reads", "vlm_objects", "vlm_c3"))
         inputs = {"packet_dir": str(packet_dir), "run_control": str(work / "run_control.json")}
+        if rec.stage == "reads":
+            inputs["immutable_corrections"] = "1"
+        if rec.stage in {"criteria", "result"}:
+            from server.orchestrator.corrections import prepare_inputs
+
+            inputs["correction_snapshot"] = str(prepare_inputs(self.state, run, rec.stage, work))
+            if rec.stage == "result":
+                inputs["criteria_artifacts"] = json.dumps(
+                    {key: value for key, value in run.artifacts.items() if key.startswith("verdicts/")}
+                )
         (work / "run_control.json").write_text(
             json.dumps({"run": run.model_dump(mode="json"), "stage": rec.stage, "call_id": call_id})
         )
diff --git a/backend/server/orchestrator/stubs/reads.py b/backend/server/orchestrator/stubs/reads.py
index cef50127..89851a1a 100644
--- a/backend/server/orchestrator/stubs/reads.py
+++ b/backend/server/orchestrator/stubs/reads.py
@@ -28,7 +28,7 @@ def run(ctx: StageContext) -> StageResult:
         from server.reads.types import ConfirmBlock, MeterRead, Reads
 
         rows = pending_slots(ctx.state, ctx.capture_id)
-        if rows:
+        if rows or ctx.inputs.get("immutable_corrections"):
             reads = Reads(
                 capture_id=ctx.capture_id,
                 status="complete",
@@ -42,7 +42,13 @@ def run(ctx: StageContext) -> StageResult:
             )
             from server.reads.pending import apply_pending
 
-            body = Reads.model_validate(apply_pending(ctx.state, ctx.capture_id, reads)).model_dump(mode="json")
+            if ctx.inputs.get("immutable_corrections"):
+                from server.orchestrator.corrections import import_legacy_corrections
+
+                import_legacy_corrections(ctx.state, ctx.capture_id, ctx.state.get_run(ctx.run_id))
+            else:
+                reads = apply_pending(ctx.state, ctx.capture_id, reads)
+            body = 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 856898f1..a10e27e1 100644
--- a/backend/server/orchestrator/stubs/result.py
+++ b/backend/server/orchestrator/stubs/result.py
@@ -16,17 +16,32 @@
 def run(ctx: StageContext) -> StageResult:
     run_rec = ctx.state.get_run(ctx.run_id)
     key = run_rec.artifacts["verdicts/verdict.C.json"]
-    verdict_bytes = ctx.storage.get_bytes(key)
+    hydrated = Path(ctx.work_dir) / "verdicts/verdict.C.json"
+    verdict_bytes = hydrated.read_bytes() if hydrated.is_file() else ctx.storage.get_bytes(key)
     digest = sha256(verdict_bytes)
     verdict = json.loads(verdict_bytes)
     decision = verdict["site"]["decision"]
     kind = "manual_review" if decision == "manual_review" else decision
-    ctx.state.append_event(
-        ctx.capture_id,
-        "verdict_ready",
-        {"runId": ctx.run_id, "kind": kind},
-        run_id=ctx.run_id,
-    )
+    if not ctx.inputs.get("correction_snapshot"):
+        ctx.state.append_event(
+            ctx.capture_id,
+            "verdict_ready",
+            {"runId": ctx.run_id, "kind": kind},
+            run_id=ctx.run_id,
+        )
+    else:
+        out = Path(ctx.work_dir) / "result"
+        out.mkdir(parents=True, exist_ok=True)
+        (out / "publication.json").write_text(
+            json.dumps(
+                {
+                    "captureId": ctx.capture_id,
+                    "events": [["verdict_ready", {"runId": ctx.run_id, "kind": kind}]],
+                    "status": "manual_review",
+                    "deliverWebhook": False,
+                }
+            )
+        )
     response = ResultResponse(
         runId=ctx.run_id,
         status="manual_review",
diff --git a/backend/server/orchestrator/sweeper.py b/backend/server/orchestrator/sweeper.py
index ecf5341e..9f95b9c9 100644
--- a/backend/server/orchestrator/sweeper.py
+++ b/backend/server/orchestrator/sweeper.py
@@ -27,6 +27,23 @@ def coordinator_is_dead(run, raw: dict, *, now, stale_s: float) -> bool:
 
 def sweep(services: Services, *, retry_schedule: list[float] | None = None) -> int:
     touched = 0
+    # A crash can happen after acceptance or publication but before the mutable
+    # run status is reopened. Terminal runs therefore also need this check.
+    from server.orchestrator.api import _recompose_after_correction
+    from server.orchestrator.corrections import claim_available_corrections, pending_corrections
+    from server.orchestrator.publication import emit_latest_publication
+    from server.orchestrator.services import bind
+
+    # Claim answers even when a healthy coordinator needs no respawn. A run
+    # that failed after claiming leaves an available next generation.
+    for run in services.state.list_runs(status={"pending", "running", "done"}):
+        claim_available_corrections(services.state, run)
+    for run in services.state.list_runs(status={"done"}):
+        emit_latest_publication(services.state, services.storage, run.run_id, services.settings)
+        if pending_corrections(services.state, run):
+            with bind(services):
+                _recompose_after_correction(run.capture_id, run)
+            touched += 1
     runs = services.state.list_runs(status={"pending", "running"})  # type: ignore[attr-defined]
     for run in runs:
         raw = dict(get_json(services.state, f"run_control:{run.run_id}") or {})
diff --git a/backend/server/orchestrator/tests/test_correction_claim_races.py b/backend/server/orchestrator/tests/test_correction_claim_races.py
new file mode 100644
index 00000000..34b1c452
--- /dev/null
+++ b/backend/server/orchestrator/tests/test_correction_claim_races.py
@@ -0,0 +1,202 @@
+"""Portable regressions for creation windows and failed correction owners."""
+
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor
+from threading import Barrier, Event, local
+
+import pytest
+
+from server.orchestrator.api import start_run
+from server.orchestrator.corrections import (
+    correction_snapshot,
+    latest_publication,
+    record_correction,
+    seed_run_corrections,
+)
+from server.orchestrator.services import bind
+from server.orchestrator.sweeper import sweep
+from server.orchestrator.tests.conftest import build_services
+from server.orchestrator.tests.test_correction_run_scope import next_run, public_events
+from server.orchestrator.tests.test_greptile_followup import _run
+from server.orchestrator.tests.test_immutable_correction_races import accept, request, served, setup_run
+
+
+def answer(identifier="answer"):
+    return {"confirmationId": identifier, "by": "device", **request().model_dump(mode="json", exclude={"clientAt"})}
+
+
+def start(services, uri, digest):
+    with bind(services):
+        return start_run("cap_test", digest, uri)
+
+
+def assert_incorporated_once(services, run, identifier):
+    assert served(services, run)["meter"]["value"] == "12345"
+    assert latest_publication(services.state, run.run_id)["correctionIds"] == [identifier]
+    assert [row["confirmationId"] for row in correction_snapshot(services.state, run)] == [identifier]
+    assert services.state.get_kv("capture_correction:cap_test:2") is None
+
+
+@pytest.mark.parametrize("delivered_predecessor", [False, True])
+def test_confirmation_between_seed_and_run_write_reaches_new_run(tmp_path, monkeypatch, delivered_predecessor):
+    import server.orchestrator.api as api
+
+    if delivered_predecessor:
+        services, old, _ = setup_run(tmp_path)
+        old.status = "done"
+        services.state.put_run(old)
+        before = served(services, old)
+        events = public_events(services, old)
+    else:
+        services, uri, digest = build_services(tmp_path, "memory")
+    seeded = Event()
+    resume = Event()
+    original = api.seed_run_corrections
+
+    def stop_after_seed(state, run):
+        original(state, run)
+        seeded.set()
+        assert resume.wait(10)
+
+    monkeypatch.setattr(api, "seed_run_corrections", stop_after_seed)
+    with ThreadPoolExecutor(max_workers=1) as pool:
+        future = (
+            pool.submit(next_run, services, 2) if delivered_predecessor else pool.submit(start, services, uri, digest)
+        )
+        try:
+            assert seeded.wait(10), "run creation never reached the seed boundary"
+            accept(services, request(), "between")
+        finally:
+            resume.set()
+        run = future.result(timeout=20)
+    assert_incorporated_once(services, run, "between")
+    if delivered_predecessor:
+        sweep(services)
+        assert served(services, old) == before
+        assert public_events(services, old) == events
+        assert correction_snapshot(services.state, old) == []
+
+
+def test_confirmation_while_run_is_pending_reaches_that_run(tmp_path, monkeypatch):
+    services, uri, digest = build_services(tmp_path, "memory")
+    pending = Event()
+    resume = Event()
+    original = services.state.put_run
+
+    def stop_after_pending_write(run):
+        original(run)
+        if run.status == "pending" and not pending.is_set():
+            pending.set()
+            assert resume.wait(10)
+
+    monkeypatch.setattr(services.state, "put_run", stop_after_pending_write)
+    with ThreadPoolExecutor(max_workers=1) as pool:
+        future = pool.submit(start, services, uri, digest)
+        try:
+            assert pending.wait(10), "run never reached its durable pending state"
+            accept(services, request(), "pending")
+        finally:
+            resume.set()
+        run = future.result(timeout=20)
+    assert_incorporated_once(services, run, "pending")
+
+
+@pytest.mark.parametrize("failure_boundary", ["journal_insert", "after_ack"])
+def test_running_owner_failure_cannot_strand_an_answer(tmp_path, monkeypatch, failure_boundary):
+    services, failed, _ = setup_run(tmp_path)
+    accepted = Event()
+    resume = Event()
+    original = services.state.put_if_absent
+
+    def pause_after_insert(key, row):
+        inserted = original(key, row)
+        if failure_boundary == "journal_insert" and key == "capture_correction:cap_test:1":
+            accepted.set()
+            assert resume.wait(10)
+        return inserted
+
+    def submit():
+        accept(services, request(), "survives-failure")
+        if failure_boundary == "after_ack":
+            accepted.set()
+            assert resume.wait(10)
+
+    monkeypatch.setattr(services.state, "put_if_absent", pause_after_insert)
+    with ThreadPoolExecutor(max_workers=1) as pool:
+        future = pool.submit(submit)
+        try:
+            assert accepted.wait(10), "acceptance never reached the selected boundary"
+            failed.status = "failed"
+            services.state.put_run(failed)
+        finally:
+            resume.set()
+        future.result(timeout=20)
+    following = next_run(services, 2)
+    assert_incorporated_once(services, following, "survives-failure")
+    assert services.state.get_run(failed.run_id).status == "failed"
+    if failure_boundary == "after_ack":
+        assert services.state.get_kv("claim:cap_test:survives-failure:1")["runId"] == failed.run_id
+        assert services.state.get_kv("claim:cap_test:survives-failure:2")["runId"] == following.run_id
+
+
+def test_two_runs_racing_to_take_over_have_one_owner(tmp_path, monkeypatch):
+    services, _, _ = build_services(tmp_path, "memory")
+    state = services.state
+    previous = _run(services, "run_previous", created_at="2026-10-01T00:00:00Z")
+    state.put_run(previous)
+    record_correction(state, "cap_test", answer())
+    seed_run_corrections(state, previous)
+    previous.status = "failed"
+    state.put_run(previous)
+    contenders = [_run(services, name) for name in ("run_b", "run_c")]
+    for run in contenders:
+        state.put_run(run)
+    observed = local()
+    original_runs = state.runs_for_capture
+    original_insert = state.put_if_absent
+    gate = Barrier(2)
+    attempts = []
+
+    def observed_newest(capture_id):
+        # Each creator observed itself as newest before the other became visible.
+        return [observed.run] if hasattr(observed, "run") else original_runs(capture_id)
+
+    def contend(key, row):
+        if key == "claim:cap_test:answer:2":
+            attempts.append(row["runId"])
+            gate.wait(timeout=5)
+        return original_insert(key, row)
+
+    def seed(run):
+        observed.run = run
+        seed_run_corrections(state, run)
+
+    with monkeypatch.context() as patch:
+        patch.setattr(state, "runs_for_capture", observed_newest)
+        patch.setattr(state, "put_if_absent", contend)
+        with ThreadPoolExecutor(max_workers=2) as pool:
+            futures = [pool.submit(seed, run) for run in contenders]
+            for future in futures:
+                future.result(timeout=10)
+    snapshots = [correction_snapshot(state, run) for run in contenders]
+    assert sorted(len(rows) for rows in snapshots) == [0, 1]
+    assert sorted(attempts) == ["run_b", "run_c"], "both contenders must reach generation two"
+    assert state.get_kv("claim:cap_test:answer:1")["runId"] == previous.run_id
+    assert state.get_kv("claim:cap_test:answer:3") is None
+
+
+def test_sweeper_never_assigns_unowned_answer_to_delivered_predecessor(tmp_path):
+    services, old, _ = setup_run(tmp_path)
+    old.status = "done"
+    services.state.put_run(old)
+    before = served(services, old)
+    events = public_events(services, old)
+    newest = _run(services, "run_newest", created_at="2099-01-01T00:00:00Z")
+    services.state.put_run(newest)
+    record_correction(services.state, "cap_test", answer())
+    sweep(services)
+    assert served(services, old) == before
+    assert public_events(services, old) == events
+    assert correction_snapshot(services.state, old) == []
+    assert [row["confirmationId"] for row in correction_snapshot(services.state, newest)] == ["answer"]
diff --git a/backend/server/orchestrator/tests/test_correction_claims.py b/backend/server/orchestrator/tests/test_correction_claims.py
new file mode 100644
index 00000000..8f45b10a
--- /dev/null
+++ b/backend/server/orchestrator/tests/test_correction_claims.py
@@ -0,0 +1,89 @@
+"""Exactly one run owns each correction claim generation."""
+
+from concurrent.futures import ThreadPoolExecutor
+from threading import Barrier
+
+import pytest
+
+from server.orchestrator.correction_claims import claim_answer, claim_is_pending, latest_claim
+from server.orchestrator.corrections import correction_snapshot, fold_reads, pending_answers, record_correction
+from server.orchestrator.tests.conftest import build_services
+from server.orchestrator.tests.test_correction_versions import correction
+from server.orchestrator.tests.test_greptile_followup import _run
+from server.reads.types import ConfirmBlock, MeterRead, Reads
+
+
+@pytest.mark.parametrize("backend", ["memory", "modal_dict"])
+def test_competing_runs_get_one_owner_per_generation(tmp_path, monkeypatch, backend):
+    services, _, _ = build_services(tmp_path, backend)
+    state = services.state
+    runs = [_run(services, name) for name in ("run_a", "run_b", "run_c")]
+    for run in runs:
+        state.put_run(run)
+    original = state.put_if_absent
+    first_slot = "claim:cap_test:answer:1"
+    gate = Barrier(2)
+    attempted = []
+
+    def contend(key, row):
+        if key == first_slot:
+            attempted.append(row["runId"])
+            gate.wait(timeout=5)
+        return original(key, row)
+
+    with monkeypatch.context() as patch:
+        patch.setattr(state, "put_if_absent", contend)
+        with ThreadPoolExecutor(max_workers=2) as pool:
+            futures = [pool.submit(claim_answer, state, run, "answer") for run in runs[:2]]
+            owners = [future.result(timeout=10) for future in futures]
+    assert sorted(attempted) == ["run_a", "run_b"]
+    assert owners[0] == owners[1]
+    assert state.get_kv("claim:cap_test:answer:2") is None
+    assert not claim_is_pending(state, "cap_test", "answer")
+    failed = state.get_run(owners[0]["runId"])
+    failed.status = "failed"
+    state.put_run(failed)
+    assert claim_is_pending(state, "cap_test", "answer")
+    second = claim_answer(state, runs[2], "answer")
+    assert second == {"runId": "run_c", "generation": 2}
+    assert state.get_kv(first_slot) == owners[0]
+    assert latest_claim(state, "cap_test", "answer") == second
+    assert not claim_is_pending(state, "cap_test", "answer")
+
+
+def test_delivered_owner_cannot_be_taken_over(tmp_path):
+    services, _, _ = build_services(tmp_path, "memory")
+    first = _run(services, "run_a", status="done")
+    second = _run(services, "run_b")
+    for run in (first, second):
+        services.state.put_run(run)
+    owner = claim_answer(services.state, first, "answer")
+    assert claim_answer(services.state, second, "answer") == owner
+    assert services.state.get_kv("claim:cap_test:answer:2") is None
+
+
+def test_late_takeover_does_not_replace_a_more_recent_field_answer(tmp_path):
+    services, _, _ = build_services(tmp_path, "memory")
+    state = services.state
+    first = _run(services, "run_a", created_at="2026-10-01T00:00:00Z")
+    state.put_run(first)
+    record_correction(state, "cap_test", correction("old", value="11111"))
+    assert [row["confirmationId"] for row in pending_answers(state, "cap_test")] == ["old"]
+    correction_snapshot(state, first)
+    second = _run(services, "run_b", created_at="2026-10-02T00:00:00Z")
+    state.put_run(second)
+    record_correction(state, "cap_test", correction("new", value="22222"))
+    assert [row["confirmationId"] for row in correction_snapshot(state, second)] == ["new"]
+    first.status = "failed"
+    state.put_run(first)
+    assert [row["confirmationId"] for row in pending_answers(state, "cap_test")] == ["old"]
+    adopted = correction_snapshot(state, second)
+    assert [row["confirmationId"] for row in adopted] == ["new", "old"]
+    base = Reads(
+        capture_id="cap_test",
+        status="complete",
+        meter=MeterRead(status="auto_fill", value="00000"),
+        confirm=ConfirmBlock(candidates=["00000"]),
+    )
+    assert fold_reads(base, adopted).meter.value == "22222"
+    assert pending_answers(state, "cap_test") == []
diff --git a/backend/server/orchestrator/tests/test_correction_run_scope.py b/backend/server/orchestrator/tests/test_correction_run_scope.py
new file mode 100644
index 00000000..221c9228
--- /dev/null
+++ b/backend/server/orchestrator/tests/test_correction_run_scope.py
@@ -0,0 +1,87 @@
+"""A correction must not reopen another run's delivered result."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from types import SimpleNamespace
+
+from server.api.events import poll_events
+from server.orchestrator.api import start_run
+from server.orchestrator.corrections import correction_snapshot, latest_publication, pending_corrections
+from server.orchestrator.services import bind, packet_sha256
+from server.orchestrator.sweeper import sweep
+from server.orchestrator.tests.test_immutable_correction_races import accept, request, served, setup_run
+
+
+def next_run(services, number):
+    key = f"captures/cap_test/amendments/test-{number}/packet.json"
+    packet = json.loads(services.storage.get_bytes("captures/cap_test/packet.json"))
+    packet["testRevision"] = number
+    raw = json.dumps(packet).encode()
+    services.storage.put_bytes(key, raw, "application/json")
+    with bind(services):
+        return start_run("cap_test", packet_sha256(raw), key)
+
+
+def public_events(services, run):
+    deps = SimpleNamespace(state=services.state, polls={}, wake={})
+    capture = services.state.get_capture(run.capture_id)
+    response = asyncio.run(poll_events(deps, capture, 0, 0))
+    run_seqs = {
+        event.seq for event in services.state.events_after(run.capture_id, 0, limit=500) if event.run_id == run.run_id
+    }
+    return [event.model_dump(mode="json") for event in response.events if event.seq in run_seqs]
+
+
+def test_correcting_newest_run_does_not_rewrite_delivered_history(tmp_path):
+    services, first, _ = setup_run(tmp_path)
+    first.status = "done"
+    services.state.put_run(first)
+    before_result = served(services, first)
+    before_publication = latest_publication(services.state, first.run_id)
+    before_events = public_events(services, first)
+    second = next_run(services, 2)
+    assert second.status == "done"
+    accept(services, request(), "only-second")
+    sweep(services)
+    assert served(services, second)["meter"]["value"] == "12345"
+    assert served(services, first) == before_result
+    assert latest_publication(services.state, first.run_id) == before_publication
+    assert public_events(services, first) == before_events
+    assert correction_snapshot(services.state, first) == []
+
+
+def test_failed_run_does_not_own_or_retry_a_later_answer(tmp_path):
+    services, failed, _ = setup_run(tmp_path)
+    failed.status = "failed"
+    services.state.put_run(failed)
+    before_events = public_events(services, failed)
+    accept(services, request(), "after-failure")
+    assert not pending_corrections(services.state, failed)
+    sweep(services)
+    assert services.state.get_run(failed.run_id).status == "failed"
+    assert public_events(services, failed) == before_events
+    assert correction_snapshot(services.state, failed) == []
+
+
+def test_answer_after_failure_is_copied_into_exactly_the_next_run(tmp_path):
+    services, failed, _ = setup_run(tmp_path)
+    failed.status = "failed"
+    services.state.put_run(failed)
+    accept(services, request(), "pre-run")
+    second = next_run(services, 2)
+    assert served(services, second)["meter"]["value"] == "12345"
+    assert [row["confirmationId"] for row in correction_snapshot(services.state, second)] == ["pre-run"]
+    third = next_run(services, 3)
+    assert correction_snapshot(services.state, third) == []
+    assert not pending_corrections(services.state, failed)
+
+
+def test_retry_does_not_move_an_answer_to_a_newer_run(tmp_path):
+    services, first, _ = setup_run(tmp_path)
+    accept(services, request(), "stable-target")
+    second = next_run(services, 2)
+    accept(services, request(), "stable-target")
+    assert [row["confirmationId"] for row in correction_snapshot(services.state, first)] == ["stable-target"]
+    assert correction_snapshot(services.state, second) == []
diff --git a/backend/server/orchestrator/tests/test_correction_versions.py b/backend/server/orchestrator/tests/test_correction_versions.py
new file mode 100644
index 00000000..8f88d580
--- /dev/null
+++ b/backend/server/orchestrator/tests/test_correction_versions.py
@@ -0,0 +1,106 @@
+"""The immutable sequences remain safe without exclusive lease ownership."""
+
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor
+from threading import Barrier
+
+import pytest
+
+from server.common.backends.state_memory import MemoryState
+from server.common.backends.state_modal_dict import ModalDictState
+from server.common.tests.test_modal_dict_files import _FakeModalDict
+from server.orchestrator.corrections import (
+    correction_snapshot,
+    latest_publication,
+    pending_corrections,
+    publish_version,
+    record_correction,
+)
+from server.orchestrator.tests.conftest import build_services
+from server.orchestrator.tests.test_greptile_followup import _run
+
+
+@pytest.fixture(params=["memory", "modal", "modal-local"])
+def state(request):
+    if request.param == "memory":
+        return MemoryState()
+    if request.param == "modal-local":
+        return ModalDictState({})
+    return ModalDictState(_FakeModalDict())
+
+
+def correction(identifier, field="meter_number", value="12345"):
+    return {
+        "confirmationId": identifier,
+        "fieldId": field,
+        "kind": "corrected",
+        "value": value,
+        "imageId": "s1",
+        "by": "device",
+    }
+
+
+def test_competing_appenders_retry_the_same_empty_slot(state, monkeypatch):
+    gate = Barrier(2)
+    original = state.put_if_absent
+
+    def insert(key, value):
+        if key == "capture_correction:cap_test:1":
+            gate.wait(timeout=5)
+        return original(key, value)
+
+    monkeypatch.setattr(state, "put_if_absent", insert)
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        futures = [pool.submit(record_correction, state, "cap_test", correction(name)) for name in ("a", "b")]
+        assert sorted(f.result(timeout=5) for f in futures) == [1, 2]
+    assert {state.get_kv(f"capture_correction:cap_test:{n}")["confirmationId"] for n in (1, 2)} == {"a", "b"}
+    assert state.get_kv("capture_correction:cap_test:3") is None
+
+
+def test_replay_is_idempotent_and_reused_identity_fails(state):
+    row = correction("a")
+    assert record_correction(state, "cap_test", row) == 1
+    assert record_correction(state, "cap_test", row) == 1
+    with pytest.raises(ValueError, match="reused"):
+        record_correction(state, "cap_test", correction("a", value="67890"))
+
+
+def test_old_composer_cannot_append_after_newer_coverage(state):
+    newest = publish_version(state, "run_test", {"correctionIds": ["a", "b"], "artifacts": {"result": "new"}})
+    assert publish_version(state, "run_test", {"correctionIds": ["a"], "artifacts": {"result": "old"}}) == newest
+    assert latest_publication(state, "run_test") == newest
+    assert state.get_kv("result_version:run_test:1") == newest
+    assert state.get_kv("result_version:run_test:2") is None
+
+
+def test_equal_coverage_can_publish_changed_artifacts_without_overwriting(state):
+    first = publish_version(state, "run_test", {"correctionIds": [], "artifacts": {"result": "eligible"}})
+    changed = {"correctionIds": [], "artifacts": {"result": "views"}}
+    second = publish_version(state, "run_test", changed)
+    assert second["version"] == 2
+    assert state.get_kv("result_version:run_test:1") == first
+    assert publish_version(state, "run_test", changed) == second
+    assert state.get_kv("result_version:run_test:3") is None
+
+
+def test_crash_between_capture_and_run_record_is_recoverable(tmp_path, state, monkeypatch):
+    services, _, _ = build_services(tmp_path, "memory")
+    run = _run(services, "run_test")
+    state.put_run(run)
+    original = state.put_if_absent
+
+    def crash(key, value):
+        if key.startswith("run_correction:"):
+            raise OSError("process stopped before run copy")
+        return original(key, value)
+
+    monkeypatch.setattr(state, "put_if_absent", crash)
+    record_correction(state, run.capture_id, correction("a"))
+    with pytest.raises(OSError, match="process stopped"):
+        correction_snapshot(state, run)
+    monkeypatch.setattr(state, "put_if_absent", original)
+    assert pending_corrections(state, run)
+    assert correction_snapshot(state, run) == [{**correction("a"), "acceptedSequence": 1}]
+    publish_version(state, run.run_id, {"correctionIds": ["a"], "artifacts": {}})
+    assert not pending_corrections(state, run)
diff --git a/backend/server/orchestrator/tests/test_greptile_followup.py b/backend/server/orchestrator/tests/test_greptile_followup.py
index 3f047910..679c8947 100644
--- a/backend/server/orchestrator/tests/test_greptile_followup.py
+++ b/backend/server/orchestrator/tests/test_greptile_followup.py
@@ -20,13 +20,14 @@
 from server.orchestrator.artifacts import overlay_stage_artifacts
 from server.orchestrator.clock import Clock
 from server.orchestrator.coordinator import RunCoordinator
+from server.orchestrator.corrections import correction_snapshot, fold_reads, pending_corrections, record_correction
 from server.orchestrator.events import emit_failed
 from server.orchestrator.services import bind
 from server.orchestrator.stubs._common import scene_document
 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
+from server.reads.types import ConfirmBlock, MeterRead, PanelBundle, PanelRead, Reads
 
 
 def _run(services, run_id: str, **overrides: object) -> RunRecord:
@@ -153,7 +154,9 @@ def test_confirmation_updates_the_reads_record(tmp_path: Path) -> None:
     )
     with bind(services):
         on_confirmation("cap_test", request, "cfm_1", by="device")
-    stored = Reads.model_validate(json.loads(services.storage.get_bytes(key)))
+    base = Reads.model_validate_json(services.storage.get_bytes(key))
+    assert base.meter.value == "00000"
+    stored = fold_reads(base, correction_snapshot(services.state, run))
     assert stored.meter is not None
     assert stored.meter.value == "12345"
     assert stored.meter.status == "auto_fill"
@@ -215,7 +218,9 @@ def test_correction_on_a_legacy_run_without_scene_artifact_fails_loudly(tmp_path
     with bind(services):
         on_confirmation("cap_test", request, "cfm_3", by="device")
     # The answer itself is still persisted either way.
-    stored_reads = Reads.model_validate(json.loads(services.storage.get_bytes(key)))
+    stored_reads = fold_reads(
+        Reads.model_validate_json(services.storage.get_bytes(key)), correction_snapshot(services.state, run)
+    )
     assert stored_reads.meter is not None
     assert stored_reads.meter.value == "12345"
     # And the stale-result silence is gone: the run reports its failed rebuild.
@@ -300,8 +305,6 @@ def test_correction_during_live_result_stage_recomposes_once(tmp_path: Path, mon
 
     import server.orchestrator.stubs.result as result_stub
     from server.orchestrator.api import start_run
-    from server.orchestrator.coordinator import pending_correction_key
-    from server.orchestrator.stateio import get_json
 
     services, uri, digest = build_services(tmp_path, "memory", coordinator_loop_s=0.05)
 
@@ -363,10 +366,253 @@ def drive() -> None:
             if event.type == "stage" and event.data.get("stage") == stage and event.data.get("status") == "running"
         ]
         assert len(started) == 2, [(e.type, e.data) for e in started]
+    finals = [event for event in events if event.type in {"verdict_ready", "retake_request", "failed"}]
+    assert [event.type for event in finals] == ["verdict_ready", "verdict_ready"]
+    assert not any(event.data.get("code") == "result_no_event" for event in events if event.type == "failed")
     # The pending marker is consumed, not left for a later finish to re-reset.
-    assert get_json(services.state, pending_correction_key(run.run_id)) is None
+    assert not pending_corrections(services.state, run)
     # The served result carries the corrected value, not the pre-correction meter.
     served_run = overlay_stage_artifacts(services.state, run)
     served = json.loads(services.storage.get_bytes(served_run.artifacts["result/result.json"]))
     assert served["meter"]["value"] == "12345"
     assert services.state.get_capture("cap_test").status == "manual_review"
+
+
+def test_pending_correction_survives_a_failed_reset(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """A reset failure must leave the durable marker for the next tick."""
+
+    import server.orchestrator.coordinator as coordinator_module
+
+    services, _uri, _digest = build_services(tmp_path, "memory")
+    run = _run(services, "run_reset_failure")
+    services.state.put_run(run)
+    record_correction(services.state, run.capture_id, {"confirmationId": "a"})
+
+    def fail_reset(*_args, **_kwargs):
+        raise OSError("state write interrupted")
+
+    monkeypatch.setattr(coordinator_module, "reset_criteria_and_result", fail_reset)
+    with pytest.raises(OSError, match="state write interrupted"):
+        with bind(services):
+            RunCoordinator(services)._consume_pending_correction(run)
+    assert pending_corrections(services.state, run)
+
+
+def test_confirmation_during_completion_remains_pending(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """A confirmation racing completion is converted into a recompose, never a stale marker."""
+
+    import threading
+
+    import server.orchestrator.api as api_module
+
+    services, _uri, _digest = build_services(tmp_path, "memory")
+    reads_key = "captures/cap_test/runs/run_race/reads/reads.json"
+    reads = Reads(
+        capture_id="cap_test",
+        status="complete",
+        meter=MeterRead(status="auto_fill", value="00000", core="00000", confidence=0.5),
+        confirm=ConfirmBlock(candidates=["00000"]),
+    )
+    services.storage.put_bytes(reads_key, reads.model_dump_json().encode(), "application/json")
+    run = _run(
+        services,
+        "run_race",
+        artifacts={"reads/reads.json": reads_key},
+    )
+    services.state.put_run(run)
+    from server.orchestrator.graph import STAGE_NAMES
+
+    for stage in STAGE_NAMES:
+        services.state.upsert_stage(
+            StageRecord(run_id=run.run_id, stage=stage, status="done", attempt=1, heartbeat=services.clock.iso())
+        )
+    services.state.put_blob(
+        f"run_control:{run.run_id}",
+        b'{"merge_done":true,"coordinator_heartbeat_at":null,"coordinator_respawns":0}',
+    )
+
+    completion_entered = threading.Event()
+    confirmation_waiting = threading.Event()
+    recompose_called = threading.Event()
+
+    def write_files(_self, _run):
+        completion_entered.set()
+        assert confirmation_waiting.wait(5)
+
+    def apply_success(_self, current):
+        current.status = "done"
+        services.state.put_run(current)
+
+    monkeypatch.setattr(RunCoordinator, "_write_run_files", write_files)
+    monkeypatch.setattr(RunCoordinator, "_apply_success", apply_success)
+
+    monkeypatch.setattr(api_module, "_recompose_after_correction", lambda *_args: recompose_called.set())
+
+    errors: list[BaseException] = []
+
+    def finish() -> None:
+        try:
+            with bind(services):
+                RunCoordinator(services)._maybe_finish(run)
+        except BaseException as exc:
+            errors.append(exc)
+
+    finish_thread = threading.Thread(target=finish)
+    finish_thread.start()
+    assert completion_entered.wait(5)
+
+    request = ConfirmationRequest(
+        fieldId="meter_number",
+        kind="corrected",
+        value="12345",
+        imageId="s1",
+        clientAt="2026-10-15T00:00:00.000Z",
+    )
+
+    def confirm() -> None:
+        try:
+            with bind(services):
+                on_confirmation("cap_test", request, "cfm_race", by="device")
+                confirmation_waiting.set()
+        except BaseException as exc:
+            errors.append(exc)
+
+    confirm_thread = threading.Thread(target=confirm)
+    confirm_thread.start()
+    assert confirmation_waiting.wait(5)
+    finish_thread.join(5)
+    confirm_thread.join(5)
+    assert not finish_thread.is_alive() and not confirm_thread.is_alive()
+    assert not errors, errors
+    assert pending_corrections(services.state, run)
+    assert services.state.get_run(run.run_id).status == "running"
+
+
+def test_confirmation_busy_leaves_a_durable_record(tmp_path: Path) -> None:
+    """LeaseBusy cannot acknowledge a correction without recovery state."""
+    services, _uri, _digest = build_services(tmp_path, "memory")
+    reads_key = "captures/cap_test/runs/run_busy/reads/reads.json"
+    reads = Reads(
+        capture_id="cap_test",
+        status="complete",
+        meter=MeterRead(status="auto_fill", value="00000", core="00000", confidence=0.5),
+        confirm=ConfirmBlock(candidates=["00000"]),
+    )
+    services.storage.put_bytes(reads_key, reads.model_dump_json().encode(), "application/json")
+    run = _run(services, "run_busy", status="done", artifacts={"reads/reads.json": reads_key})
+    services.state.put_run(run)
+    services.state.put_if_absent("lease:correction:run_busy", {"id": "other", "at": services.clock.now().timestamp()})
+    request = ConfirmationRequest(
+        fieldId="meter_number",
+        kind="corrected",
+        value="12345",
+        imageId="s1",
+        clientAt="2026-10-15T00:00:00.000Z",
+    )
+    with bind(services):
+        on_confirmation("cap_test", request, "cfm_busy", by="device")
+    assert pending_corrections(services.state, run)
+    assert correction_snapshot(services.state, run)[0]["value"] == "12345"
+    stored = Reads.model_validate(json.loads(services.storage.get_bytes(reads_key)))
+    assert stored.meter.value == "00000"
+
+
+def test_crash_after_acceptance_keeps_correction_for_recovery(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """There is no shared reads-write/marker-write gap to lose an answer."""
+    import server.orchestrator.api as api_module
+
+    services, _uri, _digest = build_services(tmp_path, "memory")
+    reads_key = "captures/cap_test/runs/run_crash/reads/reads.json"
+    reads = Reads(
+        capture_id="cap_test",
+        status="complete",
+        meter=MeterRead(status="auto_fill", value="00000", core="00000", confidence=0.5),
+        confirm=ConfirmBlock(candidates=["00000"]),
+    )
+    services.storage.put_bytes(reads_key, reads.model_dump_json().encode(), "application/json")
+    run = _run(services, "run_crash", status="done", artifacts={"reads/reads.json": reads_key})
+    services.state.put_run(run)
+
+    def crash(*_args):
+        raise OSError("crash after durable acceptance")
+
+    monkeypatch.setattr(api_module, "_recompose_after_correction", crash)
+    request = ConfirmationRequest(
+        fieldId="meter_number",
+        kind="corrected",
+        value="12345",
+        imageId="s1",
+        clientAt="2026-10-15T00:00:00.000Z",
+    )
+    with pytest.raises(OSError, match="crash after durable acceptance"):
+        with bind(services):
+            on_confirmation("cap_test", request, "cfm_crash", by="device")
+    stored = Reads.model_validate(json.loads(services.storage.get_bytes(reads_key)))
+    assert stored.meter.value == "00000"
+    assert pending_corrections(services.state, run)
+    effective = fold_reads(stored, correction_snapshot(services.state, run))
+    assert effective.meter.value == "12345"
+
+
+def test_concurrent_corrections_merge_different_fields(tmp_path: Path, monkeypatch) -> None:
+    """Both writers observe one empty append slot; neither answer is lost."""
+    import threading
+
+    services, _uri, _digest = build_services(tmp_path, "memory")
+    reads_key = "captures/cap_test/runs/run_fields/reads/reads.json"
+    reads = Reads(
+        capture_id="cap_test",
+        status="complete",
+        meter=MeterRead(status="auto_fill", value="00000", core="00000", confidence=0.5),
+        panel=PanelBundle(merged=PanelRead(brand="old", main_breaker_amps=100)),
+        confirm=ConfirmBlock(candidates=["00000"]),
+    )
+    services.storage.put_bytes(reads_key, reads.model_dump_json().encode(), "application/json")
+    services.state.put_run(_run(services, "run_fields", artifacts={"reads/reads.json": reads_key}))
+    requests = [
+        ConfirmationRequest(
+            fieldId="meter_number",
+            kind="corrected",
+            value="12345",
+            imageId="s1",
+            clientAt="2026-10-15T00:00:00.000Z",
+        ),
+        ConfirmationRequest(
+            fieldId="panel_brand",
+            kind="corrected",
+            value="new",
+            imageId="s1",
+            clientAt="2026-10-15T00:00:00.000Z",
+        ),
+    ]
+    barrier = threading.Barrier(2)
+    original_insert = services.state.put_if_absent
+
+    def insert(key, row):
+        if key == "capture_correction:cap_test:1":
+            barrier.wait(timeout=5)
+        return original_insert(key, row)
+
+    monkeypatch.setattr(services.state, "put_if_absent", insert)
+    errors: list[BaseException] = []
+
+    def submit(index: int) -> None:
+        try:
+            with bind(services):
+                on_confirmation("cap_test", requests[index], f"cfm_field_{index}", by="device")
+        except BaseException as exc:
+            errors.append(exc)
+
+    threads = [threading.Thread(target=submit, args=(index,)) for index in range(2)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join(10)
+        assert not thread.is_alive()
+    assert errors == []
+    stored = Reads.model_validate(json.loads(services.storage.get_bytes(reads_key)))
+    assert stored.meter.value == "00000"
+    run = services.state.get_run("run_fields")
+    effective = fold_reads(stored, correction_snapshot(services.state, run))
+    assert effective.meter.value == "12345"
+    assert effective.panel.merged.brand == "new"
diff --git a/backend/server/orchestrator/tests/test_immutable_correction_races.py b/backend/server/orchestrator/tests/test_immutable_correction_races.py
new file mode 100644
index 00000000..b9eda602
--- /dev/null
+++ b/backend/server/orchestrator/tests/test_immutable_correction_races.py
@@ -0,0 +1,279 @@
+"""Force races at active journal, lease, and publication boundaries."""
+
+from __future__ import annotations
+
+import json
+from concurrent.futures import ThreadPoolExecutor
+from threading import Barrier, Event, Thread
+
+import pytest
+
+from server.api.hooks import safe_hook
+from server.api.routes.result import _artifacts
+from server.common.lease import LeaseBusy, with_lease
+from server.contracts.api import ConfirmationRequest
+from server.contracts.jobs import StageRecord
+from server.orchestrator.api import on_confirmation, start_run
+from server.orchestrator.artifacts import overlay_stage_artifacts
+from server.orchestrator.coordinator import RunCoordinator
+from server.orchestrator.corrections import correction_snapshot, pending_corrections
+from server.orchestrator.runner import run_stage
+from server.orchestrator.services import bind
+from server.orchestrator.stateio import put_json
+from server.orchestrator.sweeper import sweep
+from server.orchestrator.tests.conftest import build_services
+from server.reads.types import ConfirmBlock, MeterRead, PanelBundle, PanelRead, Reads
+
+
+def setup_run(tmp_path):
+    services, uri, digest = build_services(tmp_path, "memory")
+    with bind(services):
+        run = start_run("cap_test", digest, uri)
+    run = services.state.get_run(run.run_id)
+    assert run.status == "done"
+    key = overlay_stage_artifacts(services.state, run).artifacts["reads/reads.json"]
+    base = Reads(
+        capture_id=run.capture_id,
+        status="complete",
+        meter=MeterRead(status="auto_fill", value="00000", core="00000", confidence=0.5),
+        panel=PanelBundle(merged=PanelRead(brand="old", main_breaker_amps=100)),
+        confirm=ConfirmBlock(candidates=["00000"]),
+    )
+    services.storage.put_bytes(key, base.model_dump_json().encode(), "application/json")
+    run.status = "running"
+    services.state.put_run(run)
+    return services, run, key
+
+
+def request(field="meter_number", value="12345"):
+    return ConfirmationRequest(
+        fieldId=field, kind="corrected", value=value, imageId="k0", clientAt="2026-10-15T00:00:00.000Z"
+    )
+
+
+def accept(services, row, identifier):
+    with bind(services):
+        on_confirmation("cap_test", row, identifier, by="device")
+
+
+def served(services, run):
+    artifacts = _artifacts(services, run)
+    return json.loads(services.storage.get_bytes(artifacts["result/result.json"]))
+
+
+def compose_context(services, run, stage, label):
+    rec = StageRecord(run_id=run.run_id, stage=stage, status="done", attempt=1, heartbeat=services.clock.iso())
+    # Execute both callbacks as an already admitted worker. Worker-ID checks are
+    # advisory too and are not the invariant under test here.
+    services.state.upsert_stage(rec)
+    return services.stage_context(run, rec, label)
+
+
+def prepare_result(services, run, label):
+    criteria = compose_context(services, run, "criteria", label)
+    output = run_stage(criteria, stubs=True)
+    put_json(services.state, f"stage_artifacts:{run.run_id}:criteria", output.artifacts)
+    return compose_context(services, run, "result", label)
+
+
+def test_busy_acceptance_is_recovered_after_completion(tmp_path, monkeypatch):
+    import server.orchestrator.api as api
+
+    services, run, _ = setup_run(tmp_path)
+    run.status = "done"
+    services.state.put_run(run)
+    lease_attempts = []
+
+    def busy(*_args):
+        lease_attempts.append(run.run_id)
+        assert [row["confirmationId"] for row in correction_snapshot(services.state, run)] == ["busy"]
+        assert pending_corrections(services.state, run)
+        raise LeaseBusy("deterministically occupied")
+
+    with monkeypatch.context() as patch:
+        patch.setattr(api, "with_correction_lease", busy)
+        with bind(services):
+            safe_hook("on_confirmation", on_confirmation, run.capture_id, request(), "busy", by="device", strict=True)
+    assert lease_attempts == [run.run_id], "the done-run recomposition lease was never attempted"
+    assert pending_corrections(services.state, run)
+    sweep(services)
+    assert served(services, run)["meter"]["value"] == "12345"
+    assert not pending_corrections(services.state, run)
+
+
+@pytest.mark.parametrize("after_insert", [False, True])
+def test_crash_at_journal_insert_never_acknowledges_an_unstored_answer(tmp_path, monkeypatch, after_insert):
+    services, run, reads_key = setup_run(tmp_path)
+    original = services.state.put_if_absent
+    faults = []
+    acknowledged = []
+
+    def crash(key, row):
+        if key.startswith("capture_correction:"):
+            faults.append(key)
+            if after_insert:
+                assert original(key, row)
+            raise OSError("process crashed at journal insert")
+        return original(key, row)
+
+    with monkeypatch.context() as patch:
+        patch.setattr(services.state, "put_if_absent", crash)
+        with pytest.raises(OSError, match="journal insert"), bind(services):
+            safe_hook("on_confirmation", on_confirmation, run.capture_id, request(), "crash", by="device", strict=True)
+            acknowledged.append(True)
+    assert faults == [f"capture_correction:{run.capture_id}:1"]
+    assert acknowledged == []
+    assert bool(services.state.get_kv(f"capture_correction:{run.capture_id}:1")) is after_insert
+    if after_insert:
+        # Recovery does not require the unacknowledged client to retry.
+        run.status = "done"
+        services.state.put_run(run)
+        sweep(services)
+        assert served(services, run)["meter"]["value"] == "12345"
+    accept(services, request(), "crash")
+    run.status = "done"
+    services.state.put_run(run)
+    sweep(services)
+    assert served(services, run)["meter"]["value"] == "12345"
+    assert [row["confirmationId"] for row in correction_snapshot(services.state, run)] == ["crash"]
+    assert Reads.model_validate_json(services.storage.get_bytes(reads_key)).meter.value == "00000"
+
+
+def test_two_workers_cannot_lose_different_field_corrections(tmp_path, monkeypatch):
+    import server.orchestrator.stubs.criteria as criteria
+
+    services, run, _ = setup_run(tmp_path)
+    original_insert = services.state.put_if_absent
+    same_empty_slot = Barrier(2)
+    attempts = []
+    results = []
+
+    def contend_for_journal_slot(key, row):
+        if key == f"capture_correction:{run.capture_id}:1":
+            attempts.append(row["confirmationId"])
+            same_empty_slot.wait(timeout=5)
+            inserted = original_insert(key, row)
+            results.append(inserted)
+            return inserted
+        return original_insert(key, row)
+
+    with monkeypatch.context() as patch:
+        patch.setattr(services.state, "put_if_absent", contend_for_journal_slot)
+        with ThreadPoolExecutor(max_workers=2) as pool:
+            futures = [
+                pool.submit(accept, services, row, identifier)
+                for row, identifier in ((request(), "meter"), (request("panel_brand", "new"), "panel"))
+            ]
+            for future in futures:
+                future.result(timeout=10)
+    assert sorted(attempts) == ["meter", "panel"], "both workers must reach the immutable insertion boundary"
+    assert sorted(results) == [False, True], "one insertion must lose contention and retry"
+    assert {row["confirmationId"] for row in correction_snapshot(services.state, run)} == {"meter", "panel"}
+    hydrated = []
+    original_criteria = criteria.run
+
+    def observe(ctx):
+        hydrated.append(Reads.model_validate_json((ctx.work_dir / "reads/reads.json").read_bytes()))
+        return original_criteria(ctx)
+
+    monkeypatch.setattr(criteria, "run", observe)
+    RunCoordinator(services).drive(run.run_id)
+    assert hydrated
+    assert hydrated[-1].meter.value == "12345"
+    assert hydrated[-1].panel.merged.brand == "new"
+    assert served(services, run)["meter"]["value"] == "12345"
+
+
+def test_stale_release_and_two_composers_preserve_every_published_version(tmp_path, monkeypatch):
+    services, run, _ = setup_run(tmp_path)
+    accept(services, request(), "meter")
+    stale_context = prepare_result(services, run, "stale")
+    successor_entered = Event()
+    successor_deleted = Event()
+    errors = []
+    actual_delete = services.state.delete_kv
+    key = "lease:publication-test"
+    published_bytes = {}
+    actual_write = services.storage.put_bytes
+
+    def immutable_write(storage_key, data, content_type):
+        if "/result/" in storage_key:
+            assert storage_key not in published_bytes, "a result object was overwritten"
+            published_bytes[storage_key] = data
+        actual_write(storage_key, data, content_type)
+
+    monkeypatch.setattr(services.storage, "put_bytes", immutable_write)
+
+    def successor():
+        def compose():
+            successor_entered.set()
+            assert successor_deleted.wait(5)
+            accept(services, request("panel_brand", "new"), "panel")
+            run_stage(prepare_result(services, run, "successor"), stubs=True)
+
+        try:
+            with_lease(services.state, "publication-test", compose, strict=True, wait_s=0)
+        except RuntimeError as exc:
+            # The advisory helper may report the deliberately deleted lease.
+            if type(exc).__name__ != "LeaseLost":
+                errors.append(exc)
+        except BaseException as exc:
+            errors.append(exc)
+
+    worker = Thread(target=successor)
+    deleted_successor = []
+
+    def release_after_takeover(lease_key):
+        if lease_key == key and not deleted_successor:
+            # Release has already read the old token. Expiry removes it and
+            # lets a successor claim before the delayed delete reaches storage.
+            actual_delete(key)
+            worker.start()
+            assert successor_entered.wait(5)
+            assert services.state.get_kv(key) is not None
+            deleted_successor.append(True)
+            actual_delete(key)
+            successor_deleted.set()
+            return
+        actual_delete(lease_key)
+
+    monkeypatch.setattr(services.state, "delete_kv", release_after_takeover)
+    with_lease(services.state, "publication-test", lambda: None, strict=True, wait_s=0)
+    worker.join(10)
+    assert not worker.is_alive()
+    assert errors == []
+    assert deleted_successor == [True]
+    before = {key: services.storage.get_bytes(key) for key in published_bytes}
+    # The stale composer resumes after its successor has published both fields.
+    run_stage(stale_context, stubs=True)
+    assert served(services, run)["meter"]["value"] == "12345"
+    selected = _artifacts(services, run)
+    snapshot_key = selected.get("verdicts/correction-inputs.json")
+    assert snapshot_key is not None, "the served result has no correction coverage record"
+    snapshot = json.loads(services.storage.get_bytes(snapshot_key))
+    assert snapshot["reads"]["panel"]["merged"]["brand"] == "new"
+    assert snapshot["correctionIds"] == ["meter", "panel"]
+    assert all(services.storage.get_bytes(key) == data for key, data in before.items())
+
+
+def test_crash_after_upload_before_publication_keeps_answer_pending(tmp_path, monkeypatch):
+    services, run, _ = setup_run(tmp_path)
+    accept(services, request(), "crash")
+    context = prepare_result(services, run, "interrupted")
+    original = services.state.put_if_absent
+    events_before = len(services.state.events_after(run.capture_id, 0, limit=500))
+
+    def crash(key, row):
+        if key.startswith("result_version:"):
+            raise OSError("crash before atomic publication")
+        return original(key, row)
+
+    with monkeypatch.context() as patch:
+        patch.setattr(services.state, "put_if_absent", crash)
+        with pytest.raises(OSError, match="atomic publication"):
+            run_stage(context, stubs=True)
+    assert len(services.state.events_after(run.capture_id, 0, limit=500)) == events_before
+    RunCoordinator(services).drive(run.run_id)
+    assert served(services, run)["meter"]["value"] == "12345"
+    events = services.state.events_after(run.capture_id, 0, limit=500)
+    assert any(event.type == "verdict_ready" for event in events[events_before:])
diff --git a/backend/server/orchestrator/tests/test_legacy_correction_import.py b/backend/server/orchestrator/tests/test_legacy_correction_import.py
new file mode 100644
index 00000000..0b0a32f9
--- /dev/null
+++ b/backend/server/orchestrator/tests/test_legacy_correction_import.py
@@ -0,0 +1,110 @@
+"""Old API workers' acknowledged slots survive immutable reads publication."""
+
+from __future__ import annotations
+
+import pytest
+
+from server.orchestrator.api import start_run
+from server.orchestrator.corrections import correction_snapshot, latest_publication, pending_corrections
+from server.orchestrator.runner import run_stage
+from server.orchestrator.services import bind
+from server.orchestrator.stateio import put_json
+from server.orchestrator.tests.conftest import build_services
+from server.orchestrator.tests.test_immutable_correction_races import (
+    accept,
+    compose_context,
+    prepare_result,
+    request,
+    served,
+    setup_run,
+)
+from server.reads.types import Reads
+
+
+def legacy_slot(services, identifier="legacy-answer"):
+    services.state.put_kv("confirmix:cap_test", {"fields": ["meter_number"]})
+    services.state.put_kv(
+        "confirm:cap_test:meter_number",
+        {
+            "confirmationId": identifier,
+            "fieldId": "meter_number",
+            "imageId": "k0",
+            "kind": "corrected",
+            "value": "12345",
+            "by": "device",
+            "seq": 1,
+            "receivedAt": "2026-10-15T00:00:00Z",
+        },
+    )
+
+
+@pytest.mark.parametrize("real_reads", [False, True])
+def test_legacy_answer_at_stage_start_reaches_result_exactly_once(tmp_path, monkeypatch, real_reads):
+    import server.orchestrator.runner as runner
+    import server.reads as reads
+
+    services, run, base_key = setup_run(tmp_path)
+    base = Reads.model_validate_json(services.storage.get_bytes(base_key))
+    legacy_slot(services)
+    original = runner.stage_callable
+    if real_reads:
+        monkeypatch.setattr(reads, "build_reads", lambda *_args: base.model_copy(deep=True))
+        monkeypatch.setattr(
+            runner, "stage_callable", lambda stage, **kw: reads.run if stage == "reads" else original(stage, **kw)
+        )
+    for attempt in ("first", "retry"):
+        output = run_stage(compose_context(services, run, "reads", attempt), stubs=True)
+        put_json(services.state, f"stage_artifacts:{run.run_id}:reads", output.artifacts)
+    run_stage(prepare_result(services, run, "result"), stubs=True)
+    assert served(services, run)["meter"]["value"] == "12345"
+    assert latest_publication(services.state, run.run_id)["correctionIds"] == ["legacy-answer"]
+    assert len(correction_snapshot(services.state, run)) == 1
+    assert services.state.get_kv("capture_correction:cap_test:2") is None
+
+
+def test_legacy_answer_before_run_creation_is_adopted(tmp_path):
+    services, uri, digest = build_services(tmp_path, "memory")
+    legacy_slot(services)
+    with bind(services):
+        run = start_run("cap_test", digest, uri)
+    assert served(services, run)["meter"]["value"] == "12345"
+    assert latest_publication(services.state, run.run_id)["correctionIds"] == ["legacy-answer"]
+    assert not pending_corrections(services.state, run)
+
+
+def test_interrupted_legacy_import_is_recovered_without_duplicate(tmp_path, monkeypatch):
+    services, run, _ = setup_run(tmp_path)
+    legacy_slot(services)
+    original = services.state.put_if_absent
+    claimed = []
+
+    def crash(key, row):
+        if key.startswith("claim:"):
+            claimed.append(key)
+            raise OSError("crashed before legacy assignment")
+        return original(key, row)
+
+    with monkeypatch.context() as patch:
+        patch.setattr(services.state, "put_if_absent", crash)
+        with pytest.raises(OSError, match="legacy assignment"):
+            run_stage(compose_context(services, run, "reads", "crash"), stubs=True)
+    assert claimed
+    run_stage(compose_context(services, run, "reads", "retry"), stubs=True)
+    run_stage(prepare_result(services, run, "result"), stubs=True)
+    assert served(services, run)["meter"]["value"] == "12345"
+    assert latest_publication(services.state, run.run_id)["correctionIds"] == ["legacy-answer"]
+    assert services.state.get_kv("capture_correction:cap_test:2") is None
+
+
+def test_import_does_not_retarget_a_modern_confirmation(tmp_path):
+    services, run, _ = setup_run(tmp_path)
+    accept(services, request(), "modern-answer")
+    legacy_slot(services, "modern-answer")
+    run_stage(compose_context(services, run, "reads", "import"), stubs=True)
+    run_stage(prepare_result(services, run, "result"), stubs=True)
+    assert latest_publication(services.state, run.run_id)["correctionIds"] == ["modern-answer"]
+    accepted = services.state.get_kv("capture_correction:cap_test:1")
+    assert "targetRunId" not in accepted
+    assert services.state.get_kv("claim:cap_test:modern-answer:1")["runId"] == run.run_id
+    assert not accepted.get("legacy")
+    assert services.state.get_kv("capture_correction:cap_test:2") is None
diff --git a/backend/server/orchestrator/tests/test_publication_events.py b/backend/server/orchestrator/tests/test_publication_events.py
new file mode 100644
index 00000000..7f8e0143
--- /dev/null
+++ b/backend/server/orchestrator/tests/test_publication_events.py
@@ -0,0 +1,83 @@
+"""Publication crash recovery preserves the existing public event shape."""
+
+from __future__ import annotations
+
+import asyncio
+from types import SimpleNamespace
+
+import pytest
+
+from server.api.events import poll_events
+from server.orchestrator.corrections import latest_publication, pending_corrections, publish_version
+from server.orchestrator.publication import emit_latest_publication
+from server.orchestrator.runner import run_stage
+from server.orchestrator.sweeper import sweep
+from server.orchestrator.tests.test_immutable_correction_races import (
+    accept,
+    prepare_result,
+    request,
+    served,
+    setup_run,
+)
+
+
+def test_crash_after_publication_before_final_event_is_recovered(tmp_path, monkeypatch):
+    import server.orchestrator.publication as publication
+
+    services, run, _ = setup_run(tmp_path)
+    accept(services, request(), "answer")
+    context = prepare_result(services, run, "published-no-event")
+    previous = services.state.events_after(run.capture_id, 0, limit=500)
+
+    def crash(*_args):
+        raise OSError("crash before final event")
+
+    with monkeypatch.context() as patch:
+        patch.setattr(publication, "emit_latest_publication", crash)
+        with pytest.raises(OSError, match="before final event"):
+            run_stage(context, stubs=True)
+    assert not pending_corrections(services.state, run)
+    assert served(services, run)["meter"]["value"] == "12345"
+    assert services.state.events_after(run.capture_id, 0, limit=500) == previous
+    run.status = "done"
+    services.state.put_run(run)
+    sweep(services)
+    finals = services.state.events_after(run.capture_id, previous[-1].seq, limit=500)
+    assert [event.type for event in finals] == ["verdict_ready"]
+    sweep(services)
+    assert services.state.events_after(run.capture_id, previous[-1].seq, limit=500) == finals
+
+
+def test_late_old_event_and_duplicate_new_event_are_not_exposed(tmp_path, monkeypatch):
+    services, run, _ = setup_run(tmp_path)
+    old = latest_publication(services.state, run.run_id)
+    after = services.state.events_after(run.capture_id, 0, limit=500)[-1].seq
+    new = publish_version(services.state, run.run_id, {**old, "correctionIds": ["a"]})
+    emit_latest_publication(services.state, services.storage, run.run_id, services.settings)
+    # Model two emitters racing the event-log read and an older emitter whose
+    # append reaches the store only after the newer publication's final event.
+    for version in (new, old):
+        event_type, data = version["events"][0]
+        services.state.append_event(
+            run.capture_id,
+            event_type,
+            {**data, "_resultVersion": version["version"]},
+            run_id=run.run_id,
+            stage="result",
+        )
+    deps = SimpleNamespace(state=services.state, polls={}, wake={})
+    capture = services.state.get_capture(run.capture_id)
+    response = asyncio.run(poll_events(deps, capture, after, 0))
+    assert len(response.events) == 1
+    assert response.events[0].data == new["events"][0][1]
+    assert "_resultVersion" not in response.model_dump_json()
+    assert response.next == services.state.events_after(run.capture_id, 0, limit=500)[-1].seq
+
+
+def test_replayed_answer_may_have_a_new_client_timestamp(tmp_path):
+    services, run, _ = setup_run(tmp_path)
+    first = request()
+    accept(services, first, "retry")
+    accept(services, first.model_copy(update={"clientAt": "2026-10-16T00:00:00.000Z"}), "retry")
+    assert services.state.get_kv(f"capture_correction:{run.capture_id}:1")["confirmationId"] == "retry"
+    assert services.state.get_kv(f"capture_correction:{run.capture_id}:2") is None
diff --git a/backend/server/orchestrator/tests/test_round2.py b/backend/server/orchestrator/tests/test_round2.py
index c2abd81b..e8edef71 100644
--- a/backend/server/orchestrator/tests/test_round2.py
+++ b/backend/server/orchestrator/tests/test_round2.py
@@ -301,25 +301,13 @@ def run(ctx: StageContext):
 
 
 def _views_result(ctx: StageContext, *, both: bool) -> StageResult:
-    ctx.state.set_status(ctx.capture_id, "needs_views")
-    ctx.state.append_event(
-        ctx.capture_id,
-        "verdict_ready" if both else "retake_request",
-        {"runId": ctx.run_id, "viewsNeeded": ["ground"], "memberActions": []}
-        if not both
-        else {"runId": ctx.run_id, "kind": "needs_more_photos"},
-        run_id=ctx.run_id,
-        stage="result",
-    )
+    events = [["retake_request", {"runId": ctx.run_id, "viewsNeeded": ["ground"], "memberActions": []}]]
     if both:
-        ctx.state.append_event(
-            ctx.capture_id,
-            "retake_request",
-            {"runId": ctx.run_id, "viewsNeeded": ["ground"], "memberActions": []},
-            run_id=ctx.run_id,
-            stage="result",
-        )
+        events.insert(0, ["verdict_ready", {"runId": ctx.run_id, "kind": "needs_more_photos"}])
     body = {"status": "needs_views", "outcome": {"kind": "needs_more_photos"}}
+    intent = {"captureId": ctx.capture_id, "events": events, "status": "needs_views", "deliverWebhook": False}
+    (ctx.work_dir / "result").mkdir(parents=True, exist_ok=True)
+    (ctx.work_dir / "result/publication.json").write_bytes(json_bytes(intent))
     return write_files(ctx, {"result/result.json": json_bytes(body)})
 
 
@@ -359,9 +347,10 @@ def test_no_meter_branch_rejects_two_final_events(tmp_path, monkeypatch) -> None
         start_run("cap_test", digest, uri)
     events = services.state.events_after("cap_test", 0, limit=500)
     finals = [event.type for event in events if event.type in _FINALS]
-    assert finals == ["verdict_ready", "retake_request", "failed"]
+    # Reject the invalid intent before either final event can be emitted.
+    assert finals == ["failed"]
     failed = next(event for event in events if event.type == "failed")
-    assert failed.data["code"] == "result_no_event"
+    assert failed.data["code"] == "stage_exhausted:result:result_no_event"
     assert services.state.get_capture("cap_test").status == "manual_review"
 
 
diff --git a/backend/server/orchestrator/tests/test_run.py b/backend/server/orchestrator/tests/test_run.py
index 01395c68..9cb9db3f 100644
--- a/backend/server/orchestrator/tests/test_run.py
+++ b/backend/server/orchestrator/tests/test_run.py
@@ -166,8 +166,10 @@ def publish_door(ctx):
     stored = _object_ids(services.storage.get_bytes(run.artifacts["scene/scene.json"]))
     assert stored == ["meter", "synthetic-door"]
     for stage in ("criteria", "result"):
-        path = services.work_root / run.run_id / stage / "attempt-1" / "scene" / "scene.json"
-        assert _object_ids(path.read_bytes()) == stored
+        attempt = services.work_root / run.run_id / stage / "attempt-1"
+        paths = list(attempt.glob("*/scene/scene.json"))
+        assert len(paths) == 1
+        assert _object_ids(paths[0].read_bytes()) == stored
 
 
 def test_inline_stages_read_the_packet_once(tmp_path) -> None:
diff --git a/backend/server/orchestrator/tests/test_workdir.py b/backend/server/orchestrator/tests/test_workdir.py
index 57bb227d..88dc0637 100644
--- a/backend/server/orchestrator/tests/test_workdir.py
+++ b/backend/server/orchestrator/tests/test_workdir.py
@@ -51,7 +51,8 @@ def test_packet_dir_and_prior_artifacts_are_on_disk(tmp_path: Path) -> None:
     services.storage.put_bytes("captures/cap_test/keyframes/k0.jpg", b"jpeg", "image/jpeg")
     with bind(services):
         run = start_run("cap_test", packet_sha256(raw), uri)
-    work = services.work_root / run.run_id / "result" / "attempt-1"
+    attempt = services.work_root / run.run_id / "result" / "attempt-1"
+    (work,) = list(attempt.iterdir())
     packet_dir = work / "packet"
     assert (packet_dir / "packet.json").read_bytes() == raw
     # Result reads scene and verdicts, not the capture frames or the pose model.
diff --git a/backend/server/reads/__init__.py b/backend/server/reads/__init__.py
index 0e811c38..8aa85fe1 100644
--- a/backend/server/reads/__init__.py
+++ b/backend/server/reads/__init__.py
@@ -25,14 +25,16 @@ 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:
+    # Orchestrated reads are the base, not a mutable confirmation accumulator.
+    # Criteria folds the immutable correction sequence onto its private copy.
+    if ctx.inputs.get("immutable_corrections"):
+        from server.orchestrator.corrections import import_legacy_corrections
+
+        import_legacy_corrections(ctx.state, ctx.capture_id, ctx.state.get_run(ctx.run_id))
+    else:
         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/result/compose.py b/backend/server/result/compose.py
index 7b403263..618c1180 100644
--- a/backend/server/result/compose.py
+++ b/backend/server/result/compose.py
@@ -324,9 +324,24 @@ def run(ctx: StageContext) -> StageResult:
         )
     except ResultError:
         raise
-    seqs = emit_events(composed, ctx.state)
-    ctx.state.set_status(ctx.capture_id, composed.response.status)
-    _deliver(composed, capture, config)
+    seqs = []
+    if ctx.inputs.get("correction_snapshot"):
+        # The runner uploads immutable artifacts and publishes their version
+        # before it emits this intent. A final event must not precede its bytes.
+        (out_dir / "publication.json").write_text(
+            json.dumps(
+                {
+                    "captureId": ctx.capture_id,
+                    "events": composed.events,
+                    "status": composed.response.status,
+                    "deliverWebhook": True,
+                }
+            )
+        )
+    else:
+        seqs = emit_events(composed, ctx.state)
+        ctx.state.set_status(ctx.capture_id, composed.response.status)
+        _deliver(composed, capture, config)
     artifacts = {
         "result/result.json": str(out_dir / "result.json"),
         "result/result.internal.json": str(out_dir / "result.internal.json"),
diff --git a/backend/tests/test_capture_real_downstream.py b/backend/tests/test_capture_real_downstream.py
index ee1ca9e9..ac0b0cd5 100644
--- a/backend/tests/test_capture_real_downstream.py
+++ b/backend/tests/test_capture_real_downstream.py
@@ -155,12 +155,11 @@ def test_replacing_the_case_scene_fails_the_mode(tmp_path, monkeypatch):
 
 
 def test_dropping_the_published_result_fails_the_mode(tmp_path, monkeypatch):
-    """A real result stage that does not publish result/result.json is caught by
-    the required-published-asset check."""
+    """Reject a missing result before committing an immutable publication."""
     monkeypatch.setattr(runner, "_DROP_RESULT_ARTIFACT", True)
     receipt = runner.run_real_downstream(tmp_path / "run", case_ids=["flatwall-clear"])
     assert receipt["ok"] is False
-    assert any("result/result.json" in blocker for blocker in receipt["blockers"])
+    assert any("artifact_missing" in blocker for blocker in receipt["blockers"]), receipt["blockers"]
 
 
 def test_the_selection_override_and_boundaries_are_restored(tmp_path):
diff --git a/backend/tests/test_criteria_published_inputs.py b/backend/tests/test_criteria_published_inputs.py
index 11ba55db..e913ca41 100644
--- a/backend/tests/test_criteria_published_inputs.py
+++ b/backend/tests/test_criteria_published_inputs.py
@@ -180,7 +180,8 @@ def _hydrated_context(services: Services, capture_id: str, run_id: str, packet:
     rec = StageRecord(run_id=run_id, stage="criteria", status="running", attempt=1, heartbeat=services.clock.iso())
     services.state.upsert_stage(rec)
     ctx = services.stage_context(run, rec, "call-burst9")
-    assert set(ctx.inputs) == {"packet_dir", "run_control"}  # inputs ride the work dir, not ctx.inputs
+    assert set(ctx.inputs) == {"packet_dir", "run_control", "correction_snapshot"}
+    assert Path(ctx.inputs["correction_snapshot"]).is_file()
     return ctx