hermes: close terminal replay edge cases

dockerfiles/hermes-execution-safety-regression.py

@@ -146,6 +146,68 @@ class AutomaticDecompositionSafetyTests(unittest.TestCase):
         self.assertEqual(completions.calls, 1)
         self.assertEqual(self._status(task_id), "ready")
 
+    def test_latest_ended_run_can_replay_a_durable_completion(self) -> None:
+        task_id = kanban_db.create_task(self.connection, title="journaled result")
+        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.block_task(
+                self.connection,
+                task_id,
+                reason="legacy post-journal failure",
+                kind="capability",
+                expected_run_id=run_id,
+            )
+        )
+
+        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")
+
+    def test_older_ended_run_cannot_complete_over_a_replacement(self) -> None:
+        task_id = kanban_db.create_task(self.connection, title="replacement guard")
+        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="first run",
+                kind="capability",
+                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_run = kanban_db.get_task(self.connection, task_id).current_run_id
+        self.assertNotEqual(old_run, replacement_run)
+        self.assertTrue(
+            kanban_db.block_task(
+                self.connection,
+                task_id,
+                reason="replacement run",
+                kind="capability",
+                expected_run_id=replacement_run,
+            )
+        )
+
+        self.assertFalse(
+            kanban_db.complete_task(
+                self.connection,
+                task_id,
+                result="stale result",
+                replay_ended_run_id=old_run,
+            )
+        )
+        self.assertIn(self._status(task_id), {"blocked", "triage"})
+
     def test_fresh_triage_task_still_auto_promotes(self) -> None:
         task_id = kanban_db.create_task(
             self.connection,

dockerfiles/patch-hermes-execution-safety.py

@@ -1,4 +1,4 @@
-"""Patch upstream Hermes auto-decomposition to respect execution history."""
+"""Patch upstream Hermes exact-run completion and decomposition safety."""
 
 import os
 from pathlib import Path
@@ -18,6 +18,112 @@ source_root = Path(os.environ.get("HERMES_SOURCE_ROOT", "/opt/hermes"))
 db_path = source_root / "hermes_cli/kanban_db.py"
 db = db_path.read_text(encoding="utf-8")
 
+complete_signature_before = '''def complete_task(
+    conn: sqlite3.Connection,
+    task_id: str,
+    *,
+    result: Optional[str] = None,
+    summary: Optional[str] = None,
+    metadata: Optional[dict] = None,
+    created_cards: Optional[Iterable[str]] = None,
+    expected_run_id: Optional[int] = None,
+) -> bool:
+'''
+complete_signature_after = '''def complete_task(
+    conn: sqlite3.Connection,
+    task_id: str,
+    *,
+    result: Optional[str] = None,
+    summary: Optional[str] = None,
+    metadata: Optional[dict] = None,
+    created_cards: Optional[Iterable[str]] = None,
+    expected_run_id: Optional[int] = None,
+    replay_ended_run_id: Optional[int] = None,
+) -> bool:
+'''
+db = replace_once(
+    db,
+    complete_signature_before,
+    complete_signature_after,
+    "exact ended-run completion signature",
+)
+
+complete_branch_before = '''    with write_txn(conn):
+        if expected_run_id is None:
+            cur = conn.execute(
+                """
+                UPDATE tasks
+                   SET status       = 'done',
+                       result       = ?,
+                       completed_at = ?,
+                       claim_lock   = NULL,
+                       claim_expires= NULL,
+                       worker_pid   = NULL,
+                       block_kind   = NULL,
+                       block_recurrences = 0
+                 WHERE id = ?
+                   AND status IN ('running', 'ready', 'blocked', 'scheduled')
+                """,
+                (result, now, task_id),
+            )
+        else:
+'''
+complete_branch_after = '''    # Journal recovery may arrive after a legacy error path ended the
+    # 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):
+        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:
+            cur = conn.execute(
+                """
+                UPDATE tasks
+                   SET status       = 'done',
+                       result       = ?,
+                       completed_at = ?,
+                       claim_lock   = NULL,
+                       claim_expires= NULL,
+                       worker_pid   = NULL,
+                       block_kind   = NULL,
+                       block_recurrences = 0
+                 WHERE id = ?
+                   AND status IN ('ready', 'blocked')
+                   AND current_run_id IS NULL
+                   AND ? = (
+                       SELECT id FROM task_runs
+                        WHERE task_id = ?
+                        ORDER BY id DESC
+                        LIMIT 1
+                   )
+                """,
+                (result, now, task_id, int(replay_ended_run_id), task_id),
+            )
+        elif expected_run_id is None:
+            cur = conn.execute(
+                """
+                UPDATE tasks
+                   SET status       = 'done',
+                       result       = ?,
+                       completed_at = ?,
+                       claim_lock   = NULL,
+                       claim_expires= NULL,
+                       worker_pid   = NULL,
+                       block_kind   = NULL,
+                       block_recurrences = 0
+                 WHERE id = ?
+                   AND status IN ('running', 'ready', 'blocked', 'scheduled')
+                """,
+                (result, now, task_id),
+            )
+        else:
+'''
+db = replace_once(
+    db,
+    complete_branch_before,
+    complete_branch_after,
+    "transactional exact ended-run completion",
+)
+
 specify_signature_before = '''def specify_triage_task(
     conn: sqlite3.Connection,
     task_id: str,

services/hermes/scripts/cli_lane_runner.py

@@ -522,10 +522,27 @@ def _terminal_record_valid(
         structured = json.loads(record["result"])
     except (TypeError, json.JSONDecodeError):
         return False
+    if not isinstance(structured, dict):
+        return False
+    required = set(RESULT_SCHEMA["required"])
+    if set(structured) != required:
+        return False
+    if (
+        type(structured["status"]) is not str
+        or structured["status"] not in cli_lane_goal.RESULT_STATUSES
+        or type(structured["summary"]) is not str
+        or not structured["summary"].strip()
+    ):
+        return False
+    for key in ("changed_files", "tests_run", "artifacts", "findings", "blockers"):
+        value = structured[key]
+        if type(value) is not list or any(type(item) is not str for item in value):
+            return False
     valid = (
-        isinstance(structured, dict)
+        structured["status"] == "completed"
-        and structured.get("status") == "completed"
+        and structured["blockers"] == []
-        and not structured.get("blockers")
+        and cli_lane_goal.unfinished_result_reason(structured) is None
+        and record["summary"] == structured["summary"]
         and record.get("kanban_state") in {"pending", "committed"}
     )
     if not valid or identity is None:
@@ -1077,7 +1094,7 @@ def _board_call(
 
 
 def _quarantine_terminal(path: Path, identity: TerminalIdentity | None, reason: str) -> Path:
-    """Move an unusable journal aside with a deterministic, private name."""
+    """Copy an unusable journal to a private file, then retire its exact name."""
     board_dir = path.parent
     try:
         board_stat = board_dir.stat(follow_symlinks=False)
@@ -1090,44 +1107,135 @@ def _quarantine_terminal(path: Path, identity: TerminalIdentity | None, reason:
             flush=True,
         )
         return path
-    quarantine = board_dir / "quarantine"
+    board_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
-    quarantine.mkdir(parents=True, exist_ok=True, mode=0o700)
+    board_flags |= getattr(os, "O_NOFOLLOW", 0)
-    os.chmod(quarantine, 0o700, follow_symlinks=False)
+    file_flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+    create_flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
+    create_flags |= getattr(os, "O_NOFOLLOW", 0)
+    board_descriptor = None
+    quarantine_descriptor = None
+    source_stat = None
+    destination = path
+    retired = False
     try:
-        source_stat = path.stat(follow_symlinks=False)
+        board_descriptor = os.open(board_dir, board_flags)
-        regular = stat.S_ISREG(source_stat.st_mode)
+        source_stat = os.stat(
-    except OSError:
+            path.name,
-        regular = False
+            dir_fd=board_descriptor,
-    try:
+            follow_symlinks=False,
-        if not regular:
+        )
-            raise OSError("journal is not a regular file")
+        safe_source = stat.S_ISREG(source_stat.st_mode) and source_stat.st_nlink == 1
-        flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+        payload: bytes
-        descriptor = os.open(path, flags)
+        if safe_source:
-        with os.fdopen(descriptor, "rb") as stream:
+            source_descriptor = os.open(
-            fingerprint = hashlib.sha256(stream.read()).hexdigest()[:16]
+                path.name,
-    except OSError:
+                file_flags,
-        fingerprint = hashlib.sha256(path.name.encode("utf-8")).hexdigest()[:16]
+                dir_fd=board_descriptor,
-    safe_reason = re.sub(r"[^a-zA-Z0-9_.-]+", "-", reason).strip("-") or "invalid"
-    destination = quarantine / f"{path.name}.{safe_reason}.{fingerprint}.quarantine"
-    source_exists = path.exists() or path.is_symlink()
-    if source_exists and regular and not destination.exists():
-        os.chmod(path, 0o600, follow_symlinks=False)
-        os.replace(path, destination)
-        os.chmod(destination, 0o600, follow_symlinks=False)
-        _fsync_directory(board_dir)
-        _fsync_directory(quarantine)
-    elif source_exists:
-        path.unlink()
-        _fsync_directory(board_dir)
-        if not destination.exists():
-            atomic_json(
-                destination,
-                {
-                    "original_name": path.name,
-                    "reason": reason,
-                    "recorded_at": utc_now(),
-                },
             )
+            with os.fdopen(source_descriptor, "rb") as source:
+                opened_stat = os.fstat(source.fileno())
+                if (
+                    opened_stat.st_dev != source_stat.st_dev
+                    or opened_stat.st_ino != source_stat.st_ino
+                    or opened_stat.st_nlink != 1
+                ):
+                    raise OSError("terminal journal identity changed while opening")
+                payload = source.read()
+        else:
+            payload = (
+                json.dumps(
+                    {
+                        "original_name": path.name,
+                        "reason": reason,
+                        "source_kind": "hardlink"
+                        if stat.S_ISREG(source_stat.st_mode) and source_stat.st_nlink != 1
+                        else "non-regular",
+                    },
+                    sort_keys=True,
+                )
+                + "\n"
+            ).encode("utf-8")
+        fingerprint = hashlib.sha256(payload).hexdigest()[:16]
+        safe_reason = (
+            re.sub(r"[^a-zA-Z0-9_.-]+", "-", reason).strip("-") or "invalid"
+        )
+        try:
+            os.mkdir("quarantine", 0o700, dir_fd=board_descriptor)
+        except FileExistsError:
+            pass
+        quarantine_descriptor = os.open(
+            "quarantine",
+            board_flags,
+            dir_fd=board_descriptor,
+        )
+        for sequence in range(32):
+            name = f"{path.name}.{safe_reason}.{fingerprint}.{sequence}.quarantine"
+            try:
+                destination_descriptor = os.open(
+                    name,
+                    create_flags,
+                    0o600,
+                    dir_fd=quarantine_descriptor,
+                )
+            except FileExistsError:
+                continue
+            try:
+                with os.fdopen(destination_descriptor, "wb") as target:
+                    target.write(payload)
+                    target.flush()
+                    os.fsync(target.fileno())
+                    target_stat = os.fstat(target.fileno())
+                    if (
+                        stat.S_IMODE(target_stat.st_mode) != 0o600
+                        or target_stat.st_nlink != 1
+                    ):
+                        raise OSError("quarantine destination is not private and singly linked")
+            except Exception:
+                os.unlink(name, dir_fd=quarantine_descriptor)
+                raise
+            destination = board_dir / "quarantine" / name
+            break
+        else:
+            raise OSError("could not reserve a unique quarantine destination")
+        os.fsync(quarantine_descriptor)
+        current = os.stat(
+            path.name,
+            dir_fd=board_descriptor,
+            follow_symlinks=False,
+        )
+        if (
+            current.st_dev != source_stat.st_dev
+            or current.st_ino != source_stat.st_ino
+        ):
+            print(
+                f"terminal journal changed before retirement; removing new entry: {path}",
+                file=sys.stderr,
+                flush=True,
+            )
+        os.unlink(path.name, dir_fd=board_descriptor)
+        retired = True
+        os.fsync(board_descriptor)
+    except OSError as error:
+        # A bad quarantine target must not make an attacker-controlled pending
+        # path eligible for replay forever. Retire only the directory entry;
+        # never chmod or write through the source inode.
+        if board_descriptor is not None and not retired:
+            try:
+                os.unlink(path.name, dir_fd=board_descriptor)
+                retired = True
+                os.fsync(board_descriptor)
+            except OSError:
+                pass
+        print(
+            f"terminal journal quarantine degraded safely: {type(error).__name__}: {error}",
+            file=sys.stderr,
+            flush=True,
+        )
+    finally:
+        if quarantine_descriptor is not None:
+            os.close(quarantine_descriptor)
+        if board_descriptor is not None:
+            os.close(board_descriptor)
     print(
         "quarantined terminal journal "
         f"{path.name}: {reason}; identity={identity or 'unparseable'}",
@@ -1137,12 +1245,12 @@ def _quarantine_terminal(path: Path, identity: TerminalIdentity | None, reason:
     return destination
 
 
-def _recover_quarantined_run(
+def _recover_exact_run(
     kanban_db: Any,
     identity: TerminalIdentity | None,
     reason: str,
 ) -> bool:
-    """Make an exact external run retryable after its journal is quarantined."""
+    """Make an exact external run retryable after journal recovery fails."""
     if identity is None:
         return False
 
@@ -1160,7 +1268,7 @@ def _recover_quarantined_run(
             kanban_db.reclaim_task(
                 conn,
                 identity.task_id,
-                reason=f"terminal journal quarantined ({reason}); exact run may retry",
+                reason=f"terminal journal recovery failed ({reason}); exact run may retry",
             )
         )
 
@@ -1172,7 +1280,7 @@ def _recover_quarantined_run(
     if recovered:
         print(
             f"reclaimed {identity.board}/{identity.task_id} run {identity.run_id} "
-            f"after terminal journal quarantine: {reason}",
+            f"after terminal journal recovery: {reason}",
             file=sys.stderr,
             flush=True,
         )
@@ -1221,10 +1329,16 @@ def _finalize_terminal_record(
                 if str(_task_value(task, "result", "") or "") == document["result"]
                 else "stale"
             )
-        if (
+        current_run_id = _task_value(task, "current_run_id", None)
-            status != "running"
+        completion_guard: dict[str, int]
-            or _task_value(task, "current_run_id", None) != identity.run_id
+        if status == "running" and current_run_id == identity.run_id:
-        ):
+            completion_guard = {"expected_run_id": identity.run_id}
+        elif status in {"ready", "blocked"} and current_run_id is None:
+            # The journal may have survived an older post-persistence error
+            # path that ended its run. The patched DB verifies atomically that
+            # this is still the latest ended run before allowing completion.
+            completion_guard = {"replay_ended_run_id": identity.run_id}
+        else:
             return "stale"
         completed = kanban_db.complete_task(
             conn,
@@ -1232,7 +1346,7 @@ def _finalize_terminal_record(
             result=document["result"],
             summary=document["summary"],
             metadata=document["metadata"],
-            expected_run_id=identity.run_id,
+            **completion_guard,
         )
         return "committed" if completed else "pending"
 
@@ -1255,11 +1369,11 @@ def recover_pending_finalizations() -> int:
             continue
         if not _terminal_record_valid(record):
             _quarantine_terminal(path, identity, "malformed-payload")
-            _recover_quarantined_run(kanban_db, identity, "malformed-payload")
+            _recover_exact_run(kanban_db, identity, "malformed-payload")
             continue
         if not _terminal_record_valid(record, identity):
             _quarantine_terminal(path, identity, "foreign-identity")
-            _recover_quarantined_run(kanban_db, identity, "foreign-identity")
+            _recover_exact_run(kanban_db, identity, "foreign-identity")
             continue
         try:
             outcome = _finalize_terminal_record(kanban_db, path, record)
@@ -1270,7 +1384,7 @@ def recover_pending_finalizations() -> int:
             recovered += 1
         elif outcome in {"invalid", "foreign", "stale"}:
             _quarantine_terminal(path, identity, outcome)
-            _recover_quarantined_run(kanban_db, identity, outcome)
+            _recover_exact_run(kanban_db, identity, outcome)
     return recovered
 
 
@@ -1329,6 +1443,9 @@ def gc_lane_artifacts(
             except OSError:
                 continue
             if not stat.S_ISREG(file_stat.st_mode):
+                path.unlink()
+                removed += 1
+                board_removed += 1
                 continue
             if max_age_seconds >= 0 and current - file_stat.st_mtime > max_age_seconds:
                 path.unlink()
@@ -1631,28 +1748,44 @@ def execute_claim(board: str, task_id: str) -> None:
                     and result.returncode == 0
                     and completion_problem is None
                 ):
+                    terminal_file = _terminal_path(state_file, run_id, "pending")
                     metadata["terminal_record"] = str(
                         _terminal_path(state_file, run_id, "committed")
                     )
-                    terminal_file, terminal_record = _write_terminal_record(
-                        state_file,
-                        board=board,
-                        task_id=task_id,
-                        run_id=run_id,
-                        structured=structured,
-                        summary=str(structured.get("summary") or "Completed"),
-                        metadata=metadata,
-                    )
                     try:
+                        terminal_file, terminal_record = _write_terminal_record(
+                            state_file,
+                            board=board,
+                            task_id=task_id,
+                            run_id=run_id,
+                            structured=structured,
+                            summary=str(structured.get("summary") or "Completed"),
+                            metadata=metadata,
+                        )
                         outcome = _finalize_terminal_record(
                             kanban_db,
                             terminal_file,
                             terminal_record,
                         )
                     except Exception as error:
+                        identity = _terminal_identity(terminal_file)
+                        replayable = bool(
+                            identity is not None
+                            and _terminal_record_valid(
+                                _load_terminal_json(terminal_file, identity),
+                                identity,
+                            )
+                        )
+                        if not replayable:
+                            _recover_exact_run(
+                                kanban_db,
+                                identity,
+                                f"persistence raised {type(error).__name__}",
+                            )
                         raise TerminalFinalizationPending(
-                            "accepted worker result is durably journaled; "
+                            "accepted worker result remains outside the generic block path; "
-                            f"Kanban finalization raised {type(error).__name__}"
+                            f"terminal replayable={replayable}; persistence/finalization "
+                            f"raised {type(error).__name__}"
                         ) from error
                     if outcome != "committed":
                         raise TerminalFinalizationPending(

testing/tests/test_hermes_cli_lanes.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import errno
 import importlib.util
+import hashlib
 import json
 import os
 import signal
@@ -864,6 +866,55 @@ def _completed_result(summary: str = "done") -> dict:
     }
 
 
+def test_terminal_recovery_can_complete_the_exact_latest_ended_run(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    state_file = lanes.state_path("cassandra", "t_ended")
+    pending, _record = lanes._write_terminal_record(
+        state_file,
+        board="cassandra",
+        task_id="t_ended",
+        run_id=17,
+        structured=_completed_result("accepted before legacy block"),
+        summary="accepted before legacy block",
+        metadata={},
+    )
+    task = SimpleNamespace(
+        id="t_ended",
+        status="blocked",
+        result=None,
+        current_run_id=None,
+        assignee="cli-auto",
+    )
+    guards = []
+
+    class Connection:
+        def close(self):
+            return None
+
+    def complete_task(_conn, _task_id, **kwargs):
+        guards.append(kwargs)
+        task.status = "done"
+        task.result = 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 guards[0]["replay_ended_run_id"] == 17
+    assert "expected_run_id" not in guards[0]
+    assert task.status == "done"
+    assert not pending.exists()
+
+
 def test_complete_exception_after_journal_is_replayable_and_never_blocks(
     tmp_path: Path,
     monkeypatch,
@@ -937,6 +988,103 @@ def test_complete_exception_after_journal_is_replayable_and_never_blocks(
     assert not pending[0].exists()
 
 
+def test_terminal_replace_then_directory_fsync_enospc_replays_after_restart(
+    tmp_path: Path,
+    monkeypatch,
+):
+    state_root = tmp_path / "cli-lanes"
+    monkeypatch.setattr(lanes, "STATE_ROOT", state_root)
+    task = SimpleNamespace(
+        id="t_enospc",
+        status="running",
+        result=None,
+        current_run_id=52,
+        assignee="cli-auto",
+        max_runtime_seconds=60,
+    )
+    blocks = []
+    comments = []
+    completions = []
+
+    class Connection:
+        def close(self):
+            return None
+
+    def complete_task(_conn, _task_id, **kwargs):
+        completions.append(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,
+        worker_log_path=lambda _task_id, board: tmp_path / "worker.log",
+        _resolve_worktree_workspace=lambda _task, board: (tmp_path, "wt/t_enospc"),
+        set_branch_name=lambda *_args: None,
+        set_workspace_path=lambda *_args: None,
+        build_worker_context=lambda *_args: "Finish without losing the result.",
+        heartbeat_worker=lambda *_args, **_kwargs: True,
+        add_comment=lambda _conn, _task_id, _author, body: comments.append(body),
+        complete_task=complete_task,
+        block_task=lambda *_args, **kwargs: blocks.append(kwargs),
+    )
+    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
+    monkeypatch.setattr(
+        lanes,
+        "select_route",
+        lambda *_args, **_kwargs: lanes.Route(
+            "codex", "gpt-5.6-sol", "high", "codex-high", "test", "test", 1, ()
+        ),
+    )
+    monkeypatch.setattr(
+        lanes,
+        "run_provider",
+        lambda *_args, **_kwargs: lanes.ProcessResult(
+            0, "", _completed_result("persisted before ENOSPC"), False
+        ),
+    )
+    real_replace = lanes.os.replace
+    real_fsync_directory = lanes._fsync_directory
+    terminal_replaced = {"value": False}
+    fail_once = {"value": True}
+
+    def replace_then_mark(source, destination):
+        real_replace(source, destination)
+        if str(destination).endswith(".terminal.pending.json"):
+            terminal_replaced["value"] = True
+
+    def fail_after_terminal_replace(directory):
+        if terminal_replaced["value"] and fail_once["value"]:
+            fail_once["value"] = False
+            raise OSError(errno.ENOSPC, "no space after terminal rename")
+        real_fsync_directory(directory)
+
+    monkeypatch.setattr(lanes.os, "replace", replace_then_mark)
+    monkeypatch.setattr(lanes, "_fsync_directory", fail_after_terminal_replace)
+
+    lanes.execute_claim("cassandra", "t_enospc")
+
+    pending = list((state_root / "cassandra").glob("*.terminal.pending.json"))
+    assert terminal_replaced["value"] is True
+    assert fail_once["value"] is False
+    assert len(pending) == 1
+    assert blocks == []
+    assert completions == []
+    assert task.status == "running"
+    assert any("terminal replayable=True" in body for body in comments)
+
+    # A restarted runner sees the exact pending journal and completes the run.
+    assert lanes.recover_pending_finalizations() == 1
+    assert len(completions) == 1
+    assert task.status == "done"
+    assert not pending[0].exists()
+    committed = list((state_root / "cassandra").glob("*.terminal.committed.json"))
+    assert len(committed) == 1
+
+
 def test_terminal_payload_cannot_select_a_different_board(
     tmp_path: Path,
     monkeypatch,
@@ -1050,6 +1198,147 @@ def test_malformed_exact_run_journal_is_quarantined_and_reclaimed(
     assert quarantined[0].stat().st_mode & 0o777 == 0o600
 
 
+@pytest.mark.parametrize(
+    "structured",
+    [
+        {"status": "completed", "blockers": {}},
+        {**_completed_result(), "changed_files": "src/a.py"},
+        {**_completed_result(), "unexpected": True},
+        {**_completed_result(), "blockers": ["work remains"]},
+        {**_completed_result("tests are still running")},
+    ],
+)
+def test_terminal_record_requires_exact_completed_result_contract(structured):
+    record = {
+        "board": "cassandra",
+        "task_id": "t_schema",
+        "expected_run_id": 1,
+        "result": json.dumps(structured),
+        "summary": str(structured.get("summary") or "done"),
+        "metadata": {},
+        "kanban_state": "pending",
+    }
+
+    assert lanes._terminal_record_valid(record) is False
+
+
+def test_minimal_completed_result_with_mapping_blockers_quarantines_without_completion(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    path = lanes._terminal_path(lanes.state_path("cassandra", "t_schema"), 6)
+    lanes.atomic_json(
+        path,
+        {
+            "board": "cassandra",
+            "task_id": "t_schema",
+            "expected_run_id": 6,
+            "result": json.dumps({"status": "completed", "blockers": {}}),
+            "summary": "done",
+            "metadata": {},
+            "kanban_state": "pending",
+        },
+    )
+    task = SimpleNamespace(
+        id="t_schema", status="running", current_run_id=6, assignee="cli-auto"
+    )
+    completed = []
+    reclaimed = []
+
+    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: completed.append(True),
+        reclaim_task=lambda *_args, **_kwargs: (reclaimed.append(True) or True),
+    )
+    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
+
+    assert lanes.recover_pending_finalizations() == 0
+    assert completed == []
+    assert reclaimed == [True]
+    assert not path.exists()
+    quarantined = list((path.parent / "quarantine").glob("*.quarantine"))
+    assert len(quarantined) == 1
+    assert quarantined[0].stat().st_mode & 0o777 == 0o600
+
+
+def test_quarantine_avoids_symlink_and_mode_collision_destinations(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    path = lanes._terminal_path(lanes.state_path("cassandra", "t_collision"), 2)
+    path.parent.mkdir(parents=True)
+    path.write_bytes(b"invalid")
+    quarantine = path.parent / "quarantine"
+    quarantine.mkdir()
+    fingerprint = hashlib.sha256(b"invalid").hexdigest()[:16]
+    base = f"{path.name}.malformed-payload.{fingerprint}"
+    victim = tmp_path / "victim"
+    victim.write_text("unchanged", encoding="utf-8")
+    (quarantine / f"{base}.0.quarantine").symlink_to(victim)
+    collision = quarantine / f"{base}.1.quarantine"
+    collision.write_text("attacker collision", encoding="utf-8")
+    collision.chmod(0o644)
+    monkeypatch.setattr(
+        lanes.os,
+        "chmod",
+        lambda *_args, **_kwargs: pytest.fail("quarantine must not chmod foreign inodes"),
+    )
+
+    destination = lanes._quarantine_terminal(
+        path,
+        lanes._terminal_identity(path),
+        "malformed-payload",
+    )
+
+    assert not path.exists()
+    assert destination.name == f"{base}.2.quarantine"
+    assert destination.read_bytes() == b"invalid"
+    assert destination.stat().st_mode & 0o777 == 0o600
+    assert victim.read_text(encoding="utf-8") == "unchanged"
+    assert collision.read_text(encoding="utf-8") == "attacker collision"
+    assert collision.stat().st_mode & 0o777 == 0o644
+
+
+def test_hardlinked_terminal_source_is_never_chmodded_or_copied_as_data(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    path = lanes._terminal_path(lanes.state_path("cassandra", "t_hardlink"), 5)
+    path.parent.mkdir(parents=True)
+    foreign = tmp_path / "foreign"
+    foreign.write_text("foreign inode contents", encoding="utf-8")
+    foreign.chmod(0o644)
+    os.link(foreign, path)
+    monkeypatch.setattr(
+        lanes.os,
+        "chmod",
+        lambda *_args, **_kwargs: pytest.fail("hardlinked source must not be chmodded"),
+    )
+
+    destination = lanes._quarantine_terminal(
+        path,
+        lanes._terminal_identity(path),
+        "malformed-payload",
+    )
+
+    assert not path.exists()
+    assert foreign.read_text(encoding="utf-8") == "foreign inode contents"
+    assert foreign.stat().st_mode & 0o777 == 0o644
+    assert destination.stat().st_mode & 0o777 == 0o600
+    diagnostic = json.loads(destination.read_text(encoding="utf-8"))
+    assert diagnostic["source_kind"] == "hardlink"
+    assert "foreign inode contents" not in destination.read_text(encoding="utf-8")
+
+
 def test_recovery_does_not_parse_committed_journals_every_tick(
     tmp_path: Path,
     monkeypatch,
@@ -1127,6 +1416,8 @@ def test_artifact_gc_prunes_by_age_without_touching_pending_journals(
     ]
     for artifact in old_artifacts:
         artifact.write_text("{}", encoding="utf-8")
+    quarantine_symlink = quarantine / "attacker.quarantine"
+    quarantine_symlink.symlink_to(tmp_path / "missing-target")
     pending = board / "t.run-1.terminal.pending.json"
     pending.write_text("{}", encoding="utf-8")
     old_time = 100.0
@@ -1136,8 +1427,9 @@ def test_artifact_gc_prunes_by_age_without_touching_pending_journals(
 
     assert lanes.gc_lane_artifacts(
         now=1000.0, max_age_seconds=10, max_count=100, max_bytes=10000
-    ) == len(old_artifacts)
+    ) == len(old_artifacts) + 1
     assert not any(artifact.exists() for artifact in old_artifacts)
+    assert not quarantine_symlink.is_symlink()
     assert pending.exists()
 
 
@@ -1189,9 +1481,10 @@ def test_artifact_gc_is_interval_gated(
                     "status": "completed",
                     "summary": "done",
                     "changed_files": ["src/a.py"],
-                    "tests_run": ["pytest -q"],
+                        "tests_run": ["pytest -q"],
-                    "artifacts": ["reports/result.json"],
+                        "artifacts": ["reports/result.json"],
-                    "blockers": [],
+                        "findings": [],
+                        "blockers": [],
                 },
                 False,
             ),
@@ -1205,9 +1498,10 @@ def test_artifact_gc_is_interval_gated(
                     "status": "completed",
                     "summary": "The full test suite is still running.",
                     "changed_files": ["src/a.py"],
-                    "tests_run": ["pytest -q — in progress"],
+                        "tests_run": ["pytest -q — in progress"],
-                    "artifacts": [],
+                        "artifacts": [],
-                    "blockers": [],
+                        "findings": [],
+                        "blockers": [],
                 },
                 False,
             ),
@@ -1369,6 +1663,7 @@ def test_goal_card_continues_after_local_judge_rejects_progress(
                 "changed_files": ["src/a.py"],
                 "tests_run": ["pytest focused: passed"],
                 "artifacts": [],
+                "findings": [],
                 "blockers": [],
             },
             False,
@@ -1382,6 +1677,7 @@ def test_goal_card_continues_after_local_judge_rejects_progress(
                 "changed_files": ["src/a.py"],
                 "tests_run": ["pytest full: passed"],
                 "artifacts": [],
+                "findings": [],
                 "blockers": [],
             },
             False,
@@ -1614,6 +1910,7 @@ def test_restart_provider_change_includes_explicit_workspace_handoff(tmp_path: P
                     "changed_files": [],
                     "tests_run": [],
                     "artifacts": [],
+                    "findings": [],
                     "blockers": [],
                 },
                 False,