diff --git a/dockerfiles/hermes-execution-safety-regression.py b/dockerfiles/hermes-execution-safety-regression.py index a77ca92d..23a621e2 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 @@ -208,6 +209,147 @@ class AutomaticDecompositionSafetyTests(unittest.TestCase): ) self.assertIn(self._status(task_id), {"blocked", "triage"}) + 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..70fc2f47 100644 --- a/dockerfiles/patch-hermes-execution-safety.py +++ b/dockerfiles/patch-hermes-execution-safety.py @@ -124,6 +124,166 @@ db = replace_once( "transactional exact ended-run completion", ) +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", +) + 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..9652f035 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,10 @@ _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$" +) _SAFE_BOARD = re.compile(r"^[a-zA-Z0-9_.-]+$") @@ -418,6 +458,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 +501,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 +517,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 +532,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 +805,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 +813,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 ) @@ -1183,7 +1409,7 @@ def _retire_terminal_entry( f"{path.name}\0{source_stat.st_dev:x}\0{source_stat.st_ino:x}".encode("utf-8") ).hexdigest()[:32] for sequence in range(32): - staging = f".retire.{path.name}.{identity}.{sequence}" + staging = f".retire.{identity}.{sequence}" try: _rename_noreplace( path.name, @@ -1222,7 +1448,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 +1655,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 +1699,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 +1799,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 +1839,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 +1858,85 @@ 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) -> 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, + ) + + +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) @@ -1490,78 +1968,281 @@ 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 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 _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 + 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 +2254,26 @@ 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 + return False def _artifact_gc_candidates(board_dir: Path) -> list[Path]: @@ -1590,6 +2283,7 @@ def _artifact_gc_candidates(board_dir: Path) -> list[Path]: "*.provider-*.result.json", "*.candidate-*.json", "*.terminal.committed.json", + "*.terminal.conflict-*.json", ) for pattern in patterns: candidates.extend(board_dir.glob(pattern)) @@ -1969,9 +2663,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: diff --git a/testing/tests/test_hermes_cli_lanes.py b/testing/tests/test_hermes_cli_lanes.py index 66ba6175..86586d55 100644 --- a/testing/tests/test_hermes_cli_lanes.py +++ b/testing/tests/test_hermes_cli_lanes.py @@ -10,6 +10,7 @@ import os import signal import stat import sys +import threading from contextlib import nullcontext from pathlib import Path from types import SimpleNamespace @@ -804,6 +805,28 @@ def test_restart_does_not_reclaim_an_exact_run_awaiting_finalization( assert reclaimed == [] +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 +874,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 +893,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 +1253,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 +1491,747 @@ 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" + + +@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, + 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.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, + 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.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, + } + 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, + ) + 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 +2466,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 +2475,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 +2489,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(