hermes: harden replay and decomposition races

dockerfiles/hermes-execution-safety-regression.py

@@ -167,22 +167,45 @@ class AutomaticDecompositionSafetyTests(unittest.TestCase):
         self.assertEqual(completions.calls, 1)
         self.assertEqual(self._status(task_id), "ready")
 
-    def test_execution_history_gained_during_llm_call_blocks_commit(self) -> None:
+    def _insert_run(self, task_id: str) -> None:
+        with kanban_db.connect_closing() as connection:
+            kanban_db._synthesize_ended_run(
+                connection,
+                task_id,
+                outcome="reclaimed",
+                summary="concurrent execution evidence",
+            )
+
+    def test_execution_history_gained_during_single_llm_call_blocks_commit(self) -> None:
+        task_id = kanban_db.create_task(
+            self.connection,
+            title="concurrent single objective",
+            triage=True,
+        )
+        completions = self._client(
+            {
+                "fanout": False,
+                "title": "redundant rewrite",
+                "body": "must not be committed",
+                "assignee": "default",
+            },
+            before_response=lambda: self._insert_run(task_id),
+        )
+
+        outcome = kanban_decompose.decompose_task(task_id, automatic=True)
+
+        self.assertFalse(outcome.ok)
+        self.assertIn("gained execution history", outcome.reason)
+        self.assertEqual(completions.calls, 1)
+        self.assertEqual(self._status(task_id), "triage")
+
+    def test_execution_history_gained_during_fanout_llm_call_blocks_commit(self) -> None:
         task_id = kanban_db.create_task(
             self.connection,
             title="concurrent triage objective",
             triage=True,
         )
 
-        def add_run() -> None:
-            with kanban_db.connect_closing() as connection:
-                kanban_db._synthesize_ended_run(
-                    connection,
-                    task_id,
-                    outcome="reclaimed",
-                    summary="concurrent execution evidence",
-                )
-
         completions = self._client(
             {
                 "fanout": True,
@@ -195,7 +218,7 @@ class AutomaticDecompositionSafetyTests(unittest.TestCase):
                     }
                 ],
             },
-            before_response=add_run,
+            before_response=lambda: self._insert_run(task_id),
         )
         before = len(kanban_db.list_tasks(self.connection))
 

dockerfiles/patch-hermes-execution-safety.py

@@ -15,6 +15,106 @@ def replace_once(source: str, before: str, after: str, label: str) -> str:
 
 
 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")
+
+specify_signature_before = '''def specify_triage_task(
+    conn: sqlite3.Connection,
+    task_id: str,
+    *,
+    title: Optional[str] = None,
+    body: Optional[str] = None,
+    assignee: Optional[str] = None,
+    author: Optional[str] = None,
+) -> bool:
+'''
+specify_signature_after = '''def specify_triage_task(
+    conn: sqlite3.Connection,
+    task_id: str,
+    *,
+    title: Optional[str] = None,
+    body: Optional[str] = None,
+    assignee: Optional[str] = None,
+    author: Optional[str] = None,
+    require_no_runs: bool = False,
+) -> bool:
+'''
+db = replace_once(
+    db,
+    specify_signature_before,
+    specify_signature_after,
+    "transactional specify signature",
+)
+
+specify_guard_before = '''        if existing is None:
+            return False
+        sets: list[str] = ["status = 'todo'"]
+'''
+specify_guard_after = '''        if existing is None:
+            return False
+        if require_no_runs and conn.execute(
+            "SELECT 1 FROM task_runs WHERE task_id = ? LIMIT 1",
+            (task_id,),
+        ).fetchone() is not None:
+            return False
+        sets: list[str] = ["status = 'todo'"]
+'''
+db = replace_once(
+    db,
+    specify_guard_before,
+    specify_guard_after,
+    "transactional specify execution-history guard",
+)
+
+decompose_signature_before = '''def decompose_triage_task(
+    conn: sqlite3.Connection,
+    task_id: str,
+    *,
+    root_assignee: Optional[str],
+    children: list[dict],
+    author: Optional[str] = None,
+    auto_promote: bool = True,
+) -> Optional[list[str]]:
+'''
+decompose_signature_after = '''def decompose_triage_task(
+    conn: sqlite3.Connection,
+    task_id: str,
+    *,
+    root_assignee: Optional[str],
+    children: list[dict],
+    author: Optional[str] = None,
+    auto_promote: bool = True,
+    require_no_runs: bool = False,
+) -> Optional[list[str]]:
+'''
+db = replace_once(
+    db,
+    decompose_signature_before,
+    decompose_signature_after,
+    "transactional decomposition signature",
+)
+
+decompose_guard_before = '''        if root_row["status"] != "triage":
+            return None
+        tenant = root_row["tenant"]
+'''
+decompose_guard_after = '''        if root_row["status"] != "triage":
+            return None
+        if require_no_runs and conn.execute(
+            "SELECT 1 FROM task_runs WHERE task_id = ? LIMIT 1",
+            (task_id,),
+        ).fetchone() is not None:
+            return None
+        tenant = root_row["tenant"]
+'''
+db = replace_once(
+    db,
+    decompose_guard_before,
+    decompose_guard_after,
+    "transactional decomposition execution-history guard",
+)
+db_path.write_text(db, encoding="utf-8")
+
 decompose_path = source_root / "hermes_cli/kanban_decompose.py"
 decompose = decompose_path.read_text(encoding="utf-8")
 
@@ -87,17 +187,20 @@ decompose = replace_once(
     "auto-decompose preflight guard",
 )
 
-single_before = '''        with kb.connect_closing() as conn:
+single_before = '''                author=audit_author,
-            ok = kb.specify_triage_task(
-'''
-single_after = '''        if automatic and _has_execution_history(task_id):
-            return DecomposeOutcome(
-                task_id,
-                False,
-                "task gained execution history and requires deliberate manual triage",
             )
-        with kb.connect_closing() as conn:
+        if not ok:
-            ok = kb.specify_triage_task(
+'''
+single_after = '''                author=audit_author,
+                require_no_runs=automatic,
+            )
+        if not ok:
+            if automatic and _has_execution_history(task_id):
+                return DecomposeOutcome(
+                    task_id,
+                    False,
+                    "task gained execution history and requires deliberate manual triage",
+                )
 '''
 decompose = replace_once(
     decompose,
@@ -106,25 +209,43 @@ decompose = replace_once(
     "auto-decompose single-task commit guard",
 )
 
-fanout_before = '''    try:
+fanout_before = '''                author=audit_author,
-        with kb.connect_closing() as conn:
+                auto_promote=auto_promote,
-            child_ids = kb.decompose_triage_task(
+            )
 '''
-fanout_after = '''    if automatic and _has_execution_history(task_id):
+fanout_after = '''                author=audit_author,
-        return DecomposeOutcome(
+                auto_promote=auto_promote,
-            task_id,
+                require_no_runs=automatic,
-            False,
+            )
-            "task gained execution history and requires deliberate manual triage",
-        )
-    try:
-        with kb.connect_closing() as conn:
-            child_ids = kb.decompose_triage_task(
 '''
 decompose = replace_once(
     decompose,
     fanout_before,
     fanout_after,
-    "auto-decompose fanout commit guard",
+    "auto-decompose transactional fanout guard",
+)
+
+fanout_outcome_before = '''    if child_ids is None:
+        return DecomposeOutcome(
+            task_id, False, "task moved out of triage before decomposition",
+        )
+'''
+fanout_outcome_after = '''    if child_ids is None:
+        if automatic and _has_execution_history(task_id):
+            return DecomposeOutcome(
+                task_id,
+                False,
+                "task gained execution history and requires deliberate manual triage",
+            )
+        return DecomposeOutcome(
+            task_id, False, "task moved out of triage before decomposition",
+        )
+'''
+decompose = replace_once(
+    decompose,
+    fanout_outcome_before,
+    fanout_outcome_after,
+    "auto-decompose fanout rejection reason",
 )
 decompose_path.write_text(decompose, encoding="utf-8")
 

services/hermes/scripts/cli_lane_runner.py

@@ -4,12 +4,14 @@
 from __future__ import annotations
 
 import concurrent.futures
+import hashlib
 import json
 import os
 import re
 import selectors
 import signal
 import sqlite3
+import stat
 import subprocess
 import sys
 import threading
@@ -43,12 +45,17 @@ HEARTBEAT_SECONDS = 20
 PROVIDER_HEALTH_MAX_AGE_SECONDS = 5 * 60
 PROVIDER_AUTH_FAILURE_MAX_AGE_SECONDS = 12 * 60 * 60
 KANBAN_STORAGE_ATTEMPTS = 5
+ARTIFACT_GC_INTERVAL_SECONDS = 5 * 60
+ARTIFACT_RETENTION_AGE_SECONDS = 30 * 24 * 60 * 60
+ARTIFACT_RETENTION_COUNT = 256
+ARTIFACT_RETENTION_BYTES = 128 * 1024 * 1024
 PROVIDER_HEALTH_PATHS = {
     "codex": DATA_ROOT / "provider-health/codex.json",
     "claude": DATA_ROOT / "provider-health/claude.json",
 }
 WORKTREE_LOCK = threading.Lock()
 BOARD_CORRUPTION_ERRORS: dict[str, str] = {}
+LAST_ARTIFACT_GC = 0.0
 CAPACITY_PATTERN = re.compile(
     r"(?:rate.?limit|capacity|overload|usage.?limit|quota|credit|exhaust|429|529|authentication|oauth|token.*expired)",
     re.I,
@@ -127,22 +134,65 @@ class TerminalFinalizationPending(RuntimeError):
     """An accepted worker result is durable but not committed to Kanban yet."""
 
 
+@dataclass(frozen=True)
+class TerminalIdentity:
+    """Replay authority derived only from a journal's directory and filename."""
+
+    board: str
+    task_id: str
+    run_id: int
+    state: str
+
+
 def utc_now() -> str:
     return datetime.now(timezone.utc).isoformat()
 
 
 def atomic_json(path: Path, value: dict[str, Any], mode: int = 0o600) -> None:
-    """Durably replace a small non-secret state document."""
+    """Durably replace a small state document without following temp symlinks."""
-    path.parent.mkdir(parents=True, exist_ok=True)
+    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
-    temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
+    os.chmod(path.parent, 0o700, follow_symlinks=False)
-    temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp")
-    temporary.chmod(mode)
+    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
-    os.replace(temporary, path)
+    flags |= getattr(os, "O_NOFOLLOW", 0)
+    descriptor = None
+    try:
+        descriptor = os.open(temporary, flags, mode)
+        with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
+            descriptor = None
+            json.dump(value, stream, indent=2, sort_keys=True)
+            stream.write("\n")
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.chmod(temporary, mode, follow_symlinks=False)
+        os.replace(temporary, path)
+        _fsync_directory(path.parent)
+    finally:
+        if descriptor is not None:
+            os.close(descriptor)
+        try:
+            temporary.unlink()
+        except FileNotFoundError:
+            pass
+
+
+def _fsync_directory(directory: Path) -> None:
+    """Persist directory-entry changes after an atomic rename or quarantine."""
+    flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
+    flags |= getattr(os, "O_NOFOLLOW", 0)
+    descriptor = os.open(directory, flags)
+    try:
+        os.fsync(descriptor)
+    finally:
+        os.close(descriptor)
 
 
 def load_json(path: Path) -> dict[str, Any]:
     try:
-        value = json.loads(path.read_text(encoding="utf-8"))
+        flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+        descriptor = os.open(path, flags)
+        with os.fdopen(descriptor, "r", encoding="utf-8") as stream:
+            value = json.load(stream)
     except (OSError, json.JSONDecodeError):
         return {}
     return value if isinstance(value, dict) else {}
@@ -313,10 +363,80 @@ def _candidate_path(state_file: Path, run_id: Any, sequence: int) -> Path:
     )
 
 
-def _terminal_path(state_file: Path, run_id: Any) -> Path:
+def _terminal_path(
-    """Return the replay journal path for an accepted terminal response."""
+    state_file: Path,
+    run_id: Any,
+    state: str = "pending",
+) -> Path:
+    """Return the identity-bound replay path for an accepted terminal response."""
+    if state not in {"pending", "committed"}:
+        raise ValueError(f"invalid terminal journal state: {state}")
     safe_run = re.sub(r"[^a-zA-Z0-9_.-]+", "-", str(run_id or "unknown"))
-    return state_file.with_name(f"{state_file.stem}.run-{safe_run}.terminal.json")
+    return state_file.with_name(
+        f"{state_file.stem}.run-{safe_run}.terminal.{state}.json"
+    )
+
+
+_TERMINAL_NAME = re.compile(
+    r"^(?P<task>[a-zA-Z0-9_.-]+)\.run-(?P<run>[0-9]+)\."
+    r"terminal\.(?P<state>pending|committed)\.json$"
+)
+_SAFE_BOARD = re.compile(r"^[a-zA-Z0-9_.-]+$")
+
+
+def _terminal_identity(path: Path) -> TerminalIdentity | None:
+    """Parse replay authority lexically before opening a journal or board."""
+    try:
+        relative = path.relative_to(STATE_ROOT)
+    except ValueError:
+        return None
+    if len(relative.parts) != 2:
+        return None
+    board, filename = relative.parts
+    match = _TERMINAL_NAME.fullmatch(filename)
+    try:
+        board_stat = path.parent.stat(follow_symlinks=False)
+    except OSError:
+        return None
+    if (
+        not match
+        or not stat.S_ISDIR(board_stat.st_mode)
+        or not _SAFE_BOARD.fullmatch(board)
+        or board in {".", ".."}
+        or match.group("task") in {".", ".."}
+    ):
+        return None
+    return TerminalIdentity(
+        board=board,
+        task_id=match.group("task"),
+        run_id=int(match.group("run")),
+        state=match.group("state"),
+    )
+
+
+def _load_terminal_json(path: Path, identity: TerminalIdentity) -> dict[str, Any]:
+    """Read through the already-validated board directory without symlink hops."""
+    if _terminal_identity(path) != identity:
+        return {}
+    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)
+        descriptor = os.open(path.name, file_flags, dir_fd=directory)
+        with os.fdopen(descriptor, "r", encoding="utf-8") as stream:
+            descriptor = None
+            value = json.load(stream)
+    except (OSError, json.JSONDecodeError):
+        return {}
+    finally:
+        if descriptor is not None:
+            os.close(descriptor)
+        if directory is not None:
+            os.close(directory)
+    return value if isinstance(value, dict) else {}
 
 
 def _persist_candidate(
@@ -381,7 +501,10 @@ def _write_terminal_record(
     return path, record
 
 
-def _terminal_record_valid(record: dict[str, Any]) -> bool:
+def _terminal_record_valid(
+    record: dict[str, Any],
+    identity: TerminalIdentity | None = None,
+) -> bool:
     """Reject malformed or non-terminal replay journals without side effects."""
     if not isinstance(record, dict):
         return False
@@ -399,10 +522,18 @@ def _terminal_record_valid(record: dict[str, Any]) -> bool:
         structured = json.loads(record["result"])
     except (TypeError, json.JSONDecodeError):
         return False
-    return (
+    valid = (
         isinstance(structured, dict)
         and structured.get("status") == "completed"
         and not structured.get("blockers")
+        and record.get("kanban_state") in {"pending", "committed"}
+    )
+    if not valid or identity is None:
+        return valid
+    return (
+        record.get("board") == identity.board
+        and record.get("task_id") == identity.task_id
+        and record.get("expected_run_id") == identity.run_id
     )
 
 
@@ -945,21 +1076,142 @@ def _board_call(
     raise last_storage_error
 
 
+def _quarantine_terminal(path: Path, identity: TerminalIdentity | None, reason: str) -> Path:
+    """Move an unusable journal aside with a deterministic, private name."""
+    board_dir = path.parent
+    try:
+        board_stat = board_dir.stat(follow_symlinks=False)
+    except OSError:
+        board_stat = None
+    if board_stat is None or not stat.S_ISDIR(board_stat.st_mode):
+        print(
+            f"refused terminal journal path through unsafe board directory: {path}",
+            file=sys.stderr,
+            flush=True,
+        )
+        return path
+    quarantine = board_dir / "quarantine"
+    quarantine.mkdir(parents=True, exist_ok=True, mode=0o700)
+    os.chmod(quarantine, 0o700, follow_symlinks=False)
+    try:
+        source_stat = path.stat(follow_symlinks=False)
+        regular = stat.S_ISREG(source_stat.st_mode)
+    except OSError:
+        regular = False
+    try:
+        if not regular:
+            raise OSError("journal is not a regular file")
+        flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+        descriptor = os.open(path, flags)
+        with os.fdopen(descriptor, "rb") as stream:
+            fingerprint = hashlib.sha256(stream.read()).hexdigest()[:16]
+    except OSError:
+        fingerprint = hashlib.sha256(path.name.encode("utf-8")).hexdigest()[:16]
+    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(),
+                },
+            )
+    print(
+        "quarantined terminal journal "
+        f"{path.name}: {reason}; identity={identity or 'unparseable'}",
+        file=sys.stderr,
+        flush=True,
+    )
+    return destination
+
+
+def _recover_quarantined_run(
+    kanban_db: Any,
+    identity: TerminalIdentity | None,
+    reason: str,
+) -> bool:
+    """Make an exact external run retryable after its journal is quarantined."""
+    if identity is None:
+        return False
+
+    def operation(conn: Any) -> bool:
+        task = kanban_db.get_task(conn, identity.task_id)
+        if task is None:
+            return False
+        if (
+            str(_task_value(task, "status", "")) != "running"
+            or _task_value(task, "current_run_id", None) != identity.run_id
+            or not _external(task)
+        ):
+            return False
+        return bool(
+            kanban_db.reclaim_task(
+                conn,
+                identity.task_id,
+                reason=f"terminal journal quarantined ({reason}); exact run may retry",
+            )
+        )
+
+    try:
+        recovered = bool(_board_call(kanban_db, identity.board, operation))
+    except Exception as error:
+        _record_board_access_error(identity.board, error)
+        return False
+    if recovered:
+        print(
+            f"reclaimed {identity.board}/{identity.task_id} run {identity.run_id} "
+            f"after terminal journal quarantine: {reason}",
+            file=sys.stderr,
+            flush=True,
+        )
+    return recovered
+
+
+def _commit_terminal_file(path: Path, identity: TerminalIdentity, document: dict[str, Any]) -> Path:
+    """Persist committed state, then atomically retire a pending journal."""
+    committed = _terminal_path(
+        state_path(identity.board, identity.task_id),
+        identity.run_id,
+        "committed",
+    )
+    document["kanban_state"] = "committed"
+    document["committed_at"] = utc_now()
+    atomic_json(path, document)
+    os.replace(path, committed)
+    os.chmod(committed, 0o600, follow_symlinks=False)
+    _fsync_directory(committed.parent)
+    return committed
+
+
 def _finalize_terminal_record(
     kanban_db: Any,
     path: Path,
-    record: dict[str, Any] | None = None,
+    _record: dict[str, Any] | None = None,
 ) -> str:
     """Commit one exact-run terminal journal, or leave it safely pending."""
-    document = dict(record or load_json(path))
+    identity = _terminal_identity(path)
+    if identity is None or identity.state != "pending":
+        return "invalid"
+    document = _load_terminal_json(path, identity)
     if not _terminal_record_valid(document):
         return "invalid"
-    board = str(document["board"])
+    if not _terminal_record_valid(document, identity):
-    task_id = str(document["task_id"])
+        return "foreign"
-    expected_run_id = int(document["expected_run_id"])
 
     def operation(conn: Any) -> str:
-        task = kanban_db.get_task(conn, task_id)
+        task = kanban_db.get_task(conn, identity.task_id)
         if task is None:
             return "stale"
         status = str(_task_value(task, "status", ""))
@@ -971,24 +1223,22 @@ def _finalize_terminal_record(
             )
         if (
             status != "running"
-            or _task_value(task, "current_run_id", None) != expected_run_id
+            or _task_value(task, "current_run_id", None) != identity.run_id
         ):
             return "stale"
         completed = kanban_db.complete_task(
             conn,
-            task_id,
+            identity.task_id,
             result=document["result"],
             summary=document["summary"],
             metadata=document["metadata"],
-            expected_run_id=expected_run_id,
+            expected_run_id=identity.run_id,
         )
         return "committed" if completed else "pending"
 
-    outcome = str(_board_call(kanban_db, board, operation))
+    outcome = str(_board_call(kanban_db, identity.board, operation))
     if outcome == "committed":
-        document["kanban_state"] = "committed"
+        _commit_terminal_file(path, identity, document)
-        document["committed_at"] = utc_now()
-        atomic_json(path, document)
     return outcome
 
 
@@ -997,18 +1247,30 @@ def recover_pending_finalizations() -> int:
     from hermes_cli import kanban_db
 
     recovered = 0
-    for path in sorted(STATE_ROOT.glob("*/*.terminal.json")):
+    for path in sorted(STATE_ROOT.glob("*/*.terminal.pending.json")):
-        record = load_json(path)
+        identity = _terminal_identity(path)
-        if record.get("kanban_state") == "committed":
+        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")
+            _recover_quarantined_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")
             continue
         try:
             outcome = _finalize_terminal_record(kanban_db, path, record)
         except Exception as error:
-            board = str(record.get("board") or "unknown")
+            _record_board_access_error(identity.board, error)
-            _record_board_access_error(board, error)
             continue
         if outcome == "committed":
             recovered += 1
+        elif outcome in {"invalid", "foreign", "stale"}:
+            _quarantine_terminal(path, identity, outcome)
+            _recover_quarantined_run(kanban_db, identity, outcome)
     return recovered
 
 
@@ -1016,22 +1278,101 @@ def _has_pending_finalization(board: str, task_id: str, run_id: Any) -> bool:
     """Keep an exact run claimed while its accepted result awaits replay."""
     if not isinstance(run_id, int):
         return False
-    path = _terminal_path(state_path(board, task_id), run_id)
+    path = _terminal_path(state_path(board, task_id), run_id, "pending")
     try:
         path.stat()
     except FileNotFoundError:
         return False
     except OSError:
-        return True
+        return False
-    record = load_json(path)
+    identity = _terminal_identity(path)
-    if not _terminal_record_valid(record):
+    if identity is None:
-        return True
+        return False
-    identity = (
+    record = _load_terminal_json(path, identity)
-        record.get("board") == board
+    return _terminal_record_valid(record, identity)
-        and record.get("task_id") == task_id
+
-        and record.get("expected_run_id") == run_id
+
+def _artifact_gc_candidates(board_dir: Path) -> list[Path]:
+    """List only bounded artifact classes; pending journals are excluded."""
+    candidates: list[Path] = []
+    patterns = (
+        "*.provider-*.result.json",
+        "*.candidate-*.json",
+        "*.terminal.committed.json",
     )
-    return not identity or record.get("kanban_state") != "committed"
+    for pattern in patterns:
+        candidates.extend(board_dir.glob(pattern))
+    quarantine = board_dir / "quarantine"
+    if quarantine.is_dir():
+        candidates.extend(quarantine.glob("*.quarantine"))
+    return candidates
+
+
+def gc_lane_artifacts(
+    *,
+    now: float | None = None,
+    max_age_seconds: int = ARTIFACT_RETENTION_AGE_SECONDS,
+    max_count: int = ARTIFACT_RETENTION_COUNT,
+    max_bytes: int = ARTIFACT_RETENTION_BYTES,
+) -> int:
+    """Bound non-pending lane artifacts by age, count, and bytes per board."""
+    current = time.time() if now is None else now
+    removed = 0
+    for board_dir in STATE_ROOT.iterdir() if STATE_ROOT.is_dir() else ():
+        if not board_dir.is_dir() or board_dir.is_symlink():
+            continue
+        board_removed = 0
+        entries: list[tuple[Path, os.stat_result]] = []
+        for path in _artifact_gc_candidates(board_dir):
+            try:
+                file_stat = path.stat(follow_symlinks=False)
+            except OSError:
+                continue
+            if not stat.S_ISREG(file_stat.st_mode):
+                continue
+            if max_age_seconds >= 0 and current - file_stat.st_mtime > max_age_seconds:
+                path.unlink()
+                removed += 1
+                board_removed += 1
+            else:
+                entries.append((path, file_stat))
+        retained_count = 0
+        retained_bytes = 0
+        for path, file_stat in sorted(
+            entries, key=lambda item: item[1].st_mtime, reverse=True
+        ):
+            exceeds_count = max_count >= 0 and retained_count >= max_count
+            exceeds_bytes = (
+                max_bytes >= 0 and retained_bytes + file_stat.st_size > max_bytes
+            )
+            if exceeds_count or exceeds_bytes:
+                path.unlink()
+                removed += 1
+                board_removed += 1
+                continue
+            retained_count += 1
+            retained_bytes += file_stat.st_size
+        if board_removed:
+            _fsync_directory(board_dir)
+    return removed
+
+
+def maybe_gc_lane_artifacts(*, now: float | None = None) -> int:
+    """Run artifact retention on a bounded cadence, not every dispatcher tick."""
+    global LAST_ARTIFACT_GC
+    current = time.time() if now is None else now
+    if current - LAST_ARTIFACT_GC < ARTIFACT_GC_INTERVAL_SECONDS:
+        return 0
+    LAST_ARTIFACT_GC = current
+    try:
+        return gc_lane_artifacts(now=current)
+    except OSError as error:
+        print(
+            f"lane artifact retention deferred: {type(error).__name__}: {error}",
+            file=sys.stderr,
+            flush=True,
+        )
+        return 0
 
 
 def execute_claim(board: str, task_id: str) -> None:
@@ -1176,6 +1517,16 @@ def execute_claim(board: str, task_id: str) -> None:
                     heartbeat,
                     remaining,
                 )
+                candidate_file = None
+                if result.structured:
+                    candidate_file = _persist_candidate(
+                        state,
+                        state_file,
+                        dict(result.structured),
+                        route=route,
+                        returncode=result.returncode,
+                        goal_turn=goal_turn,
+                    )
                 if result.capacity_failure:
                     unavailable_provider = route.provider
                     retry_context = (
@@ -1206,6 +1557,15 @@ def execute_claim(board: str, task_id: str) -> None:
                         max(1, int(deadline - time.monotonic())),
                     )
                     route = fallback
+                    if result.structured:
+                        candidate_file = _persist_candidate(
+                            state,
+                            state_file,
+                            dict(result.structured),
+                            route=route,
+                            returncode=result.returncode,
+                            goal_turn=goal_turn,
+                        )
                 structured = result.structured
                 if structured:
                     structured = dict(structured)
@@ -1213,16 +1573,6 @@ def execute_claim(board: str, task_id: str) -> None:
                         workspace,
                         structured.get("artifacts"),
                     )
-                    candidate_file = _persist_candidate(
-                        state,
-                        state_file,
-                        structured,
-                        route=route,
-                        returncode=result.returncode,
-                        goal_turn=goal_turn,
-                    )
-                else:
-                    candidate_file = None
                 metadata = {
                     "executor": "direct-cli-lane",
                     "provider": route.provider,
@@ -1281,8 +1631,9 @@ 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)
+                    metadata["terminal_record"] = str(
-                    metadata["terminal_record"] = str(terminal_file)
+                        _terminal_path(state_file, run_id, "committed")
+                    )
                     terminal_file, terminal_record = _write_terminal_record(
                         state_file,
                         board=board,
@@ -1292,11 +1643,17 @@ def execute_claim(board: str, task_id: str) -> None:
                         summary=str(structured.get("summary") or "Completed"),
                         metadata=metadata,
                     )
-                    outcome = _finalize_terminal_record(
+                    try:
-                        kanban_db,
+                        outcome = _finalize_terminal_record(
-                        terminal_file,
+                            kanban_db,
-                        terminal_record,
+                            terminal_file,
-                    )
+                            terminal_record,
+                        )
+                    except Exception as error:
+                        raise TerminalFinalizationPending(
+                            "accepted worker result is durably journaled; "
+                            f"Kanban finalization raised {type(error).__name__}"
+                        ) from error
                     if outcome != "committed":
                         raise TerminalFinalizationPending(
                             "accepted worker result is durably journaled but "
@@ -1537,6 +1894,7 @@ def main() -> int:
                         print(f"worker future failed: {error}", file=sys.stderr, flush=True)
                     del futures[future]
             recover_pending_finalizations()
+            maybe_gc_lane_artifacts()
             active = set(futures.values())
             try:
                 newly_claimed = claim_ready(active, workers - len(futures))

testing/tests/test_hermes_cli_lanes.py

@@ -727,7 +727,7 @@ def test_accepted_result_survives_failed_finalization_and_replays_exact_run(
     lanes.execute_claim("cassandra", "t_terminal")
 
     candidates = list((state_root / "cassandra").glob("*.candidate-*.json"))
-    terminals = list((state_root / "cassandra").glob("*.terminal.json"))
+    terminals = list((state_root / "cassandra").glob("*.terminal.pending.json"))
     assert len(candidates) == 1
     assert json.loads(candidates[0].read_text())["structured"] == structured
     assert len(terminals) == 1
@@ -739,7 +739,12 @@ def test_accepted_result_survives_failed_finalization_and_replays_exact_run(
     assert lanes.recover_pending_finalizations() == 1
     assert task.status == "done"
     assert len(completions) == 2
-    terminal = json.loads(terminals[0].read_text())
+    assert not terminals[0].exists()
+    committed = list(
+        (state_root / "cassandra").glob("*.terminal.committed.json")
+    )
+    assert len(committed) == 1
+    terminal = json.loads(committed[0].read_text())
     assert terminal["kanban_state"] == "committed"
     assert completions[-1]["result"] == terminal["result"]
 
@@ -842,7 +847,334 @@ def test_terminal_replay_never_crosses_into_a_replacement_run(
 
     assert lanes.recover_pending_finalizations() == 0
     assert completions == []
-    assert json.loads(terminal_path.read_text())["kanban_state"] == "pending"
+    assert not terminal_path.exists()
+    quarantined = list((state_root / "cassandra/quarantine").glob("*.quarantine"))
+    assert len(quarantined) == 1
+
+
+def _completed_result(summary: str = "done") -> dict:
+    return {
+        "status": "completed",
+        "summary": summary,
+        "changed_files": [],
+        "tests_run": [],
+        "artifacts": [],
+        "findings": [],
+        "blockers": [],
+    }
+
+
+def test_complete_exception_after_journal_is_replayable_and_never_blocks(
+    tmp_path: Path,
+    monkeypatch,
+):
+    state_root = tmp_path / "cli-lanes"
+    monkeypatch.setattr(lanes, "STATE_ROOT", state_root)
+    task = SimpleNamespace(
+        id="t_crash",
+        status="running",
+        result=None,
+        current_run_id=51,
+        assignee="cli-auto",
+        max_runtime_seconds=60,
+    )
+    completion_raises = {"value": True}
+    blocks = []
+    comments = []
+
+    class Connection:
+        def close(self):
+            return None
+
+    def complete_task(_conn, _task_id, **kwargs):
+        if completion_raises["value"]:
+            raise RuntimeError("crash between journal and commit")
+        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_crash"),
+        set_branch_name=lambda *_args: None,
+        set_workspace_path=lambda *_args: None,
+        build_worker_context=lambda *_args: "Finish safely.",
+        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("crash-safe result"), False
+        ),
+    )
+
+    lanes.execute_claim("cassandra", "t_crash")
+
+    pending = list((state_root / "cassandra").glob("*.terminal.pending.json"))
+    assert len(pending) == 1
+    assert blocks == []
+    assert any("RuntimeError" in body for body in comments)
+
+    completion_raises["value"] = False
+    assert lanes.recover_pending_finalizations() == 1
+    assert task.status == "done"
+    assert not pending[0].exists()
+
+
+def test_terminal_payload_cannot_select_a_different_board(
+    tmp_path: Path,
+    monkeypatch,
+):
+    state_root = tmp_path / "cli-lanes"
+    monkeypatch.setattr(lanes, "STATE_ROOT", state_root)
+    path = lanes._terminal_path(lanes.state_path("alpha", "t_alpha"), 3)
+    lanes.atomic_json(
+        path,
+        {
+            "board": "beta",
+            "task_id": "t_beta",
+            "expected_run_id": 9,
+            "result": json.dumps(_completed_result()),
+            "summary": "forged",
+            "metadata": {},
+            "kanban_state": "pending",
+        },
+    )
+    alpha_task = SimpleNamespace(
+        id="t_alpha", status="running", current_run_id=3, assignee="cli-auto"
+    )
+    opened = []
+    reclaimed = []
+
+    class Connection:
+        def close(self):
+            return None
+
+    fake_db = SimpleNamespace(
+        scoped_current_board=lambda _board: nullcontext(),
+        connect=lambda board: (opened.append(board) or Connection()),
+        get_task=lambda _conn, task_id: alpha_task if task_id == "t_alpha" else None,
+        reclaim_task=lambda _conn, task_id, **_kwargs: (reclaimed.append(task_id) or True),
+    )
+    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
+
+    assert lanes.recover_pending_finalizations() == 0
+
+    assert opened == ["alpha"]
+    assert reclaimed == ["t_alpha"]
+    assert not path.exists()
+    quarantined = list((state_root / "alpha/quarantine").glob("*.quarantine"))
+    assert len(quarantined) == 1
+    assert quarantined[0].stat().st_mode & 0o777 == 0o600
+
+
+def test_terminal_finalize_rereads_journal_after_a_path_swap(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    state_file = lanes.state_path("alpha", "t_alpha")
+    path, safe_record = lanes._write_terminal_record(
+        state_file,
+        board="alpha",
+        task_id="t_alpha",
+        run_id=4,
+        structured=_completed_result("safe"),
+        summary="safe",
+        metadata={},
+    )
+    forged = dict(safe_record)
+    forged.update({"board": "beta", "task_id": "t_beta", "expected_run_id": 7})
+    lanes.atomic_json(path, forged)
+    opened = []
+    fake_db = SimpleNamespace(
+        scoped_current_board=lambda _board: nullcontext(),
+        connect=lambda board: (opened.append(board) or pytest.fail("must not open DB")),
+    )
+
+    assert lanes._finalize_terminal_record(fake_db, path, safe_record) == "foreign"
+    assert opened == []
+
+
+@pytest.mark.parametrize("payload", [b"", b'{"board":"cassandra"'])
+def test_malformed_exact_run_journal_is_quarantined_and_reclaimed(
+    tmp_path: Path,
+    monkeypatch,
+    payload: bytes,
+):
+    state_root = tmp_path / "cli-lanes"
+    monkeypatch.setattr(lanes, "STATE_ROOT", state_root)
+    path = lanes._terminal_path(lanes.state_path("cassandra", "t_partial"), 12)
+    path.parent.mkdir(parents=True)
+    path.write_bytes(payload)
+    task = SimpleNamespace(
+        id="t_partial", status="running", current_run_id=12, assignee="cli-auto"
+    )
+    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,
+        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 reclaimed == [True]
+    assert not path.exists()
+    assert lanes._has_pending_finalization("cassandra", "t_partial", 12) is False
+    quarantined = list((state_root / "cassandra/quarantine").glob("*.quarantine"))
+    assert len(quarantined) == 1
+    assert quarantined[0].stat().st_mode & 0o777 == 0o600
+
+
+def test_recovery_does_not_parse_committed_journals_every_tick(
+    tmp_path: Path,
+    monkeypatch,
+):
+    state_root = tmp_path / "cli-lanes"
+    monkeypatch.setattr(lanes, "STATE_ROOT", state_root)
+    committed = lanes._terminal_path(
+        lanes.state_path("cassandra", "t_done"), 1, "committed"
+    )
+    lanes.atomic_json(committed, {"kanban_state": "committed"})
+    reads = []
+    monkeypatch.setattr(lanes, "load_json", lambda path: (reads.append(path) or {}))
+    monkeypatch.setitem(
+        sys.modules,
+        "hermes_cli",
+        SimpleNamespace(kanban_db=SimpleNamespace()),
+    )
+
+    assert lanes.recover_pending_finalizations() == 0
+    assert reads == []
+
+
+def test_atomic_json_fsyncs_file_and_directory_and_uses_private_mode(
+    tmp_path: Path,
+    monkeypatch,
+):
+    calls = []
+    real_fsync = lanes.os.fsync
+    monkeypatch.setattr(
+        lanes.os,
+        "fsync",
+        lambda descriptor: (calls.append(descriptor), real_fsync(descriptor))[1],
+    )
+    path = tmp_path / "state.json"
+
+    lanes.atomic_json(path, {"safe": True})
+
+    assert json.loads(path.read_text()) == {"safe": True}
+    assert path.stat().st_mode & 0o777 == 0o600
+    assert len(calls) >= 2
+
+
+def test_atomic_json_refuses_a_precreated_temp_symlink(
+    tmp_path: Path,
+    monkeypatch,
+):
+    path = tmp_path / "state.json"
+    victim = tmp_path / "credential"
+    victim.write_text("do not overwrite", encoding="utf-8")
+    temporary = tmp_path / ".state.json.fixed.tmp"
+    temporary.symlink_to(victim)
+    monkeypatch.setattr(lanes.uuid, "uuid4", lambda: SimpleNamespace(hex="fixed"))
+
+    with pytest.raises(FileExistsError):
+        lanes.atomic_json(path, {"unsafe": True})
+
+    assert victim.read_text(encoding="utf-8") == "do not overwrite"
+    assert not path.exists()
+
+
+def test_artifact_gc_prunes_by_age_without_touching_pending_journals(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    board = lanes.STATE_ROOT / "cassandra"
+    board.mkdir(parents=True)
+    quarantine = board / "quarantine"
+    quarantine.mkdir()
+    old_artifacts = [
+        board / "t.run-1.candidate-1.json",
+        board / "t.run-1.provider-1.result.json",
+        board / "t.run-1.terminal.committed.json",
+        quarantine / "t.invalid.1234.quarantine",
+    ]
+    for artifact in old_artifacts:
+        artifact.write_text("{}", encoding="utf-8")
+    pending = board / "t.run-1.terminal.pending.json"
+    pending.write_text("{}", encoding="utf-8")
+    old_time = 100.0
+    for artifact in old_artifacts:
+        os.utime(artifact, (old_time, old_time))
+    os.utime(pending, (old_time, old_time))
+
+    assert lanes.gc_lane_artifacts(
+        now=1000.0, max_age_seconds=10, max_count=100, max_bytes=10000
+    ) == len(old_artifacts)
+    assert not any(artifact.exists() for artifact in old_artifacts)
+    assert pending.exists()
+
+
+def test_artifact_gc_prunes_oldest_by_count_and_total_bytes(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    board = lanes.STATE_ROOT / "cassandra"
+    board.mkdir(parents=True)
+    files = []
+    for sequence in range(3):
+        path = board / f"t.run-1.provider-{sequence}.result.json"
+        path.write_text("x" * 40, encoding="utf-8")
+        os.utime(path, (100 + sequence, 100 + sequence))
+        files.append(path)
+
+    assert lanes.gc_lane_artifacts(
+        now=200.0, max_age_seconds=1000, max_count=2, max_bytes=45
+    ) == 2
+    assert [path.exists() for path in files] == [False, False, True]
+
+
+def test_artifact_gc_is_interval_gated(
+    tmp_path: Path,
+    monkeypatch,
+):
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    monkeypatch.setattr(lanes, "LAST_ARTIFACT_GC", 0.0)
+    calls = []
+    monkeypatch.setattr(
+        lanes, "gc_lane_artifacts", lambda **kwargs: (calls.append(kwargs) or 0)
+    )
+
+    assert lanes.maybe_gc_lane_artifacts(now=1000.0) == 0
+    assert lanes.maybe_gc_lane_artifacts(now=1001.0) == 0
+    assert len(calls) == 1
 
 
 @pytest.mark.parametrize(
@@ -936,7 +1268,7 @@ def test_claim_requires_structured_evidence_and_surfaces_artifacts(
         block_task=lambda *_args, **kwargs: calls.append(("block", kwargs)),
     )
     monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
-    monkeypatch.setattr(lanes, "state_path", lambda _board, _task_id: tmp_path / "state.json")
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
     monkeypatch.setattr(
         lanes,
         "select_route",
@@ -1004,11 +1336,7 @@ def test_goal_card_continues_after_local_judge_rejects_progress(
         block_task=lambda *_args, **kwargs: calls.append(("block", kwargs)),
     )
     monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
-    monkeypatch.setattr(
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
-        lanes,
-        "state_path",
-        lambda _board, _task_id: tmp_path / "state.json",
-    )
     claude_low = lanes.Route(
         "claude", "claude-fable-5", "low", "claude-low", "jetson", "vote", 1, ()
     )
@@ -1087,7 +1415,9 @@ def test_goal_card_continues_after_local_judge_rejects_progress(
     assert route_calls[2][1]["exclude_provider"] == "claude"
     assert "prior rejected reports" in judge_contexts[1]
     assert "commit, push, and remote verification are missing" in judge_contexts[1]
-    candidates = sorted(tmp_path.glob("state.run-12.candidate-*.json"))
+    candidates = sorted(
+        (lanes.STATE_ROOT / "cassandra").glob("t_goal.run-12.candidate-*.json")
+    )
     assert len(candidates) == 2
     assert json.loads(candidates[0].read_text())["structured"]["summary"] == (
         "Focused tests passed."
@@ -1098,6 +1428,87 @@ def test_goal_card_continues_after_local_judge_rejects_progress(
     assert reports == []
 
 
+def test_capacity_fallback_preserves_first_claude_structured_response(
+    tmp_path: Path,
+    monkeypatch,
+):
+    task = SimpleNamespace(
+        id="t_fallback",
+        status="running",
+        result=None,
+        current_run_id=14,
+        assignee="cli-auto",
+        max_runtime_seconds=60,
+    )
+
+    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,
+        worker_log_path=lambda _task_id, board: tmp_path / "worker.log",
+        _resolve_worktree_workspace=lambda _task, board: (tmp_path, "wt/t_fallback"),
+        set_branch_name=lambda *_args: None,
+        set_workspace_path=lambda *_args: None,
+        build_worker_context=lambda *_args: "Complete with a fallback if needed.",
+        heartbeat_worker=lambda *_args, **_kwargs: True,
+        add_comment=lambda *_args: None,
+        complete_task=lambda _conn, _task_id, **kwargs: (
+            setattr(task, "status", "done") or setattr(task, "result", kwargs["result"]) or True
+        ),
+        block_task=lambda *_args, **_kwargs: pytest.fail("fallback should complete"),
+    )
+    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    claude = lanes.Route(
+        "claude", "claude-fable-5", "high", "claude-high", "test", "test", 1, ()
+    )
+    codex = lanes.Route(
+        "codex", "gpt-5.6-sol", "high", "codex-high", "test", "test", 1, ()
+    )
+    monkeypatch.setattr(
+        lanes,
+        "select_route",
+        lambda _prompt, assignee, **_kwargs: codex
+        if assignee == "cli-codex-high"
+        else claude,
+    )
+    reports = [
+        lanes.ProcessResult(
+            1,
+            "subscription capacity exhausted",
+            {
+                **_completed_result("Claude preserved evidence"),
+                "status": "incomplete",
+                "blockers": ["subscription capacity exhausted"],
+            },
+            True,
+        ),
+        lanes.ProcessResult(0, "", _completed_result("Codex completed"), False),
+    ]
+    monkeypatch.setattr(lanes, "run_provider", lambda *_args, **_kwargs: reports.pop(0))
+
+    lanes.execute_claim("cassandra", "t_fallback")
+
+    candidates = sorted(
+        (lanes.STATE_ROOT / "cassandra").glob("t_fallback.run-14.candidate-*.json")
+    )
+    assert len(candidates) == 2
+    first, second = [json.loads(path.read_text()) for path in candidates]
+    assert (first["provider"], first["structured"]["summary"]) == (
+        "claude",
+        "Claude preserved evidence",
+    )
+    assert (second["provider"], second["structured"]["summary"]) == (
+        "codex",
+        "Codex completed",
+    )
+    assert task.status == "done"
+
+
 def test_workspace_preparation_failure_durably_blocks_the_claim(tmp_path: Path, monkeypatch):
     task = SimpleNamespace(id="t_bad_worktree", current_run_id=7, assignee="cli-auto")
     calls = []
@@ -1143,11 +1554,15 @@ def test_artifacts_cannot_escape_the_task_worktree(tmp_path: Path):
 def test_restart_provider_change_includes_explicit_workspace_handoff(tmp_path: Path, monkeypatch):
     task = SimpleNamespace(
         id="t_resume",
+        status="running",
+        result=None,
         current_run_id=9,
         assignee="cli-auto",
         max_runtime_seconds=60,
     )
-    state_file = tmp_path / "state.json"
+    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
+    state_file = lanes.state_path("cassandra", "t_resume")
+    state_file.parent.mkdir(parents=True)
     state_file.write_text(
         json.dumps({"current_route": {"provider": "claude"}}),
         encoding="utf-8",
@@ -1173,7 +1588,6 @@ def test_restart_provider_change_includes_explicit_workspace_handoff(tmp_path: P
         block_task=lambda *_args, **_kwargs: None,
     )
     monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
-    monkeypatch.setattr(lanes, "state_path", lambda _board, _task_id: state_file)
     monkeypatch.setattr(
         lanes,
         "select_route",