hermes: bind terminal commit to journal inode

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,

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,

services/hermes/scripts/cli_lane_runner.py

@@ -163,6 +163,27 @@ class TerminalSnapshot:
         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()
 
@@ -528,6 +549,126 @@ def _open_small_json_snapshot(path: Path) -> TerminalSnapshot | None:
             os.close(directory)
 
 
+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,
@@ -1483,7 +1624,13 @@ def _persist_conflict_evidence(
     return path
 
 
-def _quarantine_terminal(path: Path, identity: TerminalIdentity | None, reason: str) -> 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:
@@ -1508,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,
@@ -1540,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
@@ -1632,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 "
@@ -1672,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,
             )
         )
 
@@ -1859,15 +2027,23 @@ def _finalize_terminal_record(
     kanban_db: Any,
     path: Path,
     _record: dict[str, Any] | None = None,
+    *,
+    snapshot: TerminalRecoverySnapshot | None = None,
 ) -> str:
-    """Commit one exact-run terminal journal, or leave it safely pending."""
+    """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"
-    snapshot = _open_terminal_snapshot(path, identity)
-    if snapshot is None:
+    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 = snapshot.document
+    document = pinned.document
     try:
         if not _terminal_record_valid(document):
             return "invalid"
@@ -1881,7 +2057,7 @@ def _finalize_terminal_record(
                 document,
                 prepared,
             )
-            retirement = _retire_snapshot_after_db(path, snapshot)
+            retirement = _retire_snapshot_after_db(path, pinned)
         elif outcome == "stale":
             _persist_conflict_evidence(
                 identity,
@@ -1889,13 +2065,13 @@ def _finalize_terminal_record(
                 "valid terminal result no longer matches the authoritative task run",
             )
             _discard_evidence(prepared)
-            retirement = _retire_snapshot_after_db(path, snapshot)
+            retirement = _retire_snapshot_after_db(path, pinned)
             outcome = "conflict"
             evidence_state = "conflict"
         else:
             return outcome
     finally:
-        snapshot.close()
+        pinned.close()
     if outcome == "committed" and retirement not in {"retired", "missing"}:
         _resolve_pending_after_winner(path, identity, document)
     if evidence_state == "conflict":
@@ -1908,21 +2084,22 @@ def _finalize_terminal_record(
     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:
@@ -2002,46 +2179,70 @@ def recover_pending_finalizations() -> int:
     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
 
 

testing/tests/test_hermes_cli_lanes.py

@@ -893,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,
@@ -1211,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"))
@@ -1443,6 +1491,198 @@ 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",
     [