From 3a82dc0338d1a57d92ab683760648f5bf38313a8 Mon Sep 17 00:00:00 2001 From: Obvious Date: Fri, 16 Oct 2026 18:10:55 +0000 Subject: [PATCH 1/2] fix(finalize): release the worker slot when the failure publication raises A background finalize worker that failed wrote the failure to the capture record and emitted the terminal event under the finish lock, then cleared its slot and settled its Future only after that write returned. When the publication itself raised - a transient state outage makes put_capture fail - the exception escaped the thread with the slot busy and the Future unsettled: later finalize retries answered processing forever with no worker ever starting again, and drain_finalize hung waiting on a Future nobody would settle. The failure write keeps its lock; the slot release and the Future settlement moved into a finally path, so the worker's original error always reaches the client and a finalize retry starts a new worker. Regression: test_failure_publication_error_releases_the_worker_slot drives both the work and its publication into an outage, asserts the drained error is the original one, the slot is free, and a retried finalize runs the orchestrator to a terminal capture state. Co-authored-by: Sam Gu --- backend/server/api/finalize.py | 37 +++++++--- .../api/tests/test_finalize_background.py | 74 +++++++++++++++++++ 2 files changed, 101 insertions(+), 10 deletions(-) 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/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" From 1df5b441585f1706f27fcfdf3b3b8065b6f79aaa Mon Sep 17 00:00:00 2001 From: Obvious Date: Fri, 16 Oct 2026 18:10:55 +0000 Subject: [PATCH 2/2] fix(orchestrator): recompose a live run when a correction lands mid-result on_confirmation left a pending or running run alone on the assumption its result stage had not consumed the corrected reads yet. The window is real: verdict_ready is decided before the result stage finishes uploading, so a confirmation can arrive after the stage hydrated its inputs. The hook saved the corrected reads and returned; the run closed done with the pre-correction meter, and a replayed confirmation short-circuited upstream at the route's fingerprint check, so nothing ever recomposed the served payload. The hook now records a pending-correction marker when the running stages are criteria or result (earlier arrivals are applied by the reads stage itself), and the coordinator consumes the marker before marking the run done - resetting criteria and result once through the same reset the done-run recomposition uses. The consumed marker keeps a repeated finish or a replayed hook call from resetting twice, and the done-path recomposition's stale-marker delete uses the blob namespace, matching how stateio put_json stores it. Regression: test_correction_during_live_result_stage_recomposes_once gates the result stub after it reads its inputs, lands a correction and its replay in that window, and asserts one recomposition, a consumed marker, and the corrected meter in the served payload; test_confirmation_replay_recomposes_once pins the route's replay short-circuit to a single recomposition. Co-authored-by: Sam Gu --- backend/server/api/tests/conftest.py | 39 +++++++ .../api/tests/test_hooks_orchestrator.py | 102 ++++++++++------- backend/server/orchestrator/api.py | 63 +++++++---- backend/server/orchestrator/coordinator.py | 58 +++++++++- .../tests/test_greptile_followup.py | 104 +++++++++++++++++- 5 files changed, 300 insertions(+), 66 deletions(-) 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_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"