diff --git a/backend/server/api/finalize.py b/backend/server/api/finalize.py
index c0ea6781..0616231b 100644
--- a/backend/server/api/finalize.py
+++ b/backend/server/api/finalize.py
@@ -510,16 +510,33 @@ def _guard(deps: Deps, capture_id: str, future: Future[None]) -> None:
         # publishes its success under the same lock, so this older worker can
         # never reset a healed retry or emit a contradictory terminal event after
         # it.
-        with deps.locks.finish_sync(capture_id):
-            if not extras(deps, capture_id).get("run_started"):
-                failure = (
-                    exc
-                    if isinstance(exc, ApiFailure)
-                    else fail(503, "finalize_failed", "/", f"finalize background work failed ({type(exc).__name__})")
-                )
-                _on_background_failure(deps, capture_id, failure)
-        _clear_slot(capture_id)
-        future.set_exception(exc)
+        # The publication write can itself fail — a transient state outage makes
+        # put_capture raise. The thread must still release the worker slot and
+        # settle its Future, or later finalize retries answer "processing"
+        # forever with no worker ever starting again, and the drain waits on a
+        # Future nobody settles. The failure write stays under the lock; the
+        # slot release and the settlement always run.
+        try:
+            with deps.locks.finish_sync(capture_id):
+                if not extras(deps, capture_id).get("run_started"):
+                    failure = (
+                        exc
+                        if isinstance(exc, ApiFailure)
+                        else fail(
+                            503, "finalize_failed", "/", f"finalize background work failed ({type(exc).__name__})"
+                        )
+                    )
+                    _on_background_failure(deps, capture_id, failure)
+        except Exception as publication_error:
+            log_event(
+                "finalize.failure_publish_error",
+                capture_id=capture_id,
+                error_code=type(publication_error).__name__,
+                level="error",
+            )
+        finally:
+            _clear_slot(capture_id)
+            future.set_exception(exc)
         return
     future.set_result(None)
 
diff --git a/backend/server/api/tests/conftest.py b/backend/server/api/tests/conftest.py
index 1c0b454e..9da3d59e 100644
--- a/backend/server/api/tests/conftest.py
+++ b/backend/server/api/tests/conftest.py
@@ -4,6 +4,7 @@
 
 import base64
 import hashlib
+from pathlib import Path
 
 import pytest
 from fastapi.testclient import TestClient
@@ -16,6 +17,44 @@
 OTHER = "e" * 40
 REVIEWER = "f" * 40
 
+# The documented dev handoff (SD-73): hooks on, stage stubs on, no external calls.
+AUTO_ENV = {
+    "STATE_BACKEND": "memory",
+    "STORAGE_BACKEND": "volume",
+    "HSS_ORCH_STAGE_STUBS": "1",
+    "VISION_EXTERNAL_CALLS": "off",
+    "SMP_ENVIRONMENT": "dev",
+}
+
+
+@pytest.fixture
+def auto_client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> ApiClient:
+    """An app whose hooks hand finalize off to the real orchestrator entry points."""
+    for key, value in AUTO_ENV.items():
+        monkeypatch.setenv(key, value)
+    # services.current() inside the hook builds from env: it needs the same
+    # volume root the API storage uses, or the bind fails and hooks no-op.
+    monkeypatch.setenv("VOLUME_ROOT", str(tmp_path / "vol"))
+    settings = Settings(
+        STORAGE_BACKEND="volume",
+        STATE_BACKEND="memory",
+        VOLUME_ROOT=str(tmp_path / "vol"),
+        CAPTURE_API_PUBLIC_URL="http://testserver",
+        CAPTURE_API_TOKENS=f"device:{TOKEN}",
+        UPLOAD_URL_HMAC_SECRET="h" * 32,
+        HSS_INGEST_HOOKS="auto",
+        HSS_AUTH_MODE="static",
+        GIT_SHA="0123456789abcdef",
+        VISION_EXTERNAL_CALLS="off",
+        SMP_ENVIRONMENT="dev",
+        SMP_SERVICE_NAME="capture-api",
+    )
+    app = create_app(settings=settings)
+    test = ApiClient(app)
+    test.headers["Authorization"] = f"Bearer {TOKEN}"
+    test.app_state = app.state.deps
+    return test
+
 
 @pytest.fixture
 def client(tmp_path):
diff --git a/backend/server/api/tests/test_finalize_background.py b/backend/server/api/tests/test_finalize_background.py
index 6a04c161..28755d28 100644
--- a/backend/server/api/tests/test_finalize_background.py
+++ b/backend/server/api/tests/test_finalize_background.py
@@ -131,3 +131,77 @@ def probing_failure_path(*args, **kwargs):
     assert second.status_code == 202, second.text
     assert extras(deps, capture_id).get("run_started") is True
     assert second.json()["status"] == "processing"
+
+
+def test_failure_publication_error_releases_the_worker_slot(auto_client, monkeypatch):
+    """Finding: _guard (PR #293 Greptile P1) — the slot was cleared and the future
+    settled only after the failure publication. A publication that itself raises —
+    a transient state outage makes put_capture fail — exited the thread with the
+    slot busy and the future unsettled: later finalize retries answered
+    processing forever and no worker ever started again. The failure write stays
+    under the lock; the slot release and the future settlement always run.
+    """
+    import server.api.finalize as finalize_module
+    from server.api.tests.test_hooks_orchestrator import (
+        PACKET,
+        PACKET_ID,
+        _upload_and_commit,
+    )
+    from server.api.tests.test_hooks_orchestrator import (
+        _finalize as _finalize_packet,
+    )
+
+    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)
+
+    deps = client.app_state
+
+    def offline(_key):
+        raise RuntimeError("volume offline")
+
+    def failing_publication(*args, **kwargs):
+        raise RuntimeError("state outage: failure write raised")
+
+    original_get_bytes = deps.storage.get_bytes
+    original_publication = finalize_module._on_background_failure
+    # The finalize work and its failure publication both hit the outage. Both
+    # seams are swapped by hand and restored before the healed retry:
+    # monkeypatch.undo would also revert the fixture's env settings
+    # (HSS_ORCH_STAGE_STUBS), which the retry's handoff still needs.
+    deps.storage.get_bytes = offline
+    finalize_module._on_background_failure = failing_publication
+    client.hold_finalize = True
+    try:
+        body = _finalize_packet(client, capture_id)
+        assert body["status"] == "processing", body
+        # The worker's original error is what the Future carries, publication
+        # failure or not.
+        with pytest.raises(RuntimeError):
+            drain_finalize(deps)
+    finally:
+        client.hold_finalize = False
+        deps.storage.get_bytes = original_get_bytes
+        finalize_module._on_background_failure = original_publication
+
+    # The slot is free even though the publication raised.
+    assert finalize_module._slots[capture_id].inflight is False
+
+    # A finalize retry starts a new worker and the capture reaches a terminal
+    # state, instead of sitting on processing with no worker alive.
+    body = _finalize_packet(client, capture_id)
+    assert body["status"] == "processing"
+    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"
diff --git a/backend/server/api/tests/test_hooks_orchestrator.py b/backend/server/api/tests/test_hooks_orchestrator.py
index ee4925c1..0cdd2dd5 100644
--- a/backend/server/api/tests/test_hooks_orchestrator.py
+++ b/backend/server/api/tests/test_hooks_orchestrator.py
@@ -26,53 +26,13 @@
 
 import pytest
 
-from server.api.app import create_app
-from server.api.tests.conftest import TOKEN, ApiClient
-from server.common.config import Settings
+from server.api.tests.conftest import ApiClient
 
 FIXTURE = Path(__file__).resolve().parents[3] / "data" / "fixtures" / "packets" / "mini-arkit"
 PACKET_RAW = (FIXTURE / "packet.json").read_bytes()
 PACKET = json.loads(PACKET_RAW)
 PACKET_ID = PACKET["packetId"]
 
-# The documented dev handoff (SD-73): hooks on, stage stubs on, no external calls.
-AUTO_ENV = {
-    "STATE_BACKEND": "memory",
-    "STORAGE_BACKEND": "volume",
-    "HSS_ORCH_STAGE_STUBS": "1",
-    "VISION_EXTERNAL_CALLS": "off",
-    "SMP_ENVIRONMENT": "dev",
-}
-
-
-@pytest.fixture
-def auto_client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> ApiClient:
-    """An app whose hooks hand finalize off to the real orchestrator entry points."""
-    for key, value in AUTO_ENV.items():
-        monkeypatch.setenv(key, value)
-    # services.current() inside the hook builds from env: it needs the same
-    # volume root the API storage uses, or the bind fails and hooks no-op.
-    monkeypatch.setenv("VOLUME_ROOT", str(tmp_path / "vol"))
-    settings = Settings(
-        STORAGE_BACKEND="volume",
-        STATE_BACKEND="memory",
-        VOLUME_ROOT=str(tmp_path / "vol"),
-        CAPTURE_API_PUBLIC_URL="http://testserver",
-        CAPTURE_API_TOKENS=f"device:{TOKEN}",
-        UPLOAD_URL_HMAC_SECRET="h" * 32,
-        HSS_INGEST_HOOKS="auto",
-        HSS_AUTH_MODE="static",
-        GIT_SHA="0123456789abcdef",
-        VISION_EXTERNAL_CALLS="off",
-        SMP_ENVIRONMENT="dev",
-        SMP_SERVICE_NAME="capture-api",
-    )
-    app = create_app(settings=settings)
-    test = ApiClient(app)
-    test.headers["Authorization"] = f"Bearer {TOKEN}"
-    test.app_state = app.state.deps
-    return test
-
 
 def _content_type(role: str) -> str:
     if role in {"still", "keyframe"}:
@@ -241,3 +201,63 @@ def test_orchestrator_stage_failure_is_recorded_not_a_500(
     payload = result.json()
     assert payload["status"] == "manual_review"
     assert payload["outcome"]["kind"] == "manual_review"
+
+
+def test_confirmation_replay_recomposes_once(auto_client: ApiClient) -> None:
+    """A corrected answer POSTed twice recomposes the result exactly once.
+
+    The route's fingerprint check short-circuits the replayed confirmation
+    before the hook runs, so the second POST returns the stored row without
+    touching the run: the result stage runs the original pass plus one
+    recomposition, never three times.
+    """
+    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)
+    body = _finalize(client, capture_id)
+    run_id = body["runId"]
+
+    deps = client.app_state
+    assert deps.state.get_stage(run_id, "result").status == "done"
+
+    answer = {
+        "fieldId": "meter_number",
+        "kind": "corrected",
+        "value": "12345",
+        "imageId": PACKET["stills"][0]["id"],
+        "clientAt": "2026-10-16T00:00:00.000Z",
+    }
+    first = client.post(f"/v1/captures/{capture_id}/confirmations", json=answer)
+    assert first.status_code == 201, first.text
+    replay = client.post(f"/v1/captures/{capture_id}/confirmations", json=answer)
+    assert replay.status_code == 200, replay.text
+
+    run = deps.state.get_run(run_id)
+    assert run.status == "done"
+    # The recomposed pass records attempt 1 again (the coordinator recomputes
+    # spawn attempts), so "recomposed exactly once" is read from the event log:
+    # the original pass plus one re-run of each stage — the fingerprint
+    # short-circuit keeps the replayed POST from triggering a third pass.
+    events = deps.state.events_after(capture_id, 0, limit=500)
+    for stage in ("criteria", "result"):
+        started = [
+            event
+            for event in events
+            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]
+    result = client.get(f"/v1/captures/{capture_id}/result")
+    assert result.status_code == 200, result.text
+    assert result.json()["meter"]["value"] == "12345"
diff --git a/backend/server/orchestrator/api.py b/backend/server/orchestrator/api.py
index acac4d93..7ba0d0eb 100644
--- a/backend/server/orchestrator/api.py
+++ b/backend/server/orchestrator/api.py
@@ -16,7 +16,7 @@
 )
 from server.contracts.packet import Tier
 from server.orchestrator.clock import parse_iso
-from server.orchestrator.coordinator import RunCoordinator
+from server.orchestrator.coordinator import RunCoordinator, pending_correction_key, reset_criteria_and_result
 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
@@ -198,27 +198,10 @@ def _recompose_after_correction(capture_id: str, run: RunRecord) -> None:
             run_id=run.run_id,
             preserved_stages=preserved,
         )
-    now = services.clock.iso()
-    for stage in ("criteria", "result"):
-        rec = services.state.get_stage(run.run_id, stage)  # type: ignore[attr-defined]
-        services.state.upsert_stage(  # type: ignore[attr-defined]
-            StageRecord(
-                run_id=run.run_id,
-                stage=stage,
-                status="pending",
-                attempt=(rec.attempt if rec is not None else 0) + 1,
-                heartbeat=now,
-            )
-        )
-    run.status = "running"
-    # The run's deadline measured its first pass, so a correction accepted after
-    # it would hard-stop on the next tick. The re-run gets the same HARD_STOP_MIN
-    # budget start_run gives a new run, counted from the correction.
-    run.deadline_at = deadline_from(services.clock, services.settings.HARD_STOP_MIN)
-    services.state.put_run(run)  # type: ignore[attr-defined]
-    raw = dict(get_json(services.state, f"run_control:{run.run_id}") or {})  # type: ignore[attr-defined]
-    raw["merge_done"] = False
-    put_json(services.state, f"run_control:{run.run_id}", raw)  # type: ignore[attr-defined]
+    # 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)
     try:
         RunCoordinator(services).drive(run.run_id)
     except CoordinatorDied:
@@ -267,7 +250,41 @@ def on_confirmation(
     runs = services.state.runs_for_capture(capture_id)  # type: ignore[attr-defined]
     if runs:
         newest = max(runs, key=lambda item: item.created_at)
-        _recompose_after_correction(capture_id, newest)
+        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]
+    )
+    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:
diff --git a/backend/server/orchestrator/coordinator.py b/backend/server/orchestrator/coordinator.py
index 4fe3a5f3..44a1b0ca 100644
--- a/backend/server/orchestrator/coordinator.py
+++ b/backend/server/orchestrator/coordinator.py
@@ -37,11 +37,48 @@
 from server.orchestrator.outcomes import Decision, Facts, decide
 from server.orchestrator.runner import error_parts, is_retryable
 from server.orchestrator.runner import run_stage as execute_stage
-from server.orchestrator.services import Services
+from server.orchestrator.services import Services, deadline_from
 from server.orchestrator.spend import attempt_usd, gpu_spawn_blocked, vlm_budget_usd
 from server.orchestrator.stateio import get_json, put_json
 
 _FINAL = {"verdict_ready", "retake_request", "failed"}
+
+
+def pending_correction_key(run_id: str) -> str:
+    return f"run_pending_correction:{run_id}"
+
+
+def reset_criteria_and_result(services: Services, run: RunRecord) -> None:
+    """Send criteria and result back through the coordinator with the run alive.
+
+    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
+    measured its earlier pass, so the re-run gets the same HARD_STOP_MIN budget
+    ``start_run`` gives a new run, counted from now.
+    """
+    state = services.state
+    now = services.clock.iso()
+    for stage in ("criteria", "result"):
+        rec = state.get_stage(run.run_id, stage)
+        state.upsert_stage(
+            StageRecord(
+                run_id=run.run_id,
+                stage=stage,
+                status="pending",
+                attempt=(rec.attempt if rec is not None else 0) + 1,
+                heartbeat=now,
+            )
+        )
+    run.status = "running"
+    run.deadline_at = deadline_from(services.clock, services.settings.HARD_STOP_MIN)
+    state.put_run(run)
+    raw = dict(get_json(state, f"run_control:{run.run_id}") or {})
+    raw["merge_done"] = False
+    put_json(state, f"run_control:{run.run_id}", raw)
+
+
 _PROFILES = {"A", "B", "C", "S"}
 
 
@@ -542,6 +579,23 @@ 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:
+            return False
+        self.s.state.delete_blob(key)
+        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):
@@ -556,6 +610,8 @@ def _maybe_finish(self, run: RunRecord) -> None:
         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):
+            return
         self._write_run_files(run)
         self._apply_success(run)
 
diff --git a/backend/server/orchestrator/tests/test_greptile_followup.py b/backend/server/orchestrator/tests/test_greptile_followup.py
index 69a97524..3f047910 100644
--- a/backend/server/orchestrator/tests/test_greptile_followup.py
+++ b/backend/server/orchestrator/tests/test_greptile_followup.py
@@ -12,6 +12,8 @@
 from datetime import timedelta
 from pathlib import Path
 
+import pytest
+
 from server.contracts.api import ConfirmationRequest
 from server.contracts.jobs import FileRecord, RunRecord, StageRecord
 from server.orchestrator.api import on_confirmation
@@ -160,7 +162,18 @@ def test_confirmation_updates_the_reads_record(tmp_path: Path) -> None:
     # The recompose preserves the upstream work as terminal records at attempt 1
     # — the re-run reads stage that once threatened the answer never executes.
     rows = {rec.stage: rec for rec in services.state.stages("run_reads")}
-    for upstream in ("validate", "poses", "objects", "reads", "scale", "dense", "coverage", "scene", "objects_3d", "placement"):
+    for upstream in (
+        "validate",
+        "poses",
+        "objects",
+        "reads",
+        "scale",
+        "dense",
+        "coverage",
+        "scene",
+        "objects_3d",
+        "placement",
+    ):
         assert rows[upstream].status == "done", upstream
         assert rows[upstream].attempt == 1, upstream
     assert rows["criteria"].status == "done"
@@ -268,3 +281,92 @@ def test_success_reads_the_runs_profile_verdict(tmp_path: Path) -> None:
     # A-profile verdict asks for more views; verdict.C.json is absent, so the
     # old hard-coded read fell through to manual_review and blocked amendments.
     assert services.state.get_capture("cap_test").status == "needs_views"
+
+
+def test_correction_during_live_result_stage_recomposes_once(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Finding: orchestrator/api.py:169 (PR #293 Greptile P1) — a live run was left
+    alone on the assumption its result stage had not consumed the corrected reads.
+
+    A confirmation can arrive after the result stage hydrated its inputs but
+    before it finishes uploading. The hook saved the corrected reads and
+    returned: the run finished done with the pre-correction meter, and a
+    replayed confirmation short-circuits upstream, so nothing recomposed. The
+    hook must record a pending correction on the live run, and the coordinator
+    must consume it before marking the run done — resetting criteria and result
+    once, the same reset the done-run path uses.
+    """
+
+    import threading
+
+    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)
+
+    entered = threading.Event()
+    release = threading.Event()
+    original_result_run = result_stub.run
+    real_meter = result_stub._stub_meter
+
+    def gated_run(ctx):
+        # The stage has consumed its hydrated, pre-correction reads copy.
+        real_meter(ctx)
+        entered.set()
+        assert release.wait(30)
+        return original_result_run(ctx)
+
+    monkeypatch.setattr(result_stub, "run", gated_run)
+
+    errors: list[BaseException] = []
+
+    def drive() -> None:
+        try:
+            with bind(services):
+                start_run("cap_test", digest, uri)
+        except BaseException as exc:  # surfaced after the join below
+            errors.append(exc)
+
+    worker = threading.Thread(target=drive, daemon=True)
+    worker.start()
+    assert entered.wait(60), "result stage never reached its gate"
+
+    request = ConfirmationRequest(
+        fieldId="meter_number",
+        kind="corrected",
+        value="12345",
+        imageId="s1",
+        clientAt="2026-10-15T00:00:00.000Z",
+    )
+    # The correction and a replayed confirmation both land while the result
+    # stage holds its half-composed inputs. One pending marker must survive.
+    with bind(services):
+        on_confirmation("cap_test", request, "cfm_live", by="device")
+        on_confirmation("cap_test", request, "cfm_live", by="device")
+    release.set()
+    worker.join(60)
+    assert not errors, errors
+    assert not worker.is_alive(), "run never finished after the gate released"
+
+    (run,) = services.state.runs_for_capture("cap_test")
+    assert run.status == "done"
+    # The recomposed pass records attempt 1 again (the coordinator recomputes
+    # spawn attempts from failures+respawns), so "recomposed exactly once" is
+    # read from the event log: the original pass plus one re-run of each stage
+    # — the replayed confirmation is the same answer and must not add a third.
+    events = services.state.events_after("cap_test", 0, limit=500)
+    for stage in ("criteria", "result"):
+        started = [
+            event
+            for event in events
+            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]
+    # 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
+    # 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"