From 2caad7e7690096a5057d7e2155ceff372b11bc7e Mon Sep 17 00:00:00 2001 From: jenkins Date: Sun, 16 Aug 2026 21:55:28 -0300 Subject: [PATCH] hermes: bind terminal commit to journal inode --- .../hermes-execution-safety-regression.py | 212 ++- dockerfiles/patch-hermes-execution-safety.py | 327 ++++- services/hermes/scripts/cli_lane_runner.py | 1062 +++++++++++++-- .../scripts/recover_cassandra_workers.py | 4 +- testing/tests/test_hermes_cli_lanes.py | 1175 ++++++++++++++++- testing/tests/test_hermes_worker_recovery.py | 22 +- 6 files changed, 2668 insertions(+), 134 deletions(-) diff --git a/dockerfiles/hermes-execution-safety-regression.py b/dockerfiles/hermes-execution-safety-regression.py index a77ca92d..42bd79ab 100644 --- a/dockerfiles/hermes-execution-safety-regression.py +++ b/dockerfiles/hermes-execution-safety-regression.py @@ -5,6 +5,7 @@ from __future__ import annotations import json import os import tempfile +import threading import unittest from types import SimpleNamespace from unittest import mock @@ -160,16 +161,36 @@ class AutomaticDecompositionSafetyTests(unittest.TestCase): ) ) - self.assertTrue( - kanban_db.complete_task( - self.connection, - task_id, - result="durable terminal result", - summary="durable terminal result", - replay_ended_run_id=run_id, + with mock.patch.object( + kanban_db, "_fire_kanban_lifecycle_hook" + ) as lifecycle: + self.assertTrue( + kanban_db.complete_task( + self.connection, + task_id, + result="durable terminal result", + summary="durable terminal result", + replay_ended_run_id=run_id, + ) ) - ) - self.assertEqual(self._status(task_id), "done") + lifecycle.assert_called_once() + self.assertEqual(lifecycle.call_args.args[:2], ("kanban_task_completed", task_id)) + self.assertEqual(lifecycle.call_args.kwargs["run_id"], run_id) + task = kanban_db.get_task(self.connection, task_id) + self.assertEqual(task.status, "done") + self.assertEqual(task.completed_run_id, run_id) + runs = kanban_db.list_runs(self.connection, task_id) + self.assertEqual(len(runs), 1) + self.assertEqual(runs[0].id, run_id) + self.assertEqual(runs[0].status, "done") + self.assertEqual(runs[0].outcome, "completed") + self.assertEqual(runs[0].summary, "durable terminal result") + completed_events = self.connection.execute( + "SELECT run_id FROM task_events " + "WHERE task_id = ? AND kind = 'completed' ORDER BY id", + (task_id,), + ).fetchall() + self.assertEqual([row["run_id"] for row in completed_events], [run_id]) def test_older_ended_run_cannot_complete_over_a_replacement(self) -> None: task_id = kanban_db.create_task(self.connection, title="replacement guard") @@ -208,6 +229,179 @@ class AutomaticDecompositionSafetyTests(unittest.TestCase): ) self.assertIn(self._status(task_id), {"blocked", "triage"}) + def test_equal_result_bytes_remain_bound_to_the_completing_run(self) -> None: + task_id = kanban_db.create_task(self.connection, title="same result retries") + self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id)) + first_run = kanban_db.get_task(self.connection, task_id).current_run_id + self.assertTrue( + kanban_db.block_task( + self.connection, + task_id, + reason="first attempt ended", + expected_run_id=first_run, + ) + ) + self.assertTrue(kanban_db.unblock_task(self.connection, task_id)) + self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id)) + second_run = kanban_db.get_task(self.connection, task_id).current_run_id + self.assertTrue( + kanban_db.complete_task( + self.connection, + task_id, + result="identical result bytes", + summary="second run wins", + expected_run_id=second_run, + ) + ) + + task = kanban_db.get_task(self.connection, task_id) + self.assertEqual(task.completed_run_id, second_run) + self.assertNotEqual(task.completed_run_id, first_run) + runs = {run.id: run for run in kanban_db.list_runs(self.connection, task_id)} + self.assertEqual(runs[first_run].outcome, "blocked") + self.assertEqual(runs[second_run].outcome, "completed") + + def test_exact_run_reclaim_succeeds_for_the_authoritative_run(self) -> None: + task_id = kanban_db.create_task(self.connection, title="recover exact run") + self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id)) + run_id = kanban_db.get_task(self.connection, task_id).current_run_id + + self.assertTrue( + kanban_db.reclaim_task( + self.connection, + task_id, + reason="invalid exact journal", + expected_run_id=run_id, + ) + ) + task = kanban_db.get_task(self.connection, task_id) + self.assertEqual(task.status, "ready") + self.assertIsNone(task.current_run_id) + + def test_stale_reclaim_before_transaction_preserves_replacement_run(self) -> None: + task_id = kanban_db.create_task(self.connection, title="replacement before txn") + self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id)) + old_run = kanban_db.get_task(self.connection, task_id).current_run_id + self.assertTrue( + kanban_db.block_task( + self.connection, + task_id, + reason="old recovery run", + expected_run_id=old_run, + ) + ) + self.assertTrue(kanban_db.unblock_task(self.connection, task_id)) + self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id)) + replacement = kanban_db.get_task(self.connection, task_id) + replacement_run = replacement.current_run_id + self.assertNotEqual(old_run, replacement_run) + signals = [] + + self.assertFalse( + kanban_db.reclaim_task( + self.connection, + task_id, + reason="stale journal", + expected_run_id=old_run, + signal_fn=lambda *args: signals.append(args), + ) + ) + latest = kanban_db.get_task(self.connection, task_id) + self.assertEqual(latest.status, "running") + self.assertEqual(latest.current_run_id, replacement_run) + self.assertEqual(latest.claim_lock, replacement.claim_lock) + self.assertEqual(signals, []) + + def test_reclaim_update_guard_preserves_run_changed_inside_transaction(self) -> None: + task_id = kanban_db.create_task(self.connection, title="replacement in txn") + self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id)) + old_run = kanban_db.get_task(self.connection, task_id).current_run_id + replacement = {} + + def install_replacement(*_args, **_kwargs): + cursor = self.connection.execute( + "INSERT INTO task_runs (task_id, status, claim_lock, started_at) " + "VALUES (?, 'running', ?, strftime('%s','now'))", + (task_id, "replacement-lock"), + ) + replacement["run_id"] = int(cursor.lastrowid) + self.connection.execute( + "UPDATE tasks SET current_run_id = ?, claim_lock = ? WHERE id = ?", + (replacement["run_id"], "replacement-lock", task_id), + ) + return {} + + with mock.patch.object( + kanban_db, + "_terminate_reclaimed_worker", + side_effect=install_replacement, + ): + self.assertFalse( + kanban_db.reclaim_task( + self.connection, + task_id, + reason="journal for old run", + expected_run_id=old_run, + ) + ) + + latest = kanban_db.get_task(self.connection, task_id) + self.assertEqual(latest.status, "running") + self.assertEqual(latest.current_run_id, replacement["run_id"]) + self.assertEqual(latest.claim_lock, "replacement-lock") + + def test_concurrent_exact_reclaim_and_completion_have_one_winner(self) -> None: + task_id = kanban_db.create_task(self.connection, title="concurrent finalizer") + self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id)) + run_id = kanban_db.get_task(self.connection, task_id).current_run_id + barrier = threading.Barrier(2) + outcomes = {} + errors = [] + + def reclaim() -> None: + try: + with kanban_db.connect_closing() as connection: + barrier.wait() + outcomes["reclaim"] = kanban_db.reclaim_task( + connection, + task_id, + reason="concurrent invalid journal", + expected_run_id=run_id, + ) + except BaseException as error: # pragma: no cover - assertion relay + errors.append(error) + + def finalize() -> None: + try: + with kanban_db.connect_closing() as connection: + barrier.wait() + outcomes["complete"] = kanban_db.complete_task( + connection, + task_id, + result="durable winner", + summary="durable winner", + expected_run_id=run_id, + ) + except BaseException as error: # pragma: no cover - assertion relay + errors.append(error) + + threads = [threading.Thread(target=reclaim), threading.Thread(target=finalize)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=10) + self.assertFalse(thread.is_alive()) + + self.assertEqual(errors, []) + self.assertEqual(set(outcomes), {"reclaim", "complete"}) + self.assertEqual(sum(bool(value) for value in outcomes.values()), 1) + latest = kanban_db.get_task(self.connection, task_id) + self.assertIn(latest.status, {"done", "ready"}) + self.assertIsNone(latest.current_run_id) + runs = kanban_db.list_runs(self.connection, task_id) + self.assertEqual(len(runs), 1) + self.assertIsNotNone(runs[0].ended_at) + def test_fresh_triage_task_still_auto_promotes(self) -> None: task_id = kanban_db.create_task( self.connection, diff --git a/dockerfiles/patch-hermes-execution-safety.py b/dockerfiles/patch-hermes-execution-safety.py index 6abd6f26..59c3be6d 100644 --- a/dockerfiles/patch-hermes-execution-safety.py +++ b/dockerfiles/patch-hermes-execution-safety.py @@ -72,9 +72,11 @@ complete_branch_after = ''' # Journal recovery may arrive after a legacy erro # accepted run. Verify that it is still the latest run in this same # transaction; a replacement attempt must make the replay fail closed. with write_txn(conn): + replayed_run_id = None if expected_run_id is not None and replay_ended_run_id is not None: raise ValueError("expected_run_id and replay_ended_run_id are mutually exclusive") if replay_ended_run_id is not None: + replayed_run_id = int(replay_ended_run_id) cur = conn.execute( """ UPDATE tasks @@ -95,8 +97,22 @@ complete_branch_after = ''' # Journal recovery may arrive after a legacy erro ORDER BY id DESC LIMIT 1 ) + AND EXISTS ( + SELECT 1 FROM task_runs + WHERE id = ? + AND task_id = ? + AND ended_at IS NOT NULL + ) """, - (result, now, task_id, int(replay_ended_run_id), task_id), + ( + result, + now, + task_id, + replayed_run_id, + task_id, + replayed_run_id, + task_id, + ), ) elif expected_run_id is None: cur = conn.execute( @@ -124,6 +140,315 @@ db = replace_once( "transactional exact ended-run completion", ) +run_completion_before = ''' run_id = _end_run( + conn, task_id, + outcome="completed", status="done", + summary=summary if summary is not None else result, + metadata=metadata, + ) + # If complete_task was called on a never-claimed task (ready or + # blocked → done with no run in flight), synthesize a + # zero-duration run so the handoff fields are persisted in + # attempt history instead of silently lost. + if run_id is None and (summary or metadata or result): + run_id = _synthesize_ended_run( + conn, task_id, + outcome="completed", + summary=summary if summary is not None else result, + metadata=metadata, + ) +''' +run_completion_after = ''' if replayed_run_id is None: + run_id = _end_run( + conn, task_id, + outcome="completed", status="done", + summary=summary if summary is not None else result, + metadata=metadata, + ) + # If complete_task was called on a never-claimed task (ready or + # blocked → done with no run in flight), synthesize a + # zero-duration run so the handoff fields are persisted in + # attempt history instead of silently lost. + if run_id is None and (summary or metadata or result): + run_id = _synthesize_ended_run( + conn, task_id, + outcome="completed", + summary=summary if summary is not None else result, + metadata=metadata, + ) + else: + run_id = replayed_run_id + replayed = conn.execute( + """ + UPDATE task_runs + SET status = 'done', + outcome = 'completed', + summary = ?, + error = NULL, + metadata = ?, + claim_lock = NULL, + claim_expires = NULL, + worker_pid = NULL + WHERE id = ? + AND task_id = ? + AND ended_at IS NOT NULL + """, + ( + summary if summary is not None else result, + json.dumps(metadata, ensure_ascii=False) if metadata else None, + run_id, + task_id, + ), + ) + if replayed.rowcount != 1: + raise RuntimeError("exact ended-run completion lost its run identity") + conn.execute( + "UPDATE tasks SET completed_run_id = ? WHERE id = ?", + (run_id, task_id), + ) +''' +db = replace_once( + db, + run_completion_before, + run_completion_after, + "exact completion run provenance", +) + +reclaim_before = '''def reclaim_task( + conn: sqlite3.Connection, + task_id: str, + *, + reason: Optional[str] = None, + signal_fn=None, +) -> bool: + """Operator-driven reclaim: release the claim and reset to ``ready``. + + Unlike :func:`release_stale_claims` which only acts on tasks whose + ``claim_expires`` has passed, this function reclaims immediately + regardless of TTL. Intended for the dashboard/CLI recovery flow + when an operator wants to abort a running worker without waiting + for the TTL to expire (e.g. after seeing a hallucination warning). + + Returns True if a reclaim happened, False if the task isn't in a + reclaimable state (not running, or doesn't exist). + """ + row = conn.execute( + "SELECT status, claim_lock, worker_pid FROM tasks WHERE id = ?", + (task_id,), + ).fetchone() + if not row: + return False + if row["status"] != "running" and row["claim_lock"] is None: + # Nothing to reclaim — already ready / blocked / done. + return False + prev_lock = row["claim_lock"] + termination = _terminate_reclaimed_worker( + row["worker_pid"], prev_lock, signal_fn=signal_fn, + ) + with write_txn(conn): + cur = conn.execute( + "UPDATE tasks SET status = 'ready', claim_lock = NULL, " + "claim_expires = NULL, worker_pid = NULL " + "WHERE id = ? AND status IN ('running', 'ready', 'blocked') " + "AND claim_lock IS ?", + (task_id, prev_lock), + ) + if cur.rowcount != 1: + return False + run_id = _end_run( + conn, task_id, + outcome="reclaimed", status="reclaimed", + error=( + f"manual_reclaim: {reason}" if reason + else f"manual_reclaim lock={prev_lock}" + ), + metadata=termination, + ) + payload = { + "manual": True, + "reason": reason, + "prev_lock": prev_lock, + } + payload.update(termination) + _append_event( + conn, task_id, "reclaimed", + payload, + run_id=run_id, + ) + # Operator intervention — they've looked at the task, so the + # consecutive-failures counter is now stale. Give the next retry + # a fresh budget. (_clear_failure_counter opens its own write_txn, + # so it runs after the enclosing one commits.) + _clear_failure_counter(conn, task_id) + return True +''' +reclaim_after = '''def reclaim_task( + conn: sqlite3.Connection, + task_id: str, + *, + reason: Optional[str] = None, + signal_fn=None, + expected_run_id: Optional[int] = None, +) -> bool: + """Operator-driven reclaim: release the claim and reset to ``ready``. + + Unlike :func:`release_stale_claims` which only acts on tasks whose + ``claim_expires`` has passed, this function reclaims immediately + regardless of TTL. Intended for the dashboard/CLI recovery flow + when an operator wants to abort a running worker without waiting + for the TTL to expire (e.g. after seeing a hallucination warning). + + When ``expected_run_id`` is supplied, the run identity is checked after + ``BEGIN IMMEDIATE`` and included in the guarded update. A replacement run + is therefore neither signalled nor reclaimed by stale recovery evidence. + + Returns True if a reclaim happened, False if the task isn't in a + reclaimable state (not running, or doesn't exist), or its run changed. + """ + with write_txn(conn): + row = conn.execute( + "SELECT status, claim_lock, worker_pid, current_run_id " + "FROM tasks WHERE id = ?", + (task_id,), + ).fetchone() + if not row: + return False + if expected_run_id is not None and row["current_run_id"] != expected_run_id: + return False + if row["status"] != "running" and row["claim_lock"] is None: + # Nothing to reclaim — already ready / blocked / done. + return False + prev_lock = row["claim_lock"] + termination = _terminate_reclaimed_worker( + row["worker_pid"], prev_lock, signal_fn=signal_fn, + ) + if expected_run_id is None: + cur = conn.execute( + "UPDATE tasks SET status = 'ready', claim_lock = NULL, " + "claim_expires = NULL, worker_pid = NULL " + "WHERE id = ? AND status IN ('running', 'ready', 'blocked') " + "AND claim_lock IS ?", + (task_id, prev_lock), + ) + else: + cur = conn.execute( + "UPDATE tasks SET status = 'ready', claim_lock = NULL, " + "claim_expires = NULL, worker_pid = NULL " + "WHERE id = ? AND status IN ('running', 'ready', 'blocked') " + "AND claim_lock IS ? AND current_run_id = ?", + (task_id, prev_lock, int(expected_run_id)), + ) + if cur.rowcount != 1: + return False + run_id = _end_run( + conn, task_id, + outcome="reclaimed", status="reclaimed", + error=( + f"manual_reclaim: {reason}" if reason + else f"manual_reclaim lock={prev_lock}" + ), + metadata=termination, + ) + payload = { + "manual": True, + "reason": reason, + "prev_lock": prev_lock, + } + payload.update(termination) + _append_event( + conn, task_id, "reclaimed", + payload, + run_id=run_id, + ) + # Operator intervention — they've looked at the task, so the + # consecutive-failures counter is now stale. Give the next retry + # a fresh budget. (_clear_failure_counter opens its own write_txn, + # so it runs after the enclosing one commits.) + _clear_failure_counter(conn, task_id) + return True +''' +db = replace_once( + db, + reclaim_before, + reclaim_after, + "transactional exact-run reclaim", +) + +task_field_before = ''' current_run_id: Optional[int] = None + workflow_template_id: Optional[str] = None +''' +task_field_after = ''' current_run_id: Optional[int] = None + # Durable provenance for the run that authoritatively completed this task. + completed_run_id: Optional[int] = None + workflow_template_id: Optional[str] = None +''' +db = replace_once( + db, + task_field_before, + task_field_after, + "completed-run task field", +) + +task_row_before = ''' current_run_id=( + row["current_run_id"] if "current_run_id" in keys else None + ), + workflow_template_id=( +''' +task_row_after = ''' current_run_id=( + row["current_run_id"] if "current_run_id" in keys else None + ), + completed_run_id=( + row["completed_run_id"] if "completed_run_id" in keys else None + ), + workflow_template_id=( +''' +db = replace_once( + db, + task_row_before, + task_row_after, + "completed-run row mapping", +) + +task_schema_before = ''' -- run is in-flight). Denormalised for cheap reads. + current_run_id INTEGER, + -- Forward-compat for v2 workflow routing. In v1 the kernel writes +''' +task_schema_after = ''' -- run is in-flight). Denormalised for cheap reads. + current_run_id INTEGER, + -- Immutable provenance for the task result's authoritative run. + completed_run_id INTEGER, + -- Forward-compat for v2 workflow routing. In v1 the kernel writes +''' +db = replace_once( + db, + task_schema_before, + task_schema_after, + "completed-run schema", +) + +task_migration_before = ''' if "current_run_id" not in cols: + _add_column_if_missing( + conn, "tasks", "current_run_id", "current_run_id INTEGER" + ) + if "workflow_template_id" not in cols: +''' +task_migration_after = ''' if "current_run_id" not in cols: + _add_column_if_missing( + conn, "tasks", "current_run_id", "current_run_id INTEGER" + ) + if "completed_run_id" not in cols: + _add_column_if_missing( + conn, "tasks", "completed_run_id", "completed_run_id INTEGER" + ) + if "workflow_template_id" not in cols: +''' +db = replace_once( + db, + task_migration_before, + task_migration_after, + "completed-run migration", +) + specify_signature_before = '''def specify_triage_task( conn: sqlite3.Connection, task_id: str, diff --git a/services/hermes/scripts/cli_lane_runner.py b/services/hermes/scripts/cli_lane_runner.py index 30c84daf..39305c0e 100644 --- a/services/hermes/scripts/cli_lane_runner.py +++ b/services/hermes/scripts/cli_lane_runner.py @@ -148,6 +148,42 @@ class TerminalIdentity: state: str +@dataclass +class TerminalSnapshot: + """One inode-bound, bounded journal read held open across finalization.""" + + document: dict[str, Any] + file_stat: os.stat_result + descriptor: int + directory_descriptor: int + + def close(self) -> None: + """Release the pinned file and directory descriptors.""" + os.close(self.descriptor) + os.close(self.directory_descriptor) + + +@dataclass +class TerminalRecoverySnapshot: + """One raw directory entry pinned before recovery classifies its payload.""" + + document: dict[str, Any] | None + file_stat: os.stat_result + descriptor: int | None + directory_descriptor: int + prefix: bytes + invalid_reason: str | None + + def close(self) -> None: + """Release the descriptors retained across classification/quarantine.""" + if self.descriptor is not None: + os.close(self.descriptor) + self.descriptor = None + if self.directory_descriptor >= 0: + os.close(self.directory_descriptor) + self.directory_descriptor = -1 + + def utc_now() -> str: return datetime.now(timezone.utc).isoformat() @@ -385,6 +421,15 @@ _TERMINAL_NAME = re.compile( r"^(?P[a-zA-Z0-9_.-]+)\.run-(?P[0-9]+)\." r"terminal\.(?Ppending|committed)\.json$" ) +_TERMINAL_EVIDENCE_NAME = re.compile( + r"^(?P[a-zA-Z0-9_.-]+)\.run-(?P[0-9]+)\." + r"terminal\.(?Pprepared|conflict)-(?P[a-f0-9]{32})\.json$" +) +_RETIRE_NAME = re.compile( + r"^\.retire\.(?P[a-f0-9]{16})\." + r"(?P[a-f0-9]{16})\.(?P[0-9]+)$" +) +_LEGACY_RETIRE_NAME = re.compile(r"^\.retire\.[a-f0-9]{32}\.[0-9]+$") _SAFE_BOARD = re.compile(r"^[a-zA-Z0-9_.-]+$") @@ -418,6 +463,36 @@ def _terminal_identity(path: Path) -> TerminalIdentity | None: ) +def _terminal_evidence_identity(path: Path) -> TerminalIdentity | None: + """Parse immutable prepared/conflict evidence authority from its path.""" + try: + relative = path.relative_to(STATE_ROOT) + except ValueError: + return None + if len(relative.parts) != 2: + return None + board, filename = relative.parts + match = _TERMINAL_EVIDENCE_NAME.fullmatch(filename) + try: + board_stat = path.parent.stat(follow_symlinks=False) + except OSError: + return None + if ( + not match + or not stat.S_ISDIR(board_stat.st_mode) + or not _SAFE_BOARD.fullmatch(board) + or board in {".", ".."} + or match.group("task") in {".", ".."} + ): + return None + return TerminalIdentity( + board=board, + task_id=match.group("task"), + run_id=int(match.group("run")), + state=match.group("state"), + ) + + def _read_bounded(descriptor: int, limit: int) -> bytes: """Read at most ``limit`` bytes from a regular file descriptor.""" chunks: list[bytes] = [] @@ -431,10 +506,8 @@ def _read_bounded(descriptor: int, limit: int) -> bytes: return b"".join(chunks) -def _load_terminal_json(path: Path, identity: TerminalIdentity) -> dict[str, Any]: - """Read one small, singly-linked journal through its validated directory.""" - if _terminal_identity(path) != identity: - return {} +def _open_small_json_snapshot(path: Path) -> TerminalSnapshot | None: + """Open and pin one bounded, singly-linked JSON document.""" directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) directory_flags |= getattr(os, "O_NOFOLLOW", 0) file_flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) @@ -449,10 +522,10 @@ def _load_terminal_json(path: Path, identity: TerminalIdentity) -> dict[str, Any or opened.st_nlink != 1 or opened.st_size > MAX_TERMINAL_RECORD_BYTES ): - return {} + return None payload = _read_bounded(descriptor, MAX_TERMINAL_RECORD_BYTES + 1) if len(payload) > MAX_TERMINAL_RECORD_BYTES: - return {} + return None current = os.stat( path.name, dir_fd=directory, @@ -464,16 +537,173 @@ def _load_terminal_json(path: Path, identity: TerminalIdentity) -> dict[str, Any or current.st_size != opened.st_size or len(payload) != opened.st_size ): - return {} + return None value = json.loads(payload.decode("utf-8")) + if not isinstance(value, dict): + return None + snapshot = TerminalSnapshot(value, opened, descriptor, directory) + descriptor = None + directory = None + return snapshot except (OSError, UnicodeError, json.JSONDecodeError): - return {} + return None finally: if descriptor is not None: os.close(descriptor) if directory is not None: os.close(directory) - return value if isinstance(value, dict) else {} + + +def _open_terminal_recovery_snapshot(path: Path) -> TerminalRecoverySnapshot | None: + """Pin the entry recovery inspected, even when its payload is malformed.""" + directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) + directory_flags |= getattr(os, "O_NOFOLLOW", 0) + file_flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) + directory = None + descriptor = None + try: + directory = os.open(path.parent, directory_flags) + observed = os.stat( + path.name, + dir_fd=directory, + follow_symlinks=False, + ) + if not stat.S_ISREG(observed.st_mode): + snapshot = TerminalRecoverySnapshot( + None, + observed, + None, + directory, + b"", + "non-regular", + ) + directory = None + return snapshot + try: + descriptor = os.open(path.name, file_flags, dir_fd=directory) + except OSError: + snapshot = TerminalRecoverySnapshot( + None, + observed, + None, + directory, + b"", + "open-failed", + ) + directory = None + return snapshot + opened = os.fstat(descriptor) + if opened.st_dev != observed.st_dev or opened.st_ino != observed.st_ino: + os.close(descriptor) + descriptor = None + snapshot = TerminalRecoverySnapshot( + None, + observed, + None, + directory, + b"", + "identity-changed-during-open", + ) + directory = None + return snapshot + observed = opened + if observed.st_nlink != 1: + snapshot = TerminalRecoverySnapshot( + None, + observed, + descriptor, + directory, + b"", + "hardlinked", + ) + descriptor = None + directory = None + return snapshot + if observed.st_size > MAX_TERMINAL_RECORD_BYTES: + prefix = _read_bounded(descriptor, QUARANTINE_HASH_BYTES) + snapshot = TerminalRecoverySnapshot( + None, + observed, + descriptor, + directory, + prefix, + "oversized", + ) + descriptor = None + directory = None + return snapshot + payload = _read_bounded(descriptor, MAX_TERMINAL_RECORD_BYTES + 1) + after_read = os.fstat(descriptor) + unchanged = ( + after_read.st_dev == observed.st_dev + and after_read.st_ino == observed.st_ino + and after_read.st_size == observed.st_size + and after_read.st_mtime_ns == observed.st_mtime_ns + and after_read.st_ctime_ns == observed.st_ctime_ns + and len(payload) == observed.st_size + ) + document = None + invalid_reason = "unstable-payload" + if unchanged: + try: + parsed = json.loads(payload.decode("utf-8")) + if isinstance(parsed, dict): + document = parsed + invalid_reason = None + else: + invalid_reason = "non-object-payload" + except (UnicodeError, json.JSONDecodeError): + invalid_reason = "malformed-payload" + snapshot = TerminalRecoverySnapshot( + document, + after_read, + descriptor, + directory, + payload[:QUARANTINE_HASH_BYTES], + invalid_reason, + ) + descriptor = None + directory = None + return snapshot + except OSError: + return None + finally: + if descriptor is not None: + os.close(descriptor) + if directory is not None: + os.close(directory) + + +def _open_terminal_snapshot( + path: Path, + identity: TerminalIdentity, +) -> TerminalSnapshot | None: + """Open a terminal journal only after its lexical identity is verified.""" + if _terminal_identity(path) != identity: + return None + return _open_small_json_snapshot(path) + + +def _load_terminal_json(path: Path, identity: TerminalIdentity) -> dict[str, Any]: + """Read one small, singly-linked journal through its validated directory.""" + snapshot = _open_terminal_snapshot(path, identity) + if snapshot is None: + return {} + try: + return snapshot.document + finally: + snapshot.close() + + +def _load_small_json(path: Path) -> dict[str, Any]: + """Read one bounded immutable evidence document without following links.""" + snapshot = _open_small_json_snapshot(path) + if snapshot is None: + return {} + try: + return snapshot.document + finally: + snapshot.close() def _persist_candidate( @@ -580,7 +810,7 @@ def _terminal_record_valid( and structured["blockers"] == [] and cli_lane_goal.unfinished_result_reason(structured) is None and record["summary"] == structured["summary"] - and record.get("kanban_state") in {"pending", "committed"} + and record.get("kanban_state") in {"pending", "prepared", "committed"} ) if not valid or identity is None: return valid @@ -588,6 +818,7 @@ def _terminal_record_valid( record.get("board") == identity.board and record.get("task_id") == identity.task_id and record.get("expected_run_id") == identity.run_id + and record.get("kanban_state") == identity.state ) @@ -1167,6 +1398,7 @@ def _retire_terminal_entry( *, board_descriptor: int, quarantine_descriptor: int, + authority_name: str | None = None, ) -> str: """Retire only the inode previously inspected; preserve any replacement.""" try: @@ -1179,11 +1411,18 @@ def _retire_terminal_entry( return "missing" if current.st_dev != source_stat.st_dev or current.st_ino != source_stat.st_ino: return "replacement" - identity = hashlib.sha256( - f"{path.name}\0{source_stat.st_dev:x}\0{source_stat.st_ino:x}".encode("utf-8") - ).hexdigest()[:32] + # A replacement can itself be staged repeatedly while recovery races a + # writer. Keep every generation bound to the original canonical pending + # name instead of hashing an intermediate .retire name and hiding it from + # the next recovery pass. + path_digest = hashlib.sha256( + (authority_name or path.name).encode("utf-8") + ).hexdigest()[:16] + source_digest = hashlib.sha256( + f"{source_stat.st_dev:x}\0{source_stat.st_ino:x}".encode("utf-8") + ).hexdigest()[:16] for sequence in range(32): - staging = f".retire.{path.name}.{identity}.{sequence}" + staging = f".retire.{path_digest}.{source_digest}.{sequence}" try: _rename_noreplace( path.name, @@ -1222,7 +1461,189 @@ def _retire_terminal_entry( return "replacement" -def _quarantine_terminal(path: Path, identity: TerminalIdentity | None, reason: str) -> Path: +def _write_json_noreplace(path: Path, value: dict[str, Any]) -> bool: + """Durably create one immutable JSON artifact without replacing a peer.""" + payload = (json.dumps(value, indent=2, sort_keys=True) + "\n").encode("utf-8") + if len(payload) > MAX_TERMINAL_RECORD_BYTES + 4096: + raise ValueError("terminal evidence exceeds the bounded artifact limit") + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + os.chmod(path.parent, 0o700, follow_symlinks=False) + directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) + directory_flags |= getattr(os, "O_NOFOLLOW", 0) + create_flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL + create_flags |= getattr(os, "O_NOFOLLOW", 0) + directory = os.open(path.parent, directory_flags) + temporary = f".{path.name}.{uuid.uuid4().hex}.tmp" + temporary_stat = None + descriptor = None + try: + descriptor = os.open(temporary, create_flags, 0o600, dir_fd=directory) + temporary_stat = os.fstat(descriptor) + with os.fdopen(descriptor, "wb") as stream: + descriptor = None + stream.write(payload) + stream.flush() + os.fsync(stream.fileno()) + written = os.fstat(stream.fileno()) + if stat.S_IMODE(written.st_mode) != 0o600 or written.st_nlink != 1: + raise OSError("terminal evidence is not private and singly linked") + try: + _rename_noreplace( + temporary, + path.name, + source_dir=directory, + destination_dir=directory, + ) + except FileExistsError: + return False + os.fsync(directory) + return True + finally: + if descriptor is not None: + os.close(descriptor) + if temporary_stat is not None: + try: + _retire_terminal_entry( + Path(temporary), + temporary_stat, + board_descriptor=directory, + quarantine_descriptor=directory, + ) + except OSError: + pass + os.close(directory) + + +def _terminal_document_digest(document: dict[str, Any]) -> str: + """Return the stable identity for one exact accepted result document.""" + immutable = { + key: document.get(key) + for key in ( + "board", + "task_id", + "expected_run_id", + "result", + "summary", + "metadata", + ) + } + canonical = json.dumps(immutable, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:32] + + +def _terminal_evidence_path( + identity: TerminalIdentity, + state: str, + document: dict[str, Any], +) -> Path: + """Return a deterministic immutable prepared or conflict evidence path.""" + if state not in {"prepared", "conflict"}: + raise ValueError(f"invalid terminal evidence state: {state}") + base = state_path(identity.board, identity.task_id) + digest = _terminal_document_digest(document) + return base.with_name( + f"{base.stem}.run-{identity.run_id}.terminal.{state}-{digest}.json" + ) + + +def _same_terminal_document(left: dict[str, Any], right: dict[str, Any]) -> bool: + """Compare the immutable result-bearing fields of terminal evidence.""" + keys = ("board", "task_id", "expected_run_id", "result", "summary", "metadata") + return all(left.get(key) == right.get(key) for key in keys) + + +def _terminal_evidence_valid( + document: dict[str, Any], + identity: TerminalIdentity, + state: str, +) -> bool: + """Validate semantic terminal content plus its immutable evidence state.""" + if document.get("kanban_state") != state: + return False + semantic = dict(document) + semantic_identity = identity + if state == "conflict": + semantic["kanban_state"] = "prepared" + semantic_identity = TerminalIdentity( + identity.board, + identity.task_id, + identity.run_id, + "prepared", + ) + return _terminal_record_valid(semantic, semantic_identity) + + +def _persist_prepared_evidence( + identity: TerminalIdentity, + document: dict[str, Any], +) -> Path: + """Fsync an immutable accepted result before entering the DB boundary.""" + prepared = dict(document) + prepared["kanban_state"] = "prepared" + prepared["prepared_at"] = utc_now() + path = _terminal_evidence_path(identity, "prepared", document) + if _write_json_noreplace(path, prepared): + return path + if _terminal_evidence_identity(path) != TerminalIdentity( + identity.board, identity.task_id, identity.run_id, "prepared" + ): + raise OSError("prepared terminal evidence identity is invalid") + existing = _load_small_json(path) + prepared_identity = TerminalIdentity( + identity.board, + identity.task_id, + identity.run_id, + "prepared", + ) + if not _terminal_evidence_valid( + existing, + prepared_identity, + "prepared", + ) or not _same_terminal_document(existing, prepared): + raise OSError("prepared terminal evidence path contains a conflicting result") + return path + + +def _persist_conflict_evidence( + identity: TerminalIdentity, + document: dict[str, Any], + reason: str, +) -> Path: + """Preserve a losing valid result in full under deterministic retention.""" + conflict = dict(document) + conflict["kanban_state"] = "conflict" + conflict["conflict_reason"] = reason + conflict["conflicted_at"] = utc_now() + path = _terminal_evidence_path(identity, "conflict", document) + if _write_json_noreplace(path, conflict): + return path + if _terminal_evidence_identity(path) != TerminalIdentity( + identity.board, identity.task_id, identity.run_id, "conflict" + ): + raise OSError("terminal conflict evidence identity is invalid") + existing = _load_small_json(path) + conflict_identity = TerminalIdentity( + identity.board, + identity.task_id, + identity.run_id, + "conflict", + ) + if not _terminal_evidence_valid( + existing, + conflict_identity, + "conflict", + ) or not _same_terminal_document(existing, conflict): + raise OSError("terminal conflict evidence path contains a different result") + return path + + +def _quarantine_terminal( + path: Path, + identity: TerminalIdentity | None, + reason: str, + *, + snapshot: TerminalRecoverySnapshot | None = None, +) -> Path: """Record bounded metadata, then retire only the inspected journal inode.""" board_dir = path.parent try: @@ -1247,18 +1668,30 @@ def _quarantine_terminal(path: Path, identity: TerminalIdentity | None, reason: source_stat = None destination = path retirement = "not-attempted" + owns_source = snapshot is None try: - board_descriptor = os.open(board_dir, board_flags) - source_stat = os.stat( - path.name, - dir_fd=board_descriptor, - follow_symlinks=False, - ) + if snapshot is None: + board_descriptor = os.open(board_dir, board_flags) + source_stat = os.stat( + path.name, + dir_fd=board_descriptor, + follow_symlinks=False, + ) + else: + board_descriptor = snapshot.directory_descriptor + source_descriptor = snapshot.descriptor + source_stat = snapshot.file_stat + pinned_board = os.fstat(board_descriptor) + if ( + pinned_board.st_dev != board_stat.st_dev + or pinned_board.st_ino != board_stat.st_ino + ): + raise OSError("terminal recovery board identity changed") safe_source = stat.S_ISREG(source_stat.st_mode) and source_stat.st_nlink == 1 digest = None hashed_bytes = 0 hash_complete = False - if safe_source: + if safe_source and snapshot is None: source_descriptor = os.open( path.name, file_flags, @@ -1279,6 +1712,14 @@ def _quarantine_terminal(path: Path, identity: TerminalIdentity | None, reason: hashed_bytes = len(prefix) hash_complete = opened_stat.st_size <= QUARANTINE_HASH_BYTES digest = hashlib.sha256(prefix).hexdigest() + elif safe_source and source_descriptor is not None: + prefix = snapshot.prefix + hashed_bytes = len(prefix) + hash_complete = ( + source_stat.st_size <= QUARANTINE_HASH_BYTES + and hashed_bytes == source_stat.st_size + ) + digest = hashlib.sha256(prefix).hexdigest() source_kind = ( "regular" if safe_source @@ -1371,11 +1812,11 @@ def _quarantine_terminal(path: Path, identity: TerminalIdentity | None, reason: flush=True, ) finally: - if source_descriptor is not None: + if owns_source and source_descriptor is not None: os.close(source_descriptor) if quarantine_descriptor is not None: os.close(quarantine_descriptor) - if board_descriptor is not None: + if owns_source and board_descriptor is not None: os.close(board_descriptor) print( "quarantined terminal journal " @@ -1411,6 +1852,7 @@ def _recover_exact_run( conn, identity.task_id, reason=f"terminal journal recovery failed ({reason}); exact run may retry", + expected_run_id=identity.run_id, ) ) @@ -1429,36 +1871,91 @@ def _recover_exact_run( return recovered -def _commit_terminal_file(path: Path, identity: TerminalIdentity, document: dict[str, Any]) -> Path: - """Persist committed state, then atomically retire a pending journal.""" +def _retire_snapshot( + path: Path, + snapshot: TerminalSnapshot, + *, + authority_name: str | None = None, +) -> str: + """Retire only the directory entry still naming an opened snapshot inode.""" + return _retire_terminal_entry( + path, + snapshot.file_stat, + board_descriptor=snapshot.directory_descriptor, + quarantine_descriptor=snapshot.directory_descriptor, + authority_name=authority_name, + ) + + +def _retire_snapshot_after_db(path: Path, snapshot: TerminalSnapshot) -> str: + """Keep a committed DB result recoverable across retirement fsync errors.""" + try: + return _retire_snapshot(path, snapshot) + except OSError as error: + print( + f"terminal journal retirement deferred after DB commit: " + f"{type(error).__name__}: {error}", + file=sys.stderr, + flush=True, + ) + return "deferred" + + +def _discard_evidence(path: Path) -> str: + """Identity-safely remove one immutable prepared evidence artifact.""" + snapshot = _open_small_json_snapshot(path) + if snapshot is None: + return "missing" + try: + return _retire_snapshot(path, snapshot) + finally: + snapshot.close() + + +def _promote_prepared_evidence( + identity: TerminalIdentity, + document: dict[str, Any], + prepared: Path, +) -> tuple[Path, str]: + """Publish the DB-winning result without overwriting first-writer evidence.""" committed = _terminal_path( state_path(identity.board, identity.task_id), identity.run_id, "committed", ) - document["kanban_state"] = "committed" - document["committed_at"] = utc_now() - atomic_json(path, document) - os.replace(path, committed) - os.chmod(committed, 0o600, follow_symlinks=False) - _fsync_directory(committed.parent) - return committed + committed_document = dict(document) + committed_document["kanban_state"] = "committed" + committed_document["committed_at"] = utc_now() + state = "committed" + if not _write_json_noreplace(committed, committed_document): + committed_identity = TerminalIdentity( + identity.board, + identity.task_id, + identity.run_id, + "committed", + ) + existing = _load_terminal_json(committed, committed_identity) + if ( + not _terminal_record_valid(existing, committed_identity) + or existing.get("kanban_state") != "committed" + or not _same_terminal_document(existing, committed_document) + ): + _persist_conflict_evidence( + identity, + document, + "canonical committed evidence already contains a different result", + ) + state = "conflict" + _discard_evidence(prepared) + return committed, state -def _finalize_terminal_record( +def _finalize_document_db( kanban_db: Any, - path: Path, - _record: dict[str, Any] | None = None, + identity: TerminalIdentity, + document: dict[str, Any], ) -> str: - """Commit one exact-run terminal journal, or leave it safely pending.""" - identity = _terminal_identity(path) - if identity is None or identity.state != "pending": - return "invalid" - document = _load_terminal_json(path, identity) - if not _terminal_record_valid(document): - return "invalid" - if not _terminal_record_valid(document, identity): - return "foreign" + """Apply one validated exact-run result to Kanban under DB run guards.""" def operation(conn: Any) -> str: task = kanban_db.get_task(conn, identity.task_id) @@ -1468,7 +1965,11 @@ def _finalize_terminal_record( if status == "done": return ( "committed" - if str(_task_value(task, "result", "") or "") == document["result"] + if ( + _task_value(task, "completed_run_id", None) == identity.run_id + and str(_task_value(task, "result", "") or "") + == document["result"] + ) else "stale" ) current_run_id = _task_value(task, "current_run_id", None) @@ -1490,78 +1991,372 @@ def _finalize_terminal_record( metadata=document["metadata"], **completion_guard, ) - return "committed" if completed else "pending" + if completed: + return "committed" + # A duplicate finalizer can lose the guarded UPDATE to an identical + # first writer. Re-read rather than reporting a false pending state. + latest = kanban_db.get_task(conn, identity.task_id) + if latest is not None and str(_task_value(latest, "status", "")) == "done": + return ( + "committed" + if ( + _task_value(latest, "completed_run_id", None) == identity.run_id + and str(_task_value(latest, "result", "") or "") + == document["result"] + ) + else "stale" + ) + return "pending" - outcome = str(_board_call(kanban_db, identity.board, operation)) - if outcome == "committed": - _commit_terminal_file(path, identity, document) + return str(_board_call(kanban_db, identity.board, operation)) + + +def _resolve_pending_after_winner( + path: Path, + identity: TerminalIdentity, + winning_document: dict[str, Any], +) -> bool: + """Retire duplicates and preserve differing replacements as conflicts.""" + for _attempt in range(8): + snapshot = _open_terminal_snapshot(path, identity) + if snapshot is None: + return False + document = snapshot.document + if not _terminal_record_valid(document, identity): + snapshot.close() + return False + differing = not _same_terminal_document(document, winning_document) + try: + if differing: + _persist_conflict_evidence( + identity, + document, + "different valid result lost the exact-run first-writer race", + ) + retirement = _retire_snapshot_after_db(path, snapshot) + finally: + snapshot.close() + if retirement in {"retired", "missing"}: + if differing: + print( + f"preserved terminal result conflict for {identity.board}/" + f"{identity.task_id} run {identity.run_id}", + file=sys.stderr, + flush=True, + ) + return differing + if retirement not in {"replacement", "replacement-staged", "collision"}: + return differing + return False + + +def _finalize_terminal_record( + kanban_db: Any, + path: Path, + _record: dict[str, Any] | None = None, + *, + snapshot: TerminalRecoverySnapshot | None = None, +) -> str: + """Commit one inode-pinned exact-run journal, or leave it safely pending.""" + identity = _terminal_identity(path) + if identity is None or identity.state != "pending": + if snapshot is not None: + snapshot.close() + return "invalid" + pinned: TerminalSnapshot | TerminalRecoverySnapshot | None = snapshot + if pinned is None: + pinned = _open_terminal_snapshot(path, identity) + if pinned is None or pinned.document is None: + if pinned is not None: + pinned.close() + return "invalid" + document = pinned.document + try: + if not _terminal_record_valid(document): + return "invalid" + if not _terminal_record_valid(document, identity): + return "foreign" + prepared = _persist_prepared_evidence(identity, document) + outcome = _finalize_document_db(kanban_db, identity, document) + if outcome == "committed": + _committed, evidence_state = _promote_prepared_evidence( + identity, + document, + prepared, + ) + retirement = _retire_snapshot_after_db(path, pinned) + elif outcome == "stale": + _persist_conflict_evidence( + identity, + document, + "valid terminal result no longer matches the authoritative task run", + ) + _discard_evidence(prepared) + retirement = _retire_snapshot_after_db(path, pinned) + outcome = "conflict" + evidence_state = "conflict" + else: + return outcome + finally: + pinned.close() + if outcome == "committed" and retirement not in {"retired", "missing"}: + _resolve_pending_after_winner(path, identity, document) + if evidence_state == "conflict": + print( + f"terminal first-writer conflict recorded for {identity.board}/" + f"{identity.task_id} run {identity.run_id}", + file=sys.stderr, + flush=True, + ) return outcome -def _replay_replacement_terminal( - kanban_db: Any, - path: Path, - identity: TerminalIdentity, -) -> tuple[bool, bool]: - """Replay a valid journal atomically substituted during quarantine.""" - replacement = _load_terminal_json(path, identity) - if not _terminal_record_valid(replacement, identity): - return False, False +def _terminal_entry_absent(path: Path) -> bool: + """Confirm absence through a nofollow directory descriptor.""" + directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) + directory_flags |= getattr(os, "O_NOFOLLOW", 0) + directory = None try: - outcome = _finalize_terminal_record(kanban_db, path, replacement) - except Exception as error: - _record_board_access_error(identity.board, error) - return True, False - return True, outcome == "committed" + directory = os.open(path.parent, directory_flags) + os.stat(path.name, dir_fd=directory, follow_symlinks=False) + return False + except FileNotFoundError: + return True + except OSError: + return False + finally: + if directory is not None: + os.close(directory) + + +def _staged_terminal_authority( + path: Path, + snapshot: TerminalRecoverySnapshot, +) -> tuple[TerminalIdentity, Path] | None: + """Bind a staged inode to the canonical pending name it was moved from.""" + try: + relative = path.relative_to(STATE_ROOT) + except ValueError: + return None + if len(relative.parts) != 2: + return None + board, filename = relative.parts + match = _RETIRE_NAME.fullmatch(filename) + legacy = _LEGACY_RETIRE_NAME.fullmatch(filename) + document = snapshot.document + if ( + (match is None and legacy is None) + or document is None + or not _SAFE_BOARD.fullmatch(board) + or board in {".", ".."} + or document.get("board") != board + or not isinstance(document.get("task_id"), str) + or not isinstance(document.get("expected_run_id"), int) + ): + return None + identity = TerminalIdentity( + board, + document["task_id"], + document["expected_run_id"], + "pending", + ) + canonical = _terminal_path( + state_path(identity.board, identity.task_id), + identity.run_id, + "pending", + ) + if ( + canonical.parent != path.parent + or ( + match is not None + and hashlib.sha256(canonical.name.encode("utf-8")).hexdigest()[:16] + != match.group("path_digest") + ) + or not _terminal_record_valid(document, identity) + ): + return None + return identity, canonical + + +def _recover_retirement_staging() -> int: + """Promote valid hidden journals to durable prepared evidence.""" + recovered = 0 + for path in sorted(STATE_ROOT.glob("*/.retire.*")): + snapshot = _open_terminal_recovery_snapshot(path) + if snapshot is None: + continue + authority = _staged_terminal_authority(path, snapshot) + if authority is None: + try: + _quarantine_terminal( + path, + None, + "untrusted-retirement-staging", + snapshot=snapshot, + ) + finally: + snapshot.close() + continue + identity, canonical = authority + try: + _persist_prepared_evidence(identity, snapshot.document or {}) + retirement = _retire_snapshot( + path, + snapshot, + authority_name=canonical.name, + ) + except Exception as error: + _record_board_access_error(identity.board, error) + continue + finally: + snapshot.close() + if retirement in {"retired", "missing"}: + recovered += 1 + return recovered + + +def _recover_prepared_finalizations(kanban_db: Any) -> int: + """Recover an accepted result staged before an interrupted DB boundary.""" + recovered = 0 + prepared_paths = list(STATE_ROOT.glob("*/*.terminal.prepared-*.json")) + + def durable_order(path: Path) -> tuple[int, str]: + try: + return path.stat(follow_symlinks=False).st_mtime_ns, str(path) + except OSError: + return 2**63 - 1, str(path) + + # When no DB winner exists yet, the first durably prepared result owns the + # run. The DB transaction remains the final arbiter across runner processes. + for path in sorted(prepared_paths, key=durable_order): + identity = _terminal_evidence_identity(path) + snapshot = _open_small_json_snapshot(path) + if identity is None or identity.state != "prepared" or snapshot is None: + if snapshot is not None: + snapshot.close() + _quarantine_terminal(path, identity, "malformed-prepared-evidence") + continue + document = snapshot.document + try: + expected = _terminal_evidence_path(identity, "prepared", document) + if ( + expected.name != path.name + or not _terminal_evidence_valid(document, identity, "prepared") + ): + _quarantine_terminal(path, identity, "foreign-prepared-evidence") + continue + try: + outcome = _finalize_document_db(kanban_db, identity, document) + except Exception as error: + _record_board_access_error(identity.board, error) + continue + try: + if outcome == "committed": + _promote_prepared_evidence(identity, document, path) + pending = _terminal_path( + state_path(identity.board, identity.task_id), + identity.run_id, + "pending", + ) + pending_identity = TerminalIdentity( + identity.board, + identity.task_id, + identity.run_id, + "pending", + ) + _resolve_pending_after_winner( + pending, + pending_identity, + document, + ) + recovered += 1 + elif outcome == "stale": + _persist_conflict_evidence( + identity, + document, + "prepared result no longer matches the authoritative task run", + ) + _discard_evidence(path) + except Exception as error: + _record_board_access_error(identity.board, error) + continue + finally: + snapshot.close() + return recovered def recover_pending_finalizations() -> int: """Replay accepted exact-run results before scheduling more provider work.""" from hermes_cli import kanban_db - recovered = 0 + _recover_retirement_staging() + recovered = _recover_prepared_finalizations(kanban_db) for path in sorted(STATE_ROOT.glob("*/*.terminal.pending.json")): identity = _terminal_identity(path) - record = _load_terminal_json(path, identity) if identity is not None else {} - if identity is None: - _quarantine_terminal(path, None, "malformed-name") - continue - if not _terminal_record_valid(record): - _quarantine_terminal(path, identity, "malformed-payload") - replacement, committed = _replay_replacement_terminal( - kanban_db, path, identity - ) - if committed: + last_invalid_reason = "malformed-name" if identity is None else None + retired_invalid = False + for _replacement_attempt in range(16): + snapshot = _open_terminal_recovery_snapshot(path) + if snapshot is None: + if ( + identity is not None + and retired_invalid + and _terminal_entry_absent(path) + ): + _recover_exact_run( + kanban_db, + identity, + last_invalid_reason or "malformed-payload", + ) + break + record = snapshot.document or {} + reason = None + if identity is None: + reason = "malformed-name" + elif not _terminal_record_valid(record): + reason = snapshot.invalid_reason or "malformed-payload" + elif not _terminal_record_valid(record, identity): + reason = "foreign-identity" + if reason is not None: + try: + _quarantine_terminal( + path, + identity, + reason, + snapshot=snapshot, + ) + finally: + snapshot.close() + retired_invalid = True + last_invalid_reason = reason + continue + assert identity is not None + try: + outcome = _finalize_terminal_record( + kanban_db, + path, + record, + snapshot=snapshot, + ) + except Exception as error: + _record_board_access_error(identity.board, error) + break + if outcome == "committed": recovered += 1 - elif not replacement: - _recover_exact_run(kanban_db, identity, "malformed-payload") - continue - if not _terminal_record_valid(record, identity): - _quarantine_terminal(path, identity, "foreign-identity") - replacement, committed = _replay_replacement_terminal( - kanban_db, path, identity + break + if outcome in {"invalid", "foreign"}: + # The pathname changed after the recovery snapshot closed. + # Reclassify and pin the new inode instead of quarantining it + # using authority derived from the older entry. + last_invalid_reason = outcome + continue + break + else: + print( + f"terminal journal replacement churn deferred for {path.name}", + file=sys.stderr, + flush=True, ) - if committed: - recovered += 1 - elif not replacement: - _recover_exact_run(kanban_db, identity, "foreign-identity") - continue - try: - outcome = _finalize_terminal_record(kanban_db, path, record) - except Exception as error: - _record_board_access_error(identity.board, error) - continue - if outcome == "committed": - recovered += 1 - elif outcome in {"invalid", "foreign", "stale"}: - _quarantine_terminal(path, identity, outcome) - replacement, committed = _replay_replacement_terminal( - kanban_db, path, identity - ) - if committed: - recovered += 1 - elif not replacement: - _recover_exact_run(kanban_db, identity, outcome) return recovered @@ -1573,14 +2368,41 @@ def _has_pending_finalization(board: str, task_id: str, run_id: Any) -> bool: try: path.stat() except FileNotFoundError: - return False + pass except OSError: return False - identity = _terminal_identity(path) - if identity is None: - return False - record = _load_terminal_json(path, identity) - return _terminal_record_valid(record, identity) + else: + identity = _terminal_identity(path) + if identity is not None: + record = _load_terminal_json(path, identity) + if _terminal_record_valid(record, identity): + return True + base = state_path(board, task_id) + for prepared in base.parent.glob( + f"{base.stem}.run-{run_id}.terminal.prepared-*.json" + ): + identity = _terminal_evidence_identity(prepared) + if identity is None or identity.state != "prepared": + continue + record = _load_small_json(prepared) + if _terminal_evidence_valid(record, identity, "prepared"): + return True + for staged in base.parent.glob(".retire.*"): + snapshot = _open_terminal_recovery_snapshot(staged) + if snapshot is None: + continue + try: + authority = _staged_terminal_authority(staged, snapshot) + if authority is not None and authority[0] == TerminalIdentity( + board, + task_id, + run_id, + "pending", + ): + return True + finally: + snapshot.close() + return False def _artifact_gc_candidates(board_dir: Path) -> list[Path]: @@ -1590,6 +2412,8 @@ def _artifact_gc_candidates(board_dir: Path) -> list[Path]: "*.provider-*.result.json", "*.candidate-*.json", "*.terminal.committed.json", + "*.terminal.conflict-*.json", + ".retire.*", ) for pattern in patterns: candidates.extend(board_dir.glob(pattern)) @@ -1636,6 +2460,16 @@ def gc_lane_artifacts( board_removed = 0 entries: list[tuple[Path, os.stat_result]] = [] for path in _artifact_gc_candidates(board_dir): + if path.name.startswith(".retire."): + snapshot = _open_terminal_recovery_snapshot(path) + if snapshot is not None: + try: + if _staged_terminal_authority(path, snapshot) is not None: + # Accepted authority is first promoted/replayed by + # recovery; retention never deletes it directly. + continue + finally: + snapshot.close() try: file_stat = path.stat(follow_symlinks=False) except OSError: @@ -1969,9 +2803,10 @@ def execute_claim(board: str, task_id: str) -> None: identity = _terminal_identity(terminal_file) replayable = bool( identity is not None - and _terminal_record_valid( - _load_terminal_json(terminal_file, identity), - identity, + and _has_pending_finalization( + identity.board, + identity.task_id, + identity.run_id, ) ) if not replayable: @@ -2141,12 +2976,15 @@ def recover_orphans() -> None: if _external(task) and str(_task_value(task, "status", "")) == "running": task_id = str(_task_value(task, "id")) run_id = _task_value(task, "current_run_id", None) + if not isinstance(run_id, int): + continue if _has_pending_finalization(board, task_id, run_id): continue kanban_db.reclaim_task( conn, task_id, reason="direct CLI lane restarted; provider session will resume", + expected_run_id=run_id, ) BOARD_CORRUPTION_ERRORS.pop(board, None) except Exception as error: diff --git a/services/hermes/scripts/recover_cassandra_workers.py b/services/hermes/scripts/recover_cassandra_workers.py index ed0c12da..b0bb75c4 100755 --- a/services/hermes/scripts/recover_cassandra_workers.py +++ b/services/hermes/scripts/recover_cassandra_workers.py @@ -24,10 +24,12 @@ def recover_running_tasks(kanban_db: Any) -> list[str]: if str(getattr(task, "status", "")) != "running": continue task_id = str(getattr(task, "id", "")) - if not task_id or not kanban_db.reclaim_task( + run_id = getattr(task, "current_run_id", None) + if not task_id or not isinstance(run_id, int) or not kanban_db.reclaim_task( connection, task_id, reason=REASON, + expected_run_id=run_id, ): continue kanban_db.add_comment( diff --git a/testing/tests/test_hermes_cli_lanes.py b/testing/tests/test_hermes_cli_lanes.py index 66ba6175..160f2412 100644 --- a/testing/tests/test_hermes_cli_lanes.py +++ b/testing/tests/test_hermes_cli_lanes.py @@ -10,6 +10,8 @@ import os import signal import stat import sys +import threading +import time from contextlib import nullcontext from pathlib import Path from types import SimpleNamespace @@ -685,6 +687,7 @@ def test_accepted_result_survives_failed_finalization_and_replays_exact_run( return False task.status = "done" task.result = kwargs["result"] + task.completed_run_id = kwargs["expected_run_id"] task.current_run_id = None return True @@ -804,6 +807,78 @@ def test_restart_does_not_reclaim_an_exact_run_awaiting_finalization( assert reclaimed == [] +def test_orphan_recovery_passes_the_scanned_run_as_atomic_reclaim_guard( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + tasks = [ + SimpleNamespace( + id="t_guarded", + status="running", + assignee="cli-auto", + current_run_id=91, + ), + SimpleNamespace( + id="t_no_run", + status="running", + assignee="cli-auto", + current_run_id=None, + ), + ] + reclaimed = [] + + class Connection: + def close(self): + return None + + fake_db = SimpleNamespace( + list_boards=lambda include_archived=False: [{"slug": "cassandra"}], + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: Connection(), + list_tasks=lambda _conn: tasks, + reclaim_task=lambda _conn, task_id, **kwargs: ( + reclaimed.append((task_id, kwargs)) or True + ), + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) + monkeypatch.setattr(lanes, "recover_pending_finalizations", lambda: 0) + + lanes.recover_orphans() + + assert reclaimed == [ + ( + "t_guarded", + { + "reason": "direct CLI lane restarted; provider session will resume", + "expected_run_id": 91, + }, + ) + ] + + +def test_prepared_evidence_without_pending_still_pins_the_exact_run( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + pending, record = lanes._write_terminal_record( + lanes.state_path("cassandra", "t_prepared_pin"), + board="cassandra", + task_id="t_prepared_pin", + run_id=14, + structured=_completed_result("accepted and prepared"), + summary="accepted and prepared", + metadata={}, + ) + identity = lanes._terminal_identity(pending) + assert identity is not None + lanes._persist_prepared_evidence(identity, record) + pending.unlink() + + assert lanes._has_pending_finalization("cassandra", "t_prepared_pin", 14) is True + + def test_terminal_replay_never_crosses_into_a_replacement_run( tmp_path: Path, monkeypatch, @@ -851,8 +926,11 @@ def test_terminal_replay_never_crosses_into_a_replacement_run( assert lanes.recover_pending_finalizations() == 0 assert completions == [] assert not terminal_path.exists() - quarantined = list((state_root / "cassandra/quarantine").glob("*.quarantine")) - assert len(quarantined) == 1 + conflicts = list((state_root / "cassandra").glob("*.terminal.conflict-*.json")) + assert len(conflicts) == 1 + conflict = json.loads(conflicts[0].read_text(encoding="utf-8")) + assert conflict["result"] == _record["result"] + assert conflict["kanban_state"] == "conflict" def _completed_result(summary: str = "done") -> dict: @@ -867,6 +945,48 @@ def _completed_result(summary: str = "done") -> dict: } +def _pending_terminal_record(board: str, task_id: str, run_id: int, summary: str) -> dict: + structured = _completed_result(summary) + return { + "board": board, + "task_id": task_id, + "expected_run_id": run_id, + "result": json.dumps(structured, sort_keys=True), + "summary": summary, + "metadata": {}, + "kanban_state": "pending", + "recorded_at": lanes.utc_now(), + } + + +def _install_terminal_recovery_db(monkeypatch, task, completions, reclaims) -> None: + class Connection: + def close(self): + return None + + def complete_task(_conn, _task_id, **kwargs): + if task.status != "running" or task.current_run_id != kwargs["expected_run_id"]: + return False + task.status = "done" + task.result = kwargs["result"] + task.current_run_id = None + completions.append(kwargs["result"]) + return True + + def reclaim_task(_conn, _task_id, **kwargs): + reclaims.append(kwargs) + return False + + fake_db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: Connection(), + get_task=lambda _conn, _task_id: task, + complete_task=complete_task, + reclaim_task=reclaim_task, + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) + + def test_terminal_recovery_can_complete_the_exact_latest_ended_run( tmp_path: Path, monkeypatch, @@ -1185,13 +1305,19 @@ def test_malformed_exact_run_journal_is_quarantined_and_reclaimed( scoped_current_board=lambda _board: nullcontext(), connect=lambda board: Connection(), get_task=lambda _conn, _task_id: task, - reclaim_task=lambda *_args, **_kwargs: (reclaimed.append(True) or True), + reclaim_task=lambda *_args, **kwargs: (reclaimed.append(kwargs) or True), ) monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) assert lanes.recover_pending_finalizations() == 0 - assert reclaimed == [True] + assert reclaimed == [{ + "reason": ( + "terminal journal recovery failed (malformed-payload); " + "exact run may retry" + ), + "expected_run_id": 12, + }] assert not path.exists() assert lanes._has_pending_finalization("cassandra", "t_partial", 12) is False quarantined = list((state_root / "cassandra/quarantine").glob("*.quarantine")) @@ -1417,6 +1543,1042 @@ def test_quarantine_preserves_and_replays_an_atomic_replacement( assert not path.exists() +def test_recovery_quarantines_the_loaded_inode_not_a_before_open_replacement( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + path = lanes._terminal_path(lanes.state_path("cassandra", "t_gap"), 51) + path.parent.mkdir(parents=True) + path.write_bytes(b"malformed-original") + replacement = path.with_name("replacement.valid") + lanes.atomic_json( + replacement, + _pending_terminal_record("cassandra", "t_gap", 51, "preserved replacement"), + ) + task = SimpleNamespace( + id="t_gap", + status="running", + current_run_id=51, + result=None, + assignee="cli-auto", + ) + completions = [] + reclaims = [] + _install_terminal_recovery_db(monkeypatch, task, completions, reclaims) + real_quarantine = lanes._quarantine_terminal + swapped = {"value": False} + + def swap_before_quarantine_reopens(*args, **kwargs): + assert kwargs["snapshot"].prefix == b"malformed-original" + if not swapped["value"]: + os.replace(replacement, path) + swapped["value"] = True + return real_quarantine(*args, **kwargs) + + monkeypatch.setattr(lanes, "_quarantine_terminal", swap_before_quarantine_reopens) + + assert lanes.recover_pending_finalizations() == 1 + assert swapped["value"] is True + assert task.status == "done" + assert len(completions) == 1 + assert reclaims == [] + assert not path.exists() + committed = next(path.parent.glob("*.terminal.committed.json")) + assert json.loads(committed.read_text())["result"] == completions[0] + + +def test_recovery_replays_replacement_installed_after_quarantine_pins_source( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + path = lanes._terminal_path(lanes.state_path("cassandra", "t_after_open"), 52) + path.parent.mkdir(parents=True) + path.write_bytes(b"malformed-original") + replacement = path.with_name("replacement.valid") + lanes.atomic_json( + replacement, + _pending_terminal_record( + "cassandra", "t_after_open", 52, "replacement after open" + ), + ) + task = SimpleNamespace( + id="t_after_open", + status="running", + current_run_id=52, + result=None, + assignee="cli-auto", + ) + completions = [] + reclaims = [] + _install_terminal_recovery_db(monkeypatch, task, completions, reclaims) + real_fsync = lanes.os.fsync + swapped = {"value": False} + + def swap_after_quarantine_open(descriptor): + descriptor_stat = os.fstat(descriptor) + if stat.S_ISDIR(descriptor_stat.st_mode) and not swapped["value"]: + os.replace(replacement, path) + swapped["value"] = True + real_fsync(descriptor) + + monkeypatch.setattr(lanes.os, "fsync", swap_after_quarantine_open) + + assert lanes.recover_pending_finalizations() == 1 + assert swapped["value"] is True + assert task.status == "done" + assert len(completions) == 1 + assert reclaims == [] + assert not path.exists() + + +def test_recovery_survives_repeated_invalid_then_valid_replacements( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + path = lanes._terminal_path(lanes.state_path("cassandra", "t_churn"), 53) + path.parent.mkdir(parents=True) + path.write_bytes(b"malformed-zero") + second = path.with_name("replacement.invalid") + second.write_bytes(b"malformed-one") + valid = path.with_name("replacement.valid") + lanes.atomic_json( + valid, + _pending_terminal_record("cassandra", "t_churn", 53, "valid after churn"), + ) + task = SimpleNamespace( + id="t_churn", + status="running", + current_run_id=53, + result=None, + assignee="cli-auto", + ) + completions = [] + reclaims = [] + _install_terminal_recovery_db(monkeypatch, task, completions, reclaims) + real_quarantine = lanes._quarantine_terminal + replacements = [second, valid] + pinned = [] + + def churn_before_quarantine(*args, **kwargs): + pinned.append(kwargs["snapshot"].prefix) + if replacements: + os.replace(replacements.pop(0), path) + return real_quarantine(*args, **kwargs) + + monkeypatch.setattr(lanes, "_quarantine_terminal", churn_before_quarantine) + + assert lanes.recover_pending_finalizations() == 1 + assert pinned == [b"malformed-zero", b"malformed-one"] + assert replacements == [] + assert task.status == "done" + assert len(completions) == 1 + assert reclaims == [] + assert not path.exists() + diagnostics = list((path.parent / "quarantine").glob("*.quarantine")) + assert len(diagnostics) == 2 + + +def test_recovery_finalizes_the_valid_snapshot_it_classified_before_replacement( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + path, first = lanes._write_terminal_record( + lanes.state_path("cassandra", "t_valid_gap"), + board="cassandra", + task_id="t_valid_gap", + run_id=54, + structured=_completed_result("first valid snapshot"), + summary="first valid snapshot", + metadata={}, + ) + replacement = path.with_name("replacement.valid") + lanes.atomic_json( + replacement, + _pending_terminal_record( + "cassandra", "t_valid_gap", 54, "second valid replacement" + ), + ) + task = SimpleNamespace( + id="t_valid_gap", + status="running", + current_run_id=54, + result=None, + assignee="cli-auto", + ) + completions = [] + reclaims = [] + _install_terminal_recovery_db(monkeypatch, task, completions, reclaims) + real_finalize = lanes._finalize_terminal_record + swapped = {"value": False} + + def replace_before_finalize(*args, **kwargs): + assert kwargs["snapshot"].document == first + if not swapped["value"]: + os.replace(replacement, path) + swapped["value"] = True + return real_finalize(*args, **kwargs) + + monkeypatch.setattr(lanes, "_finalize_terminal_record", replace_before_finalize) + + assert lanes.recover_pending_finalizations() == 1 + assert task.result == first["result"] + assert completions == [first["result"]] + assert reclaims == [] + assert not path.exists() + conflicts = list(path.parent.glob("*.terminal.conflict-*.json")) + assert len(conflicts) == 1 + conflict = json.loads(conflicts[0].read_text()) + assert conflict["summary"] == "second valid replacement" + + +def test_replacement_staged_journal_promotes_replays_and_never_reexecutes( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + path = lanes._terminal_path(lanes.state_path("cassandra", "t_staged"), 55) + path.parent.mkdir(parents=True) + path.write_bytes(b"malformed source") + source_stat = path.stat(follow_symlinks=False) + staged_winner = path.with_name("staged-winner.tmp") + canonical_replacement = path.with_name("canonical-replacement.tmp") + first = _pending_terminal_record( + "cassandra", "t_staged", 55, "staged accepted result" + ) + second = _pending_terminal_record( + "cassandra", "t_staged", 55, "later canonical replacement" + ) + lanes.atomic_json(staged_winner, first) + lanes.atomic_json(canonical_replacement, second) + directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) + descriptor = os.open(path.parent, directory_flags) + real_rename = lanes._rename_noreplace + rename_steps = [] + + def force_replacement_staging(source, destination, **kwargs): + rename_steps.append((source, destination)) + if source == path.name: + os.replace(staged_winner, path) + elif source.startswith(".retire."): + os.replace(canonical_replacement, path) + return real_rename(source, destination, **kwargs) + + try: + with monkeypatch.context() as race: + race.setattr(lanes, "_rename_noreplace", force_replacement_staging) + outcome = lanes._retire_terminal_entry( + path, + source_stat, + board_descriptor=descriptor, + quarantine_descriptor=descriptor, + ) + finally: + os.close(descriptor) + + assert outcome == "replacement-staged" + assert len(rename_steps) == 2 + staged = list(path.parent.glob(".retire.*")) + assert len(staged) == 1 + staged_snapshot = lanes._open_terminal_recovery_snapshot(staged[0]) + assert staged_snapshot is not None + try: + authority = lanes._staged_terminal_authority(staged[0], staged_snapshot) + assert authority is not None + assert authority[0].run_id == 55 + assert authority[1] == path + finally: + staged_snapshot.close() + assert lanes._has_pending_finalization("cassandra", "t_staged", 55) is True + assert lanes.gc_lane_artifacts( + now=staged[0].stat().st_mtime + 3600, + max_age_seconds=0, + max_count=0, + max_bytes=0, + ) == 0 + assert staged[0].exists() + + task = SimpleNamespace( + id="t_staged", + status="running", + current_run_id=55, + completed_run_id=None, + result=None, + assignee="cli-auto", + ) + completions = [] + + class Connection: + def close(self): + return None + + def complete_task(_conn, _task_id, **kwargs): + task.status = "done" + task.result = kwargs["result"] + task.completed_run_id = kwargs["expected_run_id"] + task.current_run_id = None + completions.append(kwargs["result"]) + return True + + fake_db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: Connection(), + get_task=lambda _conn, _task_id: task, + complete_task=complete_task, + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) + + assert lanes.recover_pending_finalizations() == 1 + assert completions == [first["result"]] + assert not list(path.parent.glob(".retire.*")) + assert not path.exists() + conflicts = list(path.parent.glob("*.terminal.conflict-*.json")) + assert len(conflicts) == 1 + assert json.loads(conflicts[0].read_text())["result"] == second["result"] + assert lanes.recover_pending_finalizations() == 0 + assert completions == [first["result"]] + + +def test_repeated_retirement_staging_preserves_every_valid_result_once( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + canonical = lanes._terminal_path( + lanes.state_path("cassandra", "t_staged_repeat"), 57 + ) + canonical.parent.mkdir(parents=True) + path_digest = hashlib.sha256(canonical.name.encode("utf-8")).hexdigest()[:16] + staged = canonical.parent / f".retire.{path_digest}.{'a' * 16}.0" + records = [ + _pending_terminal_record( + "cassandra", "t_staged_repeat", 57, f"accepted result {number}" + ) + for number in range(3) + ] + lanes.atomic_json(staged, records[0]) + second = canonical.with_name("second-staged-replacement.tmp") + third = canonical.with_name("third-staged-replacement.tmp") + lanes.atomic_json(second, records[1]) + lanes.atomic_json(third, records[2]) + real_rename = lanes._rename_noreplace + retirement_steps = [] + + def replace_each_retirement_generation(source, destination, **kwargs): + if source == staged.name and destination.startswith(".retire."): + retirement_steps.append("move") + os.replace(second, staged) + elif source.startswith(".retire.") and destination == staged.name: + retirement_steps.append("restore") + os.replace(third, staged) + return real_rename(source, destination, **kwargs) + + with monkeypatch.context() as race: + race.setattr(lanes, "_rename_noreplace", replace_each_retirement_generation) + assert lanes._recover_retirement_staging() == 0 + + assert retirement_steps == ["move", "restore"] + staged_generations = list(canonical.parent.glob(".retire.*")) + assert len(staged_generations) == 2 + for generation in staged_generations: + snapshot = lanes._open_terminal_recovery_snapshot(generation) + assert snapshot is not None + try: + authority = lanes._staged_terminal_authority(generation, snapshot) + assert authority is not None + assert authority[1] == canonical + finally: + snapshot.close() + + task = SimpleNamespace( + id="t_staged_repeat", + status="running", + current_run_id=57, + completed_run_id=None, + result=None, + assignee="cli-auto", + ) + completions = [] + reclaims = [] + _install_terminal_recovery_db(monkeypatch, task, completions, reclaims) + + assert lanes.recover_pending_finalizations() == 1 + assert len(completions) == 1 + assert reclaims == [] + assert not list(canonical.parent.glob(".retire.*")) + conflicts = list(canonical.parent.glob("*.terminal.conflict-*.json")) + assert len(conflicts) == 2 + preserved_results = {task.result} + preserved_results.update(json.loads(path.read_text())["result"] for path in conflicts) + assert preserved_results == {record["result"] for record in records} + assert lanes.recover_pending_finalizations() == 0 + assert len(completions) == 1 + + +def test_untrusted_retirement_staging_is_bounded_by_artifact_gc( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + board = lanes.STATE_ROOT / "cassandra" + board.mkdir(parents=True) + for sequence in range(10): + (board / f".retire.{'a' * 16}.{'b' * 16}.{sequence}").write_bytes( + b"not terminal authority" + ) + (board / f".retire.{'c' * 32}.0").write_bytes(b"legacy hidden artifact") + + removed = lanes.gc_lane_artifacts( + now=time.time(), + max_age_seconds=3600, + max_count=2, + max_bytes=1024 * 1024, + ) + + assert removed == 9 + assert len(list(board.glob(".retire.*"))) == 2 + + +def test_legacy_valid_retirement_staging_is_promoted_without_reexecution( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + board = lanes.STATE_ROOT / "cassandra" + board.mkdir(parents=True) + staged = board / f".retire.{'d' * 32}.0" + record = _pending_terminal_record( + "cassandra", "t_legacy_staged", 56, "legacy staged result" + ) + lanes.atomic_json(staged, record) + task = SimpleNamespace( + id="t_legacy_staged", + status="running", + current_run_id=56, + completed_run_id=None, + result=None, + assignee="cli-auto", + ) + completions = [] + reclaims = [] + _install_terminal_recovery_db(monkeypatch, task, completions, reclaims) + + assert lanes._has_pending_finalization( + "cassandra", "t_legacy_staged", 56 + ) is True + assert lanes.recover_pending_finalizations() == 1 + assert completions == [record["result"]] + assert reclaims == [] + assert not staged.exists() + assert lanes.recover_pending_finalizations() == 0 + assert completions == [record["result"]] + + +def test_done_result_bytes_do_not_authorize_a_different_run_journal( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + path, stale = lanes._write_terminal_record( + lanes.state_path("cassandra", "t_provenance"), + board="cassandra", + task_id="t_provenance", + run_id=1, + structured=_completed_result("same bytes"), + summary="same bytes", + metadata={}, + ) + task = SimpleNamespace( + id="t_provenance", + status="done", + current_run_id=None, + completed_run_id=2, + result=stale["result"], + assignee="cli-auto", + ) + + class Connection: + def close(self): + return None + + fake_db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: Connection(), + get_task=lambda _conn, _task_id: task, + complete_task=lambda *_args, **_kwargs: pytest.fail( + "a done task with different run provenance must not be completed" + ), + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) + + assert lanes.recover_pending_finalizations() == 0 + assert not path.exists() + assert not list(path.parent.glob("*.terminal.committed.json")) + conflicts = list(path.parent.glob("*.terminal.conflict-*.json")) + assert len(conflicts) == 1 + conflict = json.loads(conflicts[0].read_text()) + assert conflict["expected_run_id"] == 1 + assert conflict["result"] == task.result + + +@pytest.mark.parametrize( + "swap_point", + [ + "during-complete", + "after-db-before-promote", + "before-committed-create", + "before-pending-retire", + ], +) +def test_terminal_first_writer_preserves_valid_replacement_conflicts( + tmp_path: Path, + monkeypatch, + swap_point: str, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + path, old_record = lanes._write_terminal_record( + lanes.state_path("cassandra", "t_first_writer"), + board="cassandra", + task_id="t_first_writer", + run_id=27, + structured=_completed_result("first result"), + summary="first result", + metadata={"writer": "first"}, + ) + replacement_structured = _completed_result("replacement result") + replacement_record = { + "board": "cassandra", + "task_id": "t_first_writer", + "expected_run_id": 27, + "result": json.dumps(replacement_structured, sort_keys=True), + "summary": "replacement result", + "metadata": {"writer": "replacement"}, + "kanban_state": "pending", + "recorded_at": lanes.utc_now(), + } + replacement = path.with_name(f"replacement-{swap_point}.tmp") + lanes.atomic_json(replacement, replacement_record) + swapped = {"value": False} + + def swap_pending(): + if not swapped["value"]: + os.replace(replacement, path) + swapped["value"] = True + + task = SimpleNamespace( + id="t_first_writer", + status="running", + result=None, + current_run_id=27, + assignee="cli-auto", + ) + + class Connection: + def close(self): + return None + + def complete_task(_conn, _task_id, **kwargs): + task.status = "done" + task.result = kwargs["result"] + task.current_run_id = None + if swap_point == "during-complete": + swap_pending() + return True + + fake_db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: Connection(), + get_task=lambda _conn, _task_id: task, + complete_task=complete_task, + ) + if swap_point == "after-db-before-promote": + real_promote = lanes._promote_prepared_evidence + + def swap_then_promote(*args, **kwargs): + swap_pending() + return real_promote(*args, **kwargs) + + monkeypatch.setattr(lanes, "_promote_prepared_evidence", swap_then_promote) + elif swap_point == "before-committed-create": + real_create = lanes._write_json_noreplace + + def swap_before_committed(path_arg, value): + if path_arg.name.endswith(".terminal.committed.json"): + swap_pending() + return real_create(path_arg, value) + + monkeypatch.setattr(lanes, "_write_json_noreplace", swap_before_committed) + elif swap_point == "before-pending-retire": + real_retire = lanes._retire_snapshot + + def swap_before_retire(path_arg, snapshot): + if path_arg == path: + swap_pending() + return real_retire(path_arg, snapshot) + + monkeypatch.setattr(lanes, "_retire_snapshot", swap_before_retire) + + assert lanes._finalize_terminal_record(fake_db, path, old_record) == "committed" + + assert swapped["value"] is True + assert task.result == old_record["result"] + assert not path.exists() + committed = list(path.parent.glob("*.terminal.committed.json")) + conflicts = list(path.parent.glob("*.terminal.conflict-*.json")) + prepared = list(path.parent.glob("*.terminal.prepared-*.json")) + assert len(committed) == 1 + assert len(conflicts) == 1 + assert prepared == [] + assert committed[0].stat().st_mode & 0o777 == 0o600 + assert conflicts[0].stat().st_mode & 0o777 == 0o600 + committed_document = json.loads(committed[0].read_text(encoding="utf-8")) + conflict_document = json.loads(conflicts[0].read_text(encoding="utf-8")) + assert committed_document["result"] == old_record["result"] + assert committed_document["kanban_state"] == "committed" + assert conflict_document["result"] == replacement_record["result"] + assert conflict_document["metadata"] == {"writer": "replacement"} + assert conflict_document["kanban_state"] == "conflict" + + +def test_terminal_commit_directory_fsync_failure_recovers_from_prepared_evidence( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + path, record = lanes._write_terminal_record( + lanes.state_path("cassandra", "t_post_db_fsync"), + board="cassandra", + task_id="t_post_db_fsync", + run_id=28, + structured=_completed_result("durable DB winner"), + summary="durable DB winner", + metadata={}, + ) + task = SimpleNamespace( + id="t_post_db_fsync", + status="running", + result=None, + current_run_id=28, + completed_run_id=None, + assignee="cli-auto", + ) + replacement_structured = _completed_result("replacement after DB commit") + replacement_record = { + "board": "cassandra", + "task_id": "t_post_db_fsync", + "expected_run_id": 28, + "result": json.dumps(replacement_structured, sort_keys=True), + "summary": "replacement after DB commit", + "metadata": {"writer": "replacement"}, + "kanban_state": "pending", + "recorded_at": lanes.utc_now(), + } + replacement = path.with_name("post-db-fsync-replacement.tmp") + lanes.atomic_json(replacement, replacement_record) + + class Connection: + def close(self): + return None + + def complete_task(_conn, _task_id, **kwargs): + task.status = "done" + task.result = kwargs["result"] + task.completed_run_id = kwargs["expected_run_id"] + task.current_run_id = None + os.replace(replacement, path) + return True + + fake_db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: Connection(), + get_task=lambda _conn, _task_id: task, + complete_task=complete_task, + ) + real_fsync = lanes.os.fsync + failed = {"value": False} + + def fail_committed_directory_fsync(descriptor): + if ( + stat.S_ISDIR(os.fstat(descriptor).st_mode) + and list(path.parent.glob("*.terminal.committed.json")) + and not failed["value"] + ): + failed["value"] = True + raise OSError(errno.ENOSPC, "post-DB committed directory fsync failed") + real_fsync(descriptor) + + monkeypatch.setattr(lanes.os, "fsync", fail_committed_directory_fsync) + + with pytest.raises(OSError, match="post-DB committed"): + lanes._finalize_terminal_record(fake_db, path, record) + + assert task.status == "done" + assert task.result == record["result"] + assert path.exists() + assert json.loads(path.read_text(encoding="utf-8"))["result"] == ( + replacement_record["result"] + ) + assert len(list(path.parent.glob("*.terminal.prepared-*.json"))) == 1 + + monkeypatch.setattr(lanes.os, "fsync", real_fsync) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) + assert lanes.recover_pending_finalizations() == 1 + assert not path.exists() + assert list(path.parent.glob("*.terminal.prepared-*.json")) == [] + committed = list(path.parent.glob("*.terminal.committed.json")) + conflicts = list(path.parent.glob("*.terminal.conflict-*.json")) + assert len(committed) == 1 + assert len(conflicts) == 1 + assert json.loads(committed[0].read_text())["result"] == record["result"] + assert json.loads(conflicts[0].read_text())["result"] == replacement_record["result"] + + +def test_unknown_db_completion_outcome_replays_from_prepared_evidence( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + path, record = lanes._write_terminal_record( + lanes.state_path("cassandra", "t_unknown_db_outcome"), + board="cassandra", + task_id="t_unknown_db_outcome", + run_id=35, + structured=_completed_result("DB committed before transport error"), + summary="DB committed before transport error", + metadata={}, + ) + task = SimpleNamespace( + id="t_unknown_db_outcome", + status="running", + result=None, + current_run_id=35, + completed_run_id=None, + assignee="cli-auto", + ) + fail_once = {"value": True} + + class Connection: + def close(self): + return None + + def complete_task(_conn, _task_id, **kwargs): + task.status = "done" + task.result = kwargs["result"] + task.completed_run_id = kwargs["expected_run_id"] + task.current_run_id = None + if fail_once["value"]: + fail_once["value"] = False + raise RuntimeError("transport failed after DB commit") + return True + + fake_db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: Connection(), + get_task=lambda _conn, _task_id: task, + complete_task=complete_task, + ) + + with pytest.raises(RuntimeError, match="after DB commit"): + lanes._finalize_terminal_record(fake_db, path, record) + + assert task.status == "done" + assert path.exists() + assert len(list(path.parent.glob("*.terminal.prepared-*.json"))) == 1 + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) + assert lanes.recover_pending_finalizations() == 1 + assert not path.exists() + assert list(path.parent.glob("*.terminal.prepared-*.json")) == [] + committed = list(path.parent.glob("*.terminal.committed.json")) + assert len(committed) == 1 + assert json.loads(committed[0].read_text())["result"] == record["result"] + + +def test_pending_retirement_fsync_failure_keeps_db_winner_committed( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + path, record = lanes._write_terminal_record( + lanes.state_path("cassandra", "t_retire_fsync"), + board="cassandra", + task_id="t_retire_fsync", + run_id=33, + structured=_completed_result("retirement fsync winner"), + summary="retirement fsync winner", + metadata={}, + ) + task = SimpleNamespace( + id="t_retire_fsync", + status="running", + result=None, + current_run_id=33, + assignee="cli-auto", + ) + + class Connection: + def close(self): + return None + + def complete_task(_conn, _task_id, **kwargs): + task.status = "done" + task.result = kwargs["result"] + task.current_run_id = None + return True + + fake_db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: Connection(), + get_task=lambda _conn, _task_id: task, + complete_task=complete_task, + ) + real_fsync = lanes.os.fsync + failed = {"value": False} + + def fail_pending_retirement_fsync(descriptor): + if ( + stat.S_ISDIR(os.fstat(descriptor).st_mode) + and not path.exists() + and list(path.parent.glob("*.terminal.committed.json")) + and not failed["value"] + ): + failed["value"] = True + raise OSError(errno.ENOSPC, "pending retirement fsync failed") + real_fsync(descriptor) + + monkeypatch.setattr(lanes.os, "fsync", fail_pending_retirement_fsync) + + assert lanes._finalize_terminal_record(fake_db, path, record) == "committed" + assert failed["value"] is True + assert task.status == "done" + assert task.result == record["result"] + assert not path.exists() + assert len(list(path.parent.glob("*.terminal.committed.json"))) == 1 + assert list(path.parent.glob("*.terminal.prepared-*.json")) == [] + + +def test_committed_first_writer_is_never_overwritten_by_a_db_winner( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + state_file = lanes.state_path("cassandra", "t_committed_collision") + pending, winner = lanes._write_terminal_record( + state_file, + board="cassandra", + task_id="t_committed_collision", + run_id=34, + structured=_completed_result("DB winner"), + summary="DB winner", + metadata={"writer": "db"}, + ) + committed_path = lanes._terminal_path(state_file, 34, "committed") + first_writer = { + "board": "cassandra", + "task_id": "t_committed_collision", + "expected_run_id": 34, + "result": json.dumps(_completed_result("evidence first writer"), sort_keys=True), + "summary": "evidence first writer", + "metadata": {"writer": "evidence"}, + "kanban_state": "committed", + "recorded_at": lanes.utc_now(), + } + lanes.atomic_json(committed_path, first_writer) + task = SimpleNamespace( + id="t_committed_collision", + status="running", + result=None, + current_run_id=34, + assignee="cli-auto", + ) + + class Connection: + def close(self): + return None + + def complete_task(_conn, _task_id, **kwargs): + task.status = "done" + task.result = kwargs["result"] + task.current_run_id = None + return True + + fake_db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: Connection(), + get_task=lambda _conn, _task_id: task, + complete_task=complete_task, + ) + + assert lanes._finalize_terminal_record(fake_db, pending, winner) == "committed" + assert task.result == winner["result"] + assert json.loads(committed_path.read_text())["result"] == first_writer["result"] + conflicts = list(pending.parent.glob("*.terminal.conflict-*.json")) + assert len(conflicts) == 1 + assert json.loads(conflicts[0].read_text())["result"] == winner["result"] + assert not pending.exists() + + +def test_concurrent_duplicate_terminal_finalizers_are_idempotent( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + path, record = lanes._write_terminal_record( + lanes.state_path("cassandra", "t_duplicate"), + board="cassandra", + task_id="t_duplicate", + run_id=29, + structured=_completed_result("same accepted result"), + summary="same accepted result", + metadata={}, + ) + task_state = { + "status": "running", + "result": None, + "current_run_id": 29, + "completed_run_id": None, + } + state_lock = threading.Lock() + readers = threading.Barrier(2) + completions = [] + + class Connection: + def close(self): + return None + + def get_task(_conn, _task_id): + with state_lock: + snapshot = SimpleNamespace( + id="t_duplicate", + assignee="cli-auto", + **task_state, + ) + if snapshot.status == "running": + readers.wait(timeout=5) + return snapshot + + def complete_task(_conn, _task_id, **kwargs): + with state_lock: + if task_state["status"] != "running": + return False + task_state.update( + status="done", + result=kwargs["result"], + current_run_id=None, + completed_run_id=kwargs["expected_run_id"], + ) + completions.append(kwargs["result"]) + return True + + fake_db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: Connection(), + get_task=get_task, + complete_task=complete_task, + ) + outcomes = [] + errors = [] + + def finalize(): + try: + outcomes.append(lanes._finalize_terminal_record(fake_db, path, record)) + except Exception as error: # pragma: no cover - assertion reports detail + errors.append(error) + + workers = [threading.Thread(target=finalize) for _index in range(2)] + for worker in workers: + worker.start() + for worker in workers: + worker.join(timeout=10) + + assert all(not worker.is_alive() for worker in workers) + assert errors == [] + assert outcomes == ["committed", "committed"] + assert len(completions) == 1 + assert not path.exists() + assert len(list(path.parent.glob("*.terminal.committed.json"))) == 1 + assert list(path.parent.glob("*.terminal.conflict-*.json")) == [] + assert list(path.parent.glob("*.terminal.prepared-*.json")) == [] + + +def test_recovery_uses_first_durable_prepared_result_as_db_writer( + tmp_path: Path, + monkeypatch, +): + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + state_file = lanes.state_path("cassandra", "t_prepared_order") + pending, first = lanes._write_terminal_record( + state_file, + board="cassandra", + task_id="t_prepared_order", + run_id=30, + structured=_completed_result("first durable result"), + summary="first durable result", + metadata={"writer": "first"}, + ) + identity = lanes._terminal_identity(pending) + assert identity is not None + first_prepared = lanes._persist_prepared_evidence(identity, first) + os.utime(first_prepared, ns=(100, 100)) + second_structured = _completed_result("second durable result") + second = { + "board": "cassandra", + "task_id": "t_prepared_order", + "expected_run_id": 30, + "result": json.dumps(second_structured, sort_keys=True), + "summary": "second durable result", + "metadata": {"writer": "second"}, + "kanban_state": "pending", + "recorded_at": lanes.utc_now(), + } + second_prepared = lanes._persist_prepared_evidence(identity, second) + os.utime(second_prepared, ns=(200, 200)) + lanes.atomic_json(pending, second) + task = SimpleNamespace( + id="t_prepared_order", + status="running", + result=None, + current_run_id=30, + assignee="cli-auto", + ) + completions = [] + + class Connection: + def close(self): + return None + + def complete_task(_conn, _task_id, **kwargs): + if task.status != "running": + return False + task.status = "done" + task.result = kwargs["result"] + task.current_run_id = None + completions.append(kwargs["result"]) + return True + + fake_db = SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: Connection(), + get_task=lambda _conn, _task_id: task, + complete_task=complete_task, + ) + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) + + assert lanes.recover_pending_finalizations() == 1 + assert completions == [first["result"]] + assert task.result == first["result"] + assert not pending.exists() + assert list(pending.parent.glob("*.terminal.prepared-*.json")) == [] + committed = list(pending.parent.glob("*.terminal.committed.json")) + conflicts = list(pending.parent.glob("*.terminal.conflict-*.json")) + assert len(committed) == 1 + assert len(conflicts) == 1 + assert json.loads(committed[0].read_text())["result"] == first["result"] + assert json.loads(conflicts[0].read_text())["result"] == second["result"] + + def test_invalid_utf8_journal_quarantines_and_does_not_stop_later_replay( tmp_path: Path, monkeypatch, @@ -1651,6 +2813,7 @@ def test_artifact_gc_prunes_by_age_without_touching_pending_journals( board / "t.run-1.candidate-1.json", board / "t.run-1.provider-1.result.json", board / "t.run-1.terminal.committed.json", + board / f"t.run-1.terminal.conflict-{'a' * 32}.json", quarantine / "t.invalid.1234.quarantine", ] for artifact in old_artifacts: @@ -1659,10 +2822,13 @@ def test_artifact_gc_prunes_by_age_without_touching_pending_journals( quarantine_symlink.symlink_to(tmp_path / "missing-target") pending = board / "t.run-1.terminal.pending.json" pending.write_text("{}", encoding="utf-8") + prepared = board / f"t.run-1.terminal.prepared-{'b' * 32}.json" + prepared.write_text("{}", encoding="utf-8") old_time = 100.0 for artifact in old_artifacts: os.utime(artifact, (old_time, old_time)) os.utime(pending, (old_time, old_time)) + os.utime(prepared, (old_time, old_time)) assert lanes.gc_lane_artifacts( now=1000.0, max_age_seconds=10, max_count=100, max_bytes=10000 @@ -1670,6 +2836,7 @@ def test_artifact_gc_prunes_by_age_without_touching_pending_journals( assert not any(artifact.exists() for artifact in old_artifacts) assert not quarantine_symlink.is_symlink() assert pending.exists() + assert prepared.exists() def test_artifact_gc_prunes_oldest_by_count_and_total_bytes( diff --git a/testing/tests/test_hermes_worker_recovery.py b/testing/tests/test_hermes_worker_recovery.py index 77307bd3..359bbc05 100644 --- a/testing/tests/test_hermes_worker_recovery.py +++ b/testing/tests/test_hermes_worker_recovery.py @@ -42,7 +42,7 @@ class FakeKanban: self.tasks = tasks self.has_board = board_exists self.connection = FakeConnection() - self.reclaimed: list[tuple[str, str]] = [] + self.reclaimed: list[tuple[str, str, int]] = [] self.comments: list[tuple[str, str, str]] = [] def board_exists(self, board: str) -> bool: @@ -61,9 +61,16 @@ class FakeKanban: assert connection is self.connection return self.tasks - def reclaim_task(self, connection, task_id: str, *, reason: str) -> bool: + def reclaim_task( + self, + connection, + task_id: str, + *, + reason: str, + expected_run_id: int, + ) -> bool: assert connection is self.connection - self.reclaimed.append((task_id, reason)) + self.reclaimed.append((task_id, reason, expected_run_id)) return task_id != "t_race" def add_comment(self, connection, task_id: str, author: str, body: str) -> None: @@ -75,16 +82,17 @@ def test_recover_running_tasks_requeues_only_claimed_workers() -> None: module = _load_module() kanban = FakeKanban( [ - SimpleNamespace(id="t_running", status="running"), + SimpleNamespace(id="t_running", status="running", current_run_id=10), SimpleNamespace(id="t_done", status="done"), - SimpleNamespace(id="t_race", status="running"), + SimpleNamespace(id="t_race", status="running", current_run_id=11), + SimpleNamespace(id="t_missing_run", status="running", current_run_id=None), ] ) assert module.recover_running_tasks(kanban) == ["t_running"] assert kanban.reclaimed == [ - ("t_running", module.REASON), - ("t_race", module.REASON), + ("t_running", module.REASON, 10), + ("t_race", module.REASON, 11), ] assert kanban.comments == [ (