fix(backend): version corrections and recover finalize retries

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

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]:

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

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,

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")

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()

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.
 

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