Skip to content

fix(backend): late corrections reach a running result, and finalize failures always free the worker #295

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
obvious-autobuild wants to merge 2 commits into
base: alpha
Choose a base branch
from
Open

fix(backend): late corrections reach a running result, and finalize failures always free the worker #295

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3a82dc0
fix(finalize): release the worker slot when the failure publication r…
ObviousApp Oct 16, 2026
1df5b44
fix(orchestrator): recompose a live run when a correction lands mid-r…
ObviousApp Oct 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 27 additions & 10 deletions backend/server/api/finalize.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Comment on lines +537 to +539

@chatgpt-codex-connector chatgpt-codex-connector Bot Oct 16, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Preserve queued finalize retries when clearing a failed slot

If another finalize request arrives while this worker is handling the exception, _kick sets the slot's again flag and returns a 202. This unconditional _clear_slot then clears that flag without starting another worker, so when failure publication also failed the capture can remain in processing with no worker and the request that was meant to retry it is lost. Preserve the queued retry or explicitly kick a new worker after releasing the slot.

Useful? React with 👍 / 👎.

return
future.set_result(None)

Expand Down
39 changes: 39 additions & 0 deletions backend/server/api/tests/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

import base64
import hashlib
from pathlib import Path

import pytest
from fastapi.testclient import TestClient
Expand All @@ -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):
Expand Down
74 changes: 74 additions & 0 deletions backend/server/api/tests/test_finalize_background.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
102 changes: 61 additions & 41 deletions backend/server/api/tests/test_hooks_orchestrator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"}:
Expand Down Expand Up @@ -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"
63 changes: 40 additions & 23 deletions backend/server/orchestrator/api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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]
Comment on lines +275 to +285

@greptile-apps greptile-apps Bot Oct 16, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Accepted corrections can stay stale

After on_confirmation reads a running run, the coordinator can finish the result and check for a marker before _record_pending_correction writes it. The hook then either finds no running stage and returns, or writes a marker that later ticks will never consume because the run is already done.

The saved answer never reaches the result, and the route skips repeated confirmations. Make recording the correction and closing the run share one lock or an atomic state change.

Prompt To Fix With AI 
This is a comment left during a code review.
Path: backend/server/orchestrator/api.py
Line: 275-285

Comment:
**Accepted corrections can stay stale**

After `on_confirmation` reads a running run, the coordinator can finish the result and check for a marker before `_record_pending_correction` writes it. The hook then either finds no running stage and returns, or writes a marker that later ticks will never consume because the run is already done.

The saved answer never reaches the result, and the route skips repeated confirmations. Make recording the correction and closing the run share one lock or an atomic state change.

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

Fix in Claude Code Fix in Codex

)
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:
Expand Down
Loading