Skip to content

fix: 2026-05-30 full-review fixes — staged-run lifecycle, contract drift, error boundary #125

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

Merged
slittycode merged 3 commits into from
May 31, 2026
Merged

fix: 2026-05-30 full-review fixes — staged-run lifecycle, contract drift, error boundary #125

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
670e0d9
fix: staged-run lifecycle reliability + chain-of-custody/contract fixes
slittycode May 30, 2026
ea0564b
fix: address review follow-ups + ErrorBoundary fallback smoke test
slittycode May 30, 2026
821175d
test: lock the measurement + interpretation setup-failure terminalizers
slittycode May 31, 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
74 changes: 60 additions & 14 deletions apps/backend/analysis_runtime.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -807,15 +807,20 @@ def complete_measurement(
# defense-in-depth so an operator who exports the env var globally
# doesn't leak a stale `transcription` key into the measurement row.
measurement_result.pop("transcription", None)
self._update_measurement_row(
updated = self._update_measurement_row(
run_id,
status="completed",
result=measurement_result,
provenance=provenance,
diagnostics=diagnostics,
error=None,
guard_terminal=True,
)
self._enqueue_requested_followups(run_id)
# Only enqueue the downstream pipeline if this completion actually landed.
# If the run was interrupted in the TOCTOU window the update no-ops, and
# enqueuing here would resurrect a fresh pipeline for an interrupted run.
if updated:
self._enqueue_requested_followups(run_id)

def fail_measurement(
self,
Expand All @@ -832,6 +837,7 @@ def fail_measurement(
provenance=provenance,
diagnostics=diagnostics,
error=error,
guard_terminal=True,
)

def create_pitch_note_attempt(
Expand Down Expand Up @@ -949,11 +955,11 @@ def complete_pitch_note_attempt(
).fetchone()
if attempt_row is None:
raise KeyError(f"Unknown pitch/note translation attempt {attempt_id}")
conn.execute(
cursor = conn.execute(
"""
UPDATE pitch_note_translation_attempts
SET status = ?, result_json = ?, provenance_json = ?, diagnostics_json = ?, error_json = ?, updated_at = ?
WHERE id = ?
WHERE id = ? AND status NOT IN ('completed', 'failed', 'interrupted')
""",
(
"completed",
Expand All @@ -965,6 +971,12 @@ def complete_pitch_note_attempt(
attempt_id,
),
)
if cursor.rowcount == 0:
# Already terminal — e.g. interrupt_run flipped this attempt to
# 'interrupted' while its (now-orphaned) subprocess was still
# finishing. Do NOT resurrect it to 'completed' or hijack the
# run's preferred pointer (the documented resurrection bug).
return
conn.execute(
"""
UPDATE analysis_runs
Expand Down Expand Up @@ -1110,11 +1122,11 @@ def complete_mt3_attempt(
).fetchone()
if attempt_row is None:
raise KeyError(f"Unknown mt3 attempt {attempt_id}")
conn.execute(
cursor = conn.execute(
"""
UPDATE mt3_attempts
SET status = ?, result_json = ?, provenance_json = ?, diagnostics_json = ?, error_json = ?, updated_at = ?
WHERE id = ?
WHERE id = ? AND status NOT IN ('completed', 'failed', 'interrupted')
""",
(
"completed",
Expand All @@ -1126,6 +1138,10 @@ def complete_mt3_attempt(
attempt_id,
),
)
if cursor.rowcount == 0:
# Already terminal (interrupted while the orphaned MT3 subprocess
# was still finishing) — do not resurrect or hijack the pointer.
return
conn.execute(
"""
UPDATE analysis_runs
Expand Down Expand Up @@ -1292,11 +1308,11 @@ def complete_interpretation_attempt(
).fetchone()
if attempt_row is None:
raise KeyError(f"Unknown interpretation attempt {attempt_id}")
conn.execute(
cursor = conn.execute(
"""
UPDATE interpretation_attempts
SET status = ?, grounded_measurement_output_id = ?, grounded_pitch_note_attempt_id = ?, result_json = ?, provenance_json = ?, diagnostics_json = ?, error_json = ?, updated_at = ?
WHERE id = ?
WHERE id = ? AND status NOT IN ('completed', 'failed', 'interrupted')
""",
(
"completed",
Expand All @@ -1310,6 +1326,10 @@ def complete_interpretation_attempt(
attempt_id,
),
)
if cursor.rowcount == 0:
# Already terminal (interrupted between the is_run_interrupted
# gate and here) — do not resurrect or hijack the pointer.
return
conn.execute(
"""
UPDATE analysis_runs
Expand All @@ -1334,7 +1354,7 @@ def fail_interpretation_attempt(
"""
UPDATE interpretation_attempts
SET status = ?, grounded_measurement_output_id = ?, grounded_pitch_note_attempt_id = ?, result_json = ?, provenance_json = ?, diagnostics_json = ?, error_json = ?, updated_at = ?
WHERE id = ?
WHERE id = ? AND status NOT IN ('completed', 'failed', 'interrupted')
""",
(
"failed",
Expand Down Expand Up @@ -1703,13 +1723,26 @@ def _update_measurement_row(
provenance: dict[str, Any] | None = None,
diagnostics: dict[str, Any] | None = None,
error: dict[str, Any] | None = None,
) -> None:
guard_terminal: bool = False,
) -> bool:
# When guard_terminal is set, refuse to transition a measurement row that
# is already terminal. This closes the interrupt TOCTOU: interrupt_run can
# flip the row to 'interrupted' after _execute_measurement_run's
# is_run_interrupted check but before complete/fail. Without the guard the
# late writer resurrects it — and, via complete_measurement, would enqueue
# a fresh follow-up pipeline for an interrupted run. Returns True iff a row
# was actually updated.
terminal_guard = (
" AND status NOT IN ('completed', 'failed', 'interrupted')"
if guard_terminal
else ""
)
with self._connect() as conn:
conn.execute(
"""
cursor = conn.execute(
f"""
UPDATE measurement_outputs
SET status = ?, result_json = ?, provenance_json = ?, diagnostics_json = ?, error_json = ?, updated_at = ?
WHERE run_id = ?
WHERE run_id = ?{terminal_guard}
""",
(
status,
Expand All @@ -1721,6 +1754,8 @@ def _update_measurement_row(
run_id,
),
)
updated = cursor.rowcount > 0
return updated

def _update_attempt_row(
self,
Expand All @@ -1738,7 +1773,7 @@ def _update_attempt_row(
f"""
UPDATE {table}
SET status = ?, result_json = ?, provenance_json = ?, diagnostics_json = ?, error_json = ?, updated_at = ?
WHERE id = ?
WHERE id = ? AND status NOT IN ('completed', 'failed', 'interrupted')
""",
(
status,
Expand Down Expand Up @@ -1814,6 +1849,17 @@ def _enqueue_requested_followups(self, run_id: str) -> None:
).fetchone()
if run_row is None:
return
# Re-assert measurement is still completed before enqueuing. complete_measurement
# commits the measurement-complete update in one transaction and calls this in a
# separate one; an interrupt_run committing in between (measurement worker thread
# vs. event-loop thread) would otherwise leave inert 'queued' follow-up rows on an
# already-interrupted run. The caller's `if updated:` gate can't see this window.
measurement_row = conn.execute(
"SELECT status FROM measurement_outputs WHERE run_id = ?",
(run_id,),
).fetchone()
if measurement_row is None or measurement_row["status"] != "completed":
return
pitch_note_exists = conn.execute(
"SELECT 1 FROM pitch_note_translation_attempts WHERE run_id = ? LIMIT 1",
(run_id,),
Expand Down
Loading
Loading