hermes: bind terminal commit to journal inode

dockerfiles/hermes-execution-safety-regression.py

@@ -161,16 +161,36 @@ class AutomaticDecompositionSafetyTests(unittest.TestCase):
             )
         )
 
-        self.assertTrue(
-            kanban_db.complete_task(
-                self.connection,
-                task_id,
-                result="durable terminal result",
-                summary="durable terminal result",
-                replay_ended_run_id=run_id,
+        with mock.patch.object(
+            kanban_db, "_fire_kanban_lifecycle_hook"
+        ) as lifecycle:
+            self.assertTrue(
+                kanban_db.complete_task(
+                    self.connection,
+                    task_id,
+                    result="durable terminal result",
+                    summary="durable terminal result",
+                    replay_ended_run_id=run_id,
+                )
             )
-        )
-        self.assertEqual(self._status(task_id), "done")
+        lifecycle.assert_called_once()
+        self.assertEqual(lifecycle.call_args.args[:2], ("kanban_task_completed", task_id))
+        self.assertEqual(lifecycle.call_args.kwargs["run_id"], run_id)
+        task = kanban_db.get_task(self.connection, task_id)
+        self.assertEqual(task.status, "done")
+        self.assertEqual(task.completed_run_id, run_id)
+        runs = kanban_db.list_runs(self.connection, task_id)
+        self.assertEqual(len(runs), 1)
+        self.assertEqual(runs[0].id, run_id)
+        self.assertEqual(runs[0].status, "done")
+        self.assertEqual(runs[0].outcome, "completed")
+        self.assertEqual(runs[0].summary, "durable terminal result")
+        completed_events = self.connection.execute(
+            "SELECT run_id FROM task_events "
+            "WHERE task_id = ? AND kind = 'completed' ORDER BY id",
+            (task_id,),
+        ).fetchall()
+        self.assertEqual([row["run_id"] for row in completed_events], [run_id])
 
     def test_older_ended_run_cannot_complete_over_a_replacement(self) -> None:
         task_id = kanban_db.create_task(self.connection, title="replacement guard")
@@ -209,6 +229,38 @@ class AutomaticDecompositionSafetyTests(unittest.TestCase):
         )
         self.assertIn(self._status(task_id), {"blocked", "triage"})
 
+    def test_equal_result_bytes_remain_bound_to_the_completing_run(self) -> None:
+        task_id = kanban_db.create_task(self.connection, title="same result retries")
+        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+        first_run = kanban_db.get_task(self.connection, task_id).current_run_id
+        self.assertTrue(
+            kanban_db.block_task(
+                self.connection,
+                task_id,
+                reason="first attempt ended",
+                expected_run_id=first_run,
+            )
+        )
+        self.assertTrue(kanban_db.unblock_task(self.connection, task_id))
+        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+        second_run = kanban_db.get_task(self.connection, task_id).current_run_id
+        self.assertTrue(
+            kanban_db.complete_task(
+                self.connection,
+                task_id,
+                result="identical result bytes",
+                summary="second run wins",
+                expected_run_id=second_run,
+            )
+        )
+
+        task = kanban_db.get_task(self.connection, task_id)
+        self.assertEqual(task.completed_run_id, second_run)
+        self.assertNotEqual(task.completed_run_id, first_run)
+        runs = {run.id: run for run in kanban_db.list_runs(self.connection, task_id)}
+        self.assertEqual(runs[first_run].outcome, "blocked")
+        self.assertEqual(runs[second_run].outcome, "completed")
+
     def test_exact_run_reclaim_succeeds_for_the_authoritative_run(self) -> None:
         task_id = kanban_db.create_task(self.connection, title="recover exact run")
         self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))

dockerfiles/patch-hermes-execution-safety.py

@@ -72,9 +72,11 @@ complete_branch_after = '''    # Journal recovery may arrive after a legacy erro
     # accepted run. Verify that it is still the latest run in this same
     # transaction; a replacement attempt must make the replay fail closed.
     with write_txn(conn):
+        replayed_run_id = None
         if expected_run_id is not None and replay_ended_run_id is not None:
             raise ValueError("expected_run_id and replay_ended_run_id are mutually exclusive")
         if replay_ended_run_id is not None:
+            replayed_run_id = int(replay_ended_run_id)
             cur = conn.execute(
                 """
                 UPDATE tasks
@@ -95,8 +97,22 @@ complete_branch_after = '''    # Journal recovery may arrive after a legacy erro
                         ORDER BY id DESC
                         LIMIT 1
                    )
+                   AND EXISTS (
+                       SELECT 1 FROM task_runs
+                        WHERE id = ?
+                          AND task_id = ?
+                          AND ended_at IS NOT NULL
+                   )
                 """,
-                (result, now, task_id, int(replay_ended_run_id), task_id),
+                (
+                    result,
+                    now,
+                    task_id,
+                    replayed_run_id,
+                    task_id,
+                    replayed_run_id,
+                    task_id,
+                ),
             )
         elif expected_run_id is None:
             cur = conn.execute(
@@ -124,6 +140,80 @@ db = replace_once(
     "transactional exact ended-run completion",
 )
 
+run_completion_before = '''        run_id = _end_run(
+            conn, task_id,
+            outcome="completed", status="done",
+            summary=summary if summary is not None else result,
+            metadata=metadata,
+        )
+        # If complete_task was called on a never-claimed task (ready or
+        # blocked → done with no run in flight), synthesize a
+        # zero-duration run so the handoff fields are persisted in
+        # attempt history instead of silently lost.
+        if run_id is None and (summary or metadata or result):
+            run_id = _synthesize_ended_run(
+                conn, task_id,
+                outcome="completed",
+                summary=summary if summary is not None else result,
+                metadata=metadata,
+            )
+'''
+run_completion_after = '''        if replayed_run_id is None:
+            run_id = _end_run(
+                conn, task_id,
+                outcome="completed", status="done",
+                summary=summary if summary is not None else result,
+                metadata=metadata,
+            )
+            # If complete_task was called on a never-claimed task (ready or
+            # blocked → done with no run in flight), synthesize a
+            # zero-duration run so the handoff fields are persisted in
+            # attempt history instead of silently lost.
+            if run_id is None and (summary or metadata or result):
+                run_id = _synthesize_ended_run(
+                    conn, task_id,
+                    outcome="completed",
+                    summary=summary if summary is not None else result,
+                    metadata=metadata,
+                )
+        else:
+            run_id = replayed_run_id
+            replayed = conn.execute(
+                """
+                UPDATE task_runs
+                   SET status = 'done',
+                       outcome = 'completed',
+                       summary = ?,
+                       error = NULL,
+                       metadata = ?,
+                       claim_lock = NULL,
+                       claim_expires = NULL,
+                       worker_pid = NULL
+                 WHERE id = ?
+                   AND task_id = ?
+                   AND ended_at IS NOT NULL
+                """,
+                (
+                    summary if summary is not None else result,
+                    json.dumps(metadata, ensure_ascii=False) if metadata else None,
+                    run_id,
+                    task_id,
+                ),
+            )
+            if replayed.rowcount != 1:
+                raise RuntimeError("exact ended-run completion lost its run identity")
+        conn.execute(
+            "UPDATE tasks SET completed_run_id = ? WHERE id = ?",
+            (run_id, task_id),
+        )
+'''
+db = replace_once(
+    db,
+    run_completion_before,
+    run_completion_after,
+    "exact completion run provenance",
+)
+
 reclaim_before = '''def reclaim_task(
     conn: sqlite3.Connection,
     task_id: str,
@@ -284,6 +374,81 @@ db = replace_once(
     "transactional exact-run reclaim",
 )
 
+task_field_before = '''    current_run_id: Optional[int] = None
+    workflow_template_id: Optional[str] = None
+'''
+task_field_after = '''    current_run_id: Optional[int] = None
+    # Durable provenance for the run that authoritatively completed this task.
+    completed_run_id: Optional[int] = None
+    workflow_template_id: Optional[str] = None
+'''
+db = replace_once(
+    db,
+    task_field_before,
+    task_field_after,
+    "completed-run task field",
+)
+
+task_row_before = '''            current_run_id=(
+                row["current_run_id"] if "current_run_id" in keys else None
+            ),
+            workflow_template_id=(
+'''
+task_row_after = '''            current_run_id=(
+                row["current_run_id"] if "current_run_id" in keys else None
+            ),
+            completed_run_id=(
+                row["completed_run_id"] if "completed_run_id" in keys else None
+            ),
+            workflow_template_id=(
+'''
+db = replace_once(
+    db,
+    task_row_before,
+    task_row_after,
+    "completed-run row mapping",
+)
+
+task_schema_before = '''    -- run is in-flight). Denormalised for cheap reads.
+    current_run_id       INTEGER,
+    -- Forward-compat for v2 workflow routing. In v1 the kernel writes
+'''
+task_schema_after = '''    -- run is in-flight). Denormalised for cheap reads.
+    current_run_id       INTEGER,
+    -- Immutable provenance for the task result's authoritative run.
+    completed_run_id     INTEGER,
+    -- Forward-compat for v2 workflow routing. In v1 the kernel writes
+'''
+db = replace_once(
+    db,
+    task_schema_before,
+    task_schema_after,
+    "completed-run schema",
+)
+
+task_migration_before = '''    if "current_run_id" not in cols:
+        _add_column_if_missing(
+            conn, "tasks", "current_run_id", "current_run_id INTEGER"
+        )
+    if "workflow_template_id" not in cols:
+'''
+task_migration_after = '''    if "current_run_id" not in cols:
+        _add_column_if_missing(
+            conn, "tasks", "current_run_id", "current_run_id INTEGER"
+        )
+    if "completed_run_id" not in cols:
+        _add_column_if_missing(
+            conn, "tasks", "completed_run_id", "completed_run_id INTEGER"
+        )
+    if "workflow_template_id" not in cols:
+'''
+db = replace_once(
+    db,
+    task_migration_before,
+    task_migration_after,
+    "completed-run migration",
+)
+
 specify_signature_before = '''def specify_triage_task(
     conn: sqlite3.Connection,
     task_id: str,

services/hermes/scripts/cli_lane_runner.py

@@ -425,6 +425,11 @@ _TERMINAL_EVIDENCE_NAME = re.compile(
     r"^(?P<task>[a-zA-Z0-9_.-]+)\.run-(?P<run>[0-9]+)\."
     r"terminal\.(?P<state>prepared|conflict)-(?P<digest>[a-f0-9]{32})\.json$"
 )
+_RETIRE_NAME = re.compile(
+    r"^\.retire\.(?P<path_digest>[a-f0-9]{16})\."
+    r"(?P<source_digest>[a-f0-9]{16})\.(?P<sequence>[0-9]+)$"
+)
+_LEGACY_RETIRE_NAME = re.compile(r"^\.retire\.[a-f0-9]{32}\.[0-9]+$")
 _SAFE_BOARD = re.compile(r"^[a-zA-Z0-9_.-]+$")
 
 
@@ -1393,6 +1398,7 @@ def _retire_terminal_entry(
     *,
     board_descriptor: int,
     quarantine_descriptor: int,
+    authority_name: str | None = None,
 ) -> str:
     """Retire only the inode previously inspected; preserve any replacement."""
     try:
@@ -1405,11 +1411,18 @@ def _retire_terminal_entry(
         return "missing"
     if current.st_dev != source_stat.st_dev or current.st_ino != source_stat.st_ino:
         return "replacement"
-    identity = hashlib.sha256(
-        f"{path.name}\0{source_stat.st_dev:x}\0{source_stat.st_ino:x}".encode("utf-8")
-    ).hexdigest()[:32]
+    # A replacement can itself be staged repeatedly while recovery races a
+    # writer. Keep every generation bound to the original canonical pending
+    # name instead of hashing an intermediate .retire name and hiding it from
+    # the next recovery pass.
+    path_digest = hashlib.sha256(
+        (authority_name or path.name).encode("utf-8")
+    ).hexdigest()[:16]
+    source_digest = hashlib.sha256(
+        f"{source_stat.st_dev:x}\0{source_stat.st_ino:x}".encode("utf-8")
+    ).hexdigest()[:16]
     for sequence in range(32):
-        staging = f".retire.{identity}.{sequence}"
+        staging = f".retire.{path_digest}.{source_digest}.{sequence}"
         try:
             _rename_noreplace(
                 path.name,
@@ -1858,13 +1871,19 @@ def _recover_exact_run(
     return recovered
 
 
-def _retire_snapshot(path: Path, snapshot: TerminalSnapshot) -> str:
+def _retire_snapshot(
+    path: Path,
+    snapshot: TerminalSnapshot,
+    *,
+    authority_name: str | None = None,
+) -> str:
     """Retire only the directory entry still naming an opened snapshot inode."""
     return _retire_terminal_entry(
         path,
         snapshot.file_stat,
         board_descriptor=snapshot.directory_descriptor,
         quarantine_descriptor=snapshot.directory_descriptor,
+        authority_name=authority_name,
     )
 
 
@@ -1946,7 +1965,11 @@ def _finalize_document_db(
         if status == "done":
             return (
                 "committed"
-                if str(_task_value(task, "result", "") or "") == document["result"]
+                if (
+                    _task_value(task, "completed_run_id", None) == identity.run_id
+                    and str(_task_value(task, "result", "") or "")
+                    == document["result"]
+                )
                 else "stale"
             )
         current_run_id = _task_value(task, "current_run_id", None)
@@ -1976,7 +1999,11 @@ def _finalize_document_db(
         if latest is not None and str(_task_value(latest, "status", "")) == "done":
             return (
                 "committed"
-                if str(_task_value(latest, "result", "") or "") == document["result"]
+                if (
+                    _task_value(latest, "completed_run_id", None) == identity.run_id
+                    and str(_task_value(latest, "result", "") or "")
+                    == document["result"]
+                )
                 else "stale"
             )
         return "pending"
@@ -2102,6 +2129,92 @@ def _terminal_entry_absent(path: Path) -> bool:
             os.close(directory)
 
 
+def _staged_terminal_authority(
+    path: Path,
+    snapshot: TerminalRecoverySnapshot,
+) -> tuple[TerminalIdentity, Path] | None:
+    """Bind a staged inode to the canonical pending name it was moved from."""
+    try:
+        relative = path.relative_to(STATE_ROOT)
+    except ValueError:
+        return None
+    if len(relative.parts) != 2:
+        return None
+    board, filename = relative.parts
+    match = _RETIRE_NAME.fullmatch(filename)
+    legacy = _LEGACY_RETIRE_NAME.fullmatch(filename)
+    document = snapshot.document
+    if (
+        (match is None and legacy is None)
+        or document is None
+        or not _SAFE_BOARD.fullmatch(board)
+        or board in {".", ".."}
+        or document.get("board") != board
+        or not isinstance(document.get("task_id"), str)
+        or not isinstance(document.get("expected_run_id"), int)
+    ):
+        return None
+    identity = TerminalIdentity(
+        board,
+        document["task_id"],
+        document["expected_run_id"],
+        "pending",
+    )
+    canonical = _terminal_path(
+        state_path(identity.board, identity.task_id),
+        identity.run_id,
+        "pending",
+    )
+    if (
+        canonical.parent != path.parent
+        or (
+            match is not None
+            and hashlib.sha256(canonical.name.encode("utf-8")).hexdigest()[:16]
+            != match.group("path_digest")
+        )
+        or not _terminal_record_valid(document, identity)
+    ):
+        return None
+    return identity, canonical
+
+
+def _recover_retirement_staging() -> int:
+    """Promote valid hidden journals to durable prepared evidence."""
+    recovered = 0
+    for path in sorted(STATE_ROOT.glob("*/.retire.*")):
+        snapshot = _open_terminal_recovery_snapshot(path)
+        if snapshot is None:
+            continue
+        authority = _staged_terminal_authority(path, snapshot)
+        if authority is None:
+            try:
+                _quarantine_terminal(
+                    path,
+                    None,
+                    "untrusted-retirement-staging",
+                    snapshot=snapshot,
+                )
+            finally:
+                snapshot.close()
+            continue
+        identity, canonical = authority
+        try:
+            _persist_prepared_evidence(identity, snapshot.document or {})
+            retirement = _retire_snapshot(
+                path,
+                snapshot,
+                authority_name=canonical.name,
+            )
+        except Exception as error:
+            _record_board_access_error(identity.board, error)
+            continue
+        finally:
+            snapshot.close()
+        if retirement in {"retired", "missing"}:
+            recovered += 1
+    return recovered
+
+
 def _recover_prepared_finalizations(kanban_db: Any) -> int:
     """Recover an accepted result staged before an interrupted DB boundary."""
     recovered = 0
@@ -2176,6 +2289,7 @@ def recover_pending_finalizations() -> int:
     """Replay accepted exact-run results before scheduling more provider work."""
     from hermes_cli import kanban_db
 
+    _recover_retirement_staging()
     recovered = _recover_prepared_finalizations(kanban_db)
     for path in sorted(STATE_ROOT.glob("*/*.terminal.pending.json")):
         identity = _terminal_identity(path)
@@ -2273,6 +2387,21 @@ def _has_pending_finalization(board: str, task_id: str, run_id: Any) -> bool:
         record = _load_small_json(prepared)
         if _terminal_evidence_valid(record, identity, "prepared"):
             return True
+    for staged in base.parent.glob(".retire.*"):
+        snapshot = _open_terminal_recovery_snapshot(staged)
+        if snapshot is None:
+            continue
+        try:
+            authority = _staged_terminal_authority(staged, snapshot)
+            if authority is not None and authority[0] == TerminalIdentity(
+                board,
+                task_id,
+                run_id,
+                "pending",
+            ):
+                return True
+        finally:
+            snapshot.close()
     return False
 
 
@@ -2284,6 +2413,7 @@ def _artifact_gc_candidates(board_dir: Path) -> list[Path]:
         "*.candidate-*.json",
         "*.terminal.committed.json",
         "*.terminal.conflict-*.json",
+        ".retire.*",
     )
     for pattern in patterns:
         candidates.extend(board_dir.glob(pattern))
@@ -2330,6 +2460,16 @@ def gc_lane_artifacts(
         board_removed = 0
         entries: list[tuple[Path, os.stat_result]] = []
         for path in _artifact_gc_candidates(board_dir):
+            if path.name.startswith(".retire."):
+                snapshot = _open_terminal_recovery_snapshot(path)
+                if snapshot is not None:
+                    try:
+                        if _staged_terminal_authority(path, snapshot) is not None:
+                            # Accepted authority is first promoted/replayed by
+                            # recovery; retention never deletes it directly.
+                            continue
+                    finally:
+                        snapshot.close()
             try:
                 file_stat = path.stat(follow_symlinks=False)
             except OSError:
@@ -2836,12 +2976,15 @@ def recover_orphans() -> None:
                     if _external(task) and str(_task_value(task, "status", "")) == "running":
                         task_id = str(_task_value(task, "id"))
                         run_id = _task_value(task, "current_run_id", None)
+                        if not isinstance(run_id, int):
+                            continue
                         if _has_pending_finalization(board, task_id, run_id):
                             continue
                         kanban_db.reclaim_task(
                             conn,
                             task_id,
                             reason="direct CLI lane restarted; provider session will resume",
+                            expected_run_id=run_id,
                         )
                 BOARD_CORRUPTION_ERRORS.pop(board, None)
             except Exception as error:

services/hermes/scripts/recover_cassandra_workers.py

@@ -24,10 +24,12 @@ def recover_running_tasks(kanban_db: Any) -> list[str]:
                 if str(getattr(task, "status", "")) != "running":
                     continue
                 task_id = str(getattr(task, "id", ""))
-                if not task_id or not kanban_db.reclaim_task(
+                run_id = getattr(task, "current_run_id", None)
+                if not task_id or not isinstance(run_id, int) or not kanban_db.reclaim_task(
                     connection,
                     task_id,
                     reason=REASON,
+                    expected_run_id=run_id,
                 ):
                     continue
                 kanban_db.add_comment(

testing/tests/test_hermes_cli_lanes.py

@@ -11,6 +11,7 @@ import signal
 import stat
 import sys
 import threading
+import time
 from contextlib import nullcontext
 from pathlib import Path
 from types import SimpleNamespace
@@ -686,6 +687,7 @@ def test_accepted_result_survives_failed_finalization_and_replays_exact_run(
             return False
         task.status = "done"
         task.result = kwargs["result"]
+        task.completed_run_id = kwargs["expected_run_id"]
         task.current_run_id = None
         return True
 
@@ -805,6 +807,56 @@ def test_restart_does_not_reclaim_an_exact_run_awaiting_finalization(
     assert reclaimed == []
 
 
+def test_orphan_recovery_passes_the_scanned_run_as_atomic_reclaim_guard(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    tasks = [
+        SimpleNamespace(
+            id="t_guarded",
+            status="running",
+            assignee="cli-auto",
+            current_run_id=91,
+        ),
+        SimpleNamespace(
+            id="t_no_run",
+            status="running",
+            assignee="cli-auto",
+            current_run_id=None,
+        ),
+    ]
+    reclaimed = []
+
+    class Connection:
+        def close(self):
+            return None
+
+    fake_db = SimpleNamespace(
+        list_boards=lambda include_archived=False: [{"slug": "cassandra"}],
+        scoped_current_board=lambda _board: nullcontext(),
+        connect=lambda board: Connection(),
+        list_tasks=lambda _conn: tasks,
+        reclaim_task=lambda _conn, task_id, **kwargs: (
+            reclaimed.append((task_id, kwargs)) or True
+        ),
+    )
+    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
+    monkeypatch.setattr(lanes, "recover_pending_finalizations", lambda: 0)
+
+    lanes.recover_orphans()
+
+    assert reclaimed == [
+        (
+            "t_guarded",
+            {
+                "reason": "direct CLI lane restarted; provider session will resume",
+                "expected_run_id": 91,
+            },
+        )
+    ]
+
+
 def test_prepared_evidence_without_pending_still_pins_the_exact_run(
     tmp_path: Path,
     monkeypatch,
@@ -1683,6 +1735,295 @@ def test_recovery_finalizes_the_valid_snapshot_it_classified_before_replacement(
     assert conflict["summary"] == "second valid replacement"
 
 
+def test_replacement_staged_journal_promotes_replays_and_never_reexecutes(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    path = lanes._terminal_path(lanes.state_path("cassandra", "t_staged"), 55)
+    path.parent.mkdir(parents=True)
+    path.write_bytes(b"malformed source")
+    source_stat = path.stat(follow_symlinks=False)
+    staged_winner = path.with_name("staged-winner.tmp")
+    canonical_replacement = path.with_name("canonical-replacement.tmp")
+    first = _pending_terminal_record(
+        "cassandra", "t_staged", 55, "staged accepted result"
+    )
+    second = _pending_terminal_record(
+        "cassandra", "t_staged", 55, "later canonical replacement"
+    )
+    lanes.atomic_json(staged_winner, first)
+    lanes.atomic_json(canonical_replacement, second)
+    directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
+    descriptor = os.open(path.parent, directory_flags)
+    real_rename = lanes._rename_noreplace
+    rename_steps = []
+
+    def force_replacement_staging(source, destination, **kwargs):
+        rename_steps.append((source, destination))
+        if source == path.name:
+            os.replace(staged_winner, path)
+        elif source.startswith(".retire."):
+            os.replace(canonical_replacement, path)
+        return real_rename(source, destination, **kwargs)
+
+    try:
+        with monkeypatch.context() as race:
+            race.setattr(lanes, "_rename_noreplace", force_replacement_staging)
+            outcome = lanes._retire_terminal_entry(
+                path,
+                source_stat,
+                board_descriptor=descriptor,
+                quarantine_descriptor=descriptor,
+            )
+    finally:
+        os.close(descriptor)
+
+    assert outcome == "replacement-staged"
+    assert len(rename_steps) == 2
+    staged = list(path.parent.glob(".retire.*"))
+    assert len(staged) == 1
+    staged_snapshot = lanes._open_terminal_recovery_snapshot(staged[0])
+    assert staged_snapshot is not None
+    try:
+        authority = lanes._staged_terminal_authority(staged[0], staged_snapshot)
+        assert authority is not None
+        assert authority[0].run_id == 55
+        assert authority[1] == path
+    finally:
+        staged_snapshot.close()
+    assert lanes._has_pending_finalization("cassandra", "t_staged", 55) is True
+    assert lanes.gc_lane_artifacts(
+        now=staged[0].stat().st_mtime + 3600,
+        max_age_seconds=0,
+        max_count=0,
+        max_bytes=0,
+    ) == 0
+    assert staged[0].exists()
+
+    task = SimpleNamespace(
+        id="t_staged",
+        status="running",
+        current_run_id=55,
+        completed_run_id=None,
+        result=None,
+        assignee="cli-auto",
+    )
+    completions = []
+
+    class Connection:
+        def close(self):
+            return None
+
+    def complete_task(_conn, _task_id, **kwargs):
+        task.status = "done"
+        task.result = kwargs["result"]
+        task.completed_run_id = kwargs["expected_run_id"]
+        task.current_run_id = None
+        completions.append(kwargs["result"])
+        return True
+
+    fake_db = SimpleNamespace(
+        scoped_current_board=lambda _board: nullcontext(),
+        connect=lambda board: Connection(),
+        get_task=lambda _conn, _task_id: task,
+        complete_task=complete_task,
+    )
+    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
+
+    assert lanes.recover_pending_finalizations() == 1
+    assert completions == [first["result"]]
+    assert not list(path.parent.glob(".retire.*"))
+    assert not path.exists()
+    conflicts = list(path.parent.glob("*.terminal.conflict-*.json"))
+    assert len(conflicts) == 1
+    assert json.loads(conflicts[0].read_text())["result"] == second["result"]
+    assert lanes.recover_pending_finalizations() == 0
+    assert completions == [first["result"]]
+
+
+def test_repeated_retirement_staging_preserves_every_valid_result_once(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    canonical = lanes._terminal_path(
+        lanes.state_path("cassandra", "t_staged_repeat"), 57
+    )
+    canonical.parent.mkdir(parents=True)
+    path_digest = hashlib.sha256(canonical.name.encode("utf-8")).hexdigest()[:16]
+    staged = canonical.parent / f".retire.{path_digest}.{'a' * 16}.0"
+    records = [
+        _pending_terminal_record(
+            "cassandra", "t_staged_repeat", 57, f"accepted result {number}"
+        )
+        for number in range(3)
+    ]
+    lanes.atomic_json(staged, records[0])
+    second = canonical.with_name("second-staged-replacement.tmp")
+    third = canonical.with_name("third-staged-replacement.tmp")
+    lanes.atomic_json(second, records[1])
+    lanes.atomic_json(third, records[2])
+    real_rename = lanes._rename_noreplace
+    retirement_steps = []
+
+    def replace_each_retirement_generation(source, destination, **kwargs):
+        if source == staged.name and destination.startswith(".retire."):
+            retirement_steps.append("move")
+            os.replace(second, staged)
+        elif source.startswith(".retire.") and destination == staged.name:
+            retirement_steps.append("restore")
+            os.replace(third, staged)
+        return real_rename(source, destination, **kwargs)
+
+    with monkeypatch.context() as race:
+        race.setattr(lanes, "_rename_noreplace", replace_each_retirement_generation)
+        assert lanes._recover_retirement_staging() == 0
+
+    assert retirement_steps == ["move", "restore"]
+    staged_generations = list(canonical.parent.glob(".retire.*"))
+    assert len(staged_generations) == 2
+    for generation in staged_generations:
+        snapshot = lanes._open_terminal_recovery_snapshot(generation)
+        assert snapshot is not None
+        try:
+            authority = lanes._staged_terminal_authority(generation, snapshot)
+            assert authority is not None
+            assert authority[1] == canonical
+        finally:
+            snapshot.close()
+
+    task = SimpleNamespace(
+        id="t_staged_repeat",
+        status="running",
+        current_run_id=57,
+        completed_run_id=None,
+        result=None,
+        assignee="cli-auto",
+    )
+    completions = []
+    reclaims = []
+    _install_terminal_recovery_db(monkeypatch, task, completions, reclaims)
+
+    assert lanes.recover_pending_finalizations() == 1
+    assert len(completions) == 1
+    assert reclaims == []
+    assert not list(canonical.parent.glob(".retire.*"))
+    conflicts = list(canonical.parent.glob("*.terminal.conflict-*.json"))
+    assert len(conflicts) == 2
+    preserved_results = {task.result}
+    preserved_results.update(json.loads(path.read_text())["result"] for path in conflicts)
+    assert preserved_results == {record["result"] for record in records}
+    assert lanes.recover_pending_finalizations() == 0
+    assert len(completions) == 1
+
+
+def test_untrusted_retirement_staging_is_bounded_by_artifact_gc(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    board = lanes.STATE_ROOT / "cassandra"
+    board.mkdir(parents=True)
+    for sequence in range(10):
+        (board / f".retire.{'a' * 16}.{'b' * 16}.{sequence}").write_bytes(
+            b"not terminal authority"
+        )
+    (board / f".retire.{'c' * 32}.0").write_bytes(b"legacy hidden artifact")
+
+    removed = lanes.gc_lane_artifacts(
+        now=time.time(),
+        max_age_seconds=3600,
+        max_count=2,
+        max_bytes=1024 * 1024,
+    )
+
+    assert removed == 9
+    assert len(list(board.glob(".retire.*"))) == 2
+
+
+def test_legacy_valid_retirement_staging_is_promoted_without_reexecution(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    board = lanes.STATE_ROOT / "cassandra"
+    board.mkdir(parents=True)
+    staged = board / f".retire.{'d' * 32}.0"
+    record = _pending_terminal_record(
+        "cassandra", "t_legacy_staged", 56, "legacy staged result"
+    )
+    lanes.atomic_json(staged, record)
+    task = SimpleNamespace(
+        id="t_legacy_staged",
+        status="running",
+        current_run_id=56,
+        completed_run_id=None,
+        result=None,
+        assignee="cli-auto",
+    )
+    completions = []
+    reclaims = []
+    _install_terminal_recovery_db(monkeypatch, task, completions, reclaims)
+
+    assert lanes._has_pending_finalization(
+        "cassandra", "t_legacy_staged", 56
+    ) is True
+    assert lanes.recover_pending_finalizations() == 1
+    assert completions == [record["result"]]
+    assert reclaims == []
+    assert not staged.exists()
+    assert lanes.recover_pending_finalizations() == 0
+    assert completions == [record["result"]]
+
+
+def test_done_result_bytes_do_not_authorize_a_different_run_journal(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    path, stale = lanes._write_terminal_record(
+        lanes.state_path("cassandra", "t_provenance"),
+        board="cassandra",
+        task_id="t_provenance",
+        run_id=1,
+        structured=_completed_result("same bytes"),
+        summary="same bytes",
+        metadata={},
+    )
+    task = SimpleNamespace(
+        id="t_provenance",
+        status="done",
+        current_run_id=None,
+        completed_run_id=2,
+        result=stale["result"],
+        assignee="cli-auto",
+    )
+
+    class Connection:
+        def close(self):
+            return None
+
+    fake_db = SimpleNamespace(
+        scoped_current_board=lambda _board: nullcontext(),
+        connect=lambda board: Connection(),
+        get_task=lambda _conn, _task_id: task,
+        complete_task=lambda *_args, **_kwargs: pytest.fail(
+            "a done task with different run provenance must not be completed"
+        ),
+    )
+    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
+
+    assert lanes.recover_pending_finalizations() == 0
+    assert not path.exists()
+    assert not list(path.parent.glob("*.terminal.committed.json"))
+    conflicts = list(path.parent.glob("*.terminal.conflict-*.json"))
+    assert len(conflicts) == 1
+    conflict = json.loads(conflicts[0].read_text())
+    assert conflict["expected_run_id"] == 1
+    assert conflict["result"] == task.result
+
+
 @pytest.mark.parametrize(
     "swap_point",
     [
@@ -1821,6 +2162,7 @@ def test_terminal_commit_directory_fsync_failure_recovers_from_prepared_evidence
         status="running",
         result=None,
         current_run_id=28,
+        completed_run_id=None,
         assignee="cli-auto",
     )
     replacement_structured = _completed_result("replacement after DB commit")
@@ -1844,6 +2186,7 @@ def test_terminal_commit_directory_fsync_failure_recovers_from_prepared_evidence
     def complete_task(_conn, _task_id, **kwargs):
         task.status = "done"
         task.result = kwargs["result"]
+        task.completed_run_id = kwargs["expected_run_id"]
         task.current_run_id = None
         os.replace(replacement, path)
         return True
@@ -1912,6 +2255,7 @@ def test_unknown_db_completion_outcome_replays_from_prepared_evidence(
         status="running",
         result=None,
         current_run_id=35,
+        completed_run_id=None,
         assignee="cli-auto",
     )
     fail_once = {"value": True}
@@ -1923,6 +2267,7 @@ def test_unknown_db_completion_outcome_replays_from_prepared_evidence(
     def complete_task(_conn, _task_id, **kwargs):
         task.status = "done"
         task.result = kwargs["result"]
+        task.completed_run_id = kwargs["expected_run_id"]
         task.current_run_id = None
         if fail_once["value"]:
             fail_once["value"] = False
@@ -2093,6 +2438,7 @@ def test_concurrent_duplicate_terminal_finalizers_are_idempotent(
         "status": "running",
         "result": None,
         "current_run_id": 29,
+        "completed_run_id": None,
     }
     state_lock = threading.Lock()
     readers = threading.Barrier(2)
@@ -2118,10 +2464,11 @@ def test_concurrent_duplicate_terminal_finalizers_are_idempotent(
             if task_state["status"] != "running":
                 return False
             task_state.update(
-                status="done",
-                result=kwargs["result"],
-                current_run_id=None,
-            )
+                    status="done",
+                    result=kwargs["result"],
+                    current_run_id=None,
+                    completed_run_id=kwargs["expected_run_id"],
+                )
             completions.append(kwargs["result"])
             return True
 

testing/tests/test_hermes_worker_recovery.py

@@ -42,7 +42,7 @@ class FakeKanban:
         self.tasks = tasks
         self.has_board = board_exists
         self.connection = FakeConnection()
-        self.reclaimed: list[tuple[str, str]] = []
+        self.reclaimed: list[tuple[str, str, int]] = []
         self.comments: list[tuple[str, str, str]] = []
 
     def board_exists(self, board: str) -> bool:
@@ -61,9 +61,16 @@ class FakeKanban:
         assert connection is self.connection
         return self.tasks
 
-    def reclaim_task(self, connection, task_id: str, *, reason: str) -> bool:
+    def reclaim_task(
+        self,
+        connection,
+        task_id: str,
+        *,
+        reason: str,
+        expected_run_id: int,
+    ) -> bool:
         assert connection is self.connection
-        self.reclaimed.append((task_id, reason))
+        self.reclaimed.append((task_id, reason, expected_run_id))
         return task_id != "t_race"
 
     def add_comment(self, connection, task_id: str, author: str, body: str) -> None:
@@ -75,16 +82,17 @@ def test_recover_running_tasks_requeues_only_claimed_workers() -> None:
     module = _load_module()
     kanban = FakeKanban(
         [
-            SimpleNamespace(id="t_running", status="running"),
+            SimpleNamespace(id="t_running", status="running", current_run_id=10),
             SimpleNamespace(id="t_done", status="done"),
-            SimpleNamespace(id="t_race", status="running"),
+            SimpleNamespace(id="t_race", status="running", current_run_id=11),
+            SimpleNamespace(id="t_missing_run", status="running", current_run_id=None),
         ]
     )
 
     assert module.recover_running_tasks(kanban) == ["t_running"]
     assert kanban.reclaimed == [
-        ("t_running", module.REASON),
-        ("t_race", module.REASON),
+        ("t_running", module.REASON, 10),
+        ("t_race", module.REASON, 11),
     ]
     assert kanban.comments == [
         (