hermes: preserve terminal results and prevent duplicate fanout

dockerfiles/Dockerfile.hermes-agent

@@ -1544,6 +1544,13 @@ function RootRedirect() {
 }
 NODE
 
+COPY dockerfiles/patch-hermes-execution-safety.py /tmp/patch-hermes-execution-safety.py
+COPY dockerfiles/hermes-execution-safety-regression.py /tmp/hermes-execution-safety-regression.py
+RUN /opt/hermes/.venv/bin/python /tmp/patch-hermes-execution-safety.py \
+    && /opt/hermes/.venv/bin/python /tmp/hermes-execution-safety-regression.py \
+    && rm /tmp/patch-hermes-execution-safety.py \
+        /tmp/hermes-execution-safety-regression.py
+
 COPY dockerfiles/hermes-session-migrate.py /opt/hermes/bin/hermes-session-migrate
 
 RUN cd /opt/hermes/web \
@@ -1602,6 +1609,8 @@ RUN cd /opt/hermes/web \
         /opt/hermes/agent/turn_context.py \
         /opt/hermes/agent/conversation_loop.py \
         /opt/hermes/hermes_cli/oneshot.py \
+        /opt/hermes/hermes_cli/kanban_decompose.py \
+        /opt/hermes/gateway/kanban_watchers.py \
         /opt/hermes/tools/delegate_tool.py \
         /opt/hermes/tools/web_tools.py \
         /opt/hermes/tools/python_sandbox_tool.py \

dockerfiles/Dockerfile.hermes-agent.dockerignore

@@ -5,3 +5,5 @@
 !dockerfiles/hermes-public-extract/**
 !dockerfiles/hermes-session-activity-panel.tsx
 !dockerfiles/hermes-session-migrate.py
+!dockerfiles/patch-hermes-execution-safety.py
+!dockerfiles/hermes-execution-safety-regression.py

dockerfiles/hermes-execution-safety-regression.py

@@ -0,0 +1,212 @@
+"""Build-time regressions for Hermes automatic decomposition safety."""
+
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import unittest
+from types import SimpleNamespace
+from unittest import mock
+
+
+_HERMES_HOME = tempfile.TemporaryDirectory(prefix="hermes-execution-test-home-")
+os.environ["HERMES_HOME"] = _HERMES_HOME.name
+
+from agent import auxiliary_client  # noqa: E402
+from hermes_cli import kanban_db  # noqa: E402
+from hermes_cli import kanban_decompose  # noqa: E402
+
+
+class _FakeCompletions:
+    def __init__(self, payload: dict, before_response=None) -> None:
+        self.payload = payload
+        self.before_response = before_response
+        self.calls = 0
+
+    def create(self, **_kwargs):
+        self.calls += 1
+        if self.before_response is not None:
+            self.before_response()
+        return SimpleNamespace(
+            choices=[
+                SimpleNamespace(
+                    message=SimpleNamespace(content=json.dumps(self.payload))
+                )
+            ]
+        )
+
+
+class _FakeClient:
+    def __init__(self, completions: _FakeCompletions) -> None:
+        self.chat = SimpleNamespace(completions=completions)
+
+
+class AutomaticDecompositionSafetyTests(unittest.TestCase):
+    """Exercise the patched upstream APIs against an isolated real database."""
+
+    def setUp(self) -> None:
+        self.connection = kanban_db.connect()
+        self.patches = [
+            mock.patch.object(
+                kanban_decompose,
+                "_build_roster",
+                return_value=([], {"default"}),
+            ),
+            mock.patch.object(
+                kanban_decompose,
+                "_load_config",
+                return_value={"kanban": {"auto_promote_children": True}},
+            ),
+            mock.patch.object(
+                auxiliary_client,
+                "get_auxiliary_extra_body",
+                return_value={},
+            ),
+        ]
+        for patch in self.patches:
+            patch.start()
+
+    def tearDown(self) -> None:
+        for patch in reversed(self.patches):
+            patch.stop()
+        self.connection.close()
+
+    def _client(self, payload: dict, before_response=None) -> _FakeCompletions:
+        completions = _FakeCompletions(payload, before_response)
+        client = _FakeClient(completions)
+        patch = mock.patch.object(
+            auxiliary_client,
+            "get_text_auxiliary_client",
+            return_value=(client, "test-decomposer"),
+        )
+        patch.start()
+        self.addCleanup(patch.stop)
+        return completions
+
+    def _status(self, task_id: str) -> str:
+        task = kanban_db.get_task(self.connection, task_id)
+        self.assertIsNotNone(task)
+        return task.status
+
+    def _route_executed_task_to_triage(self) -> str:
+        task_id = kanban_db.create_task(
+            self.connection,
+            title="already implemented parent",
+        )
+        for attempt in range(kanban_db.BLOCK_RECURRENCE_LIMIT):
+            if attempt:
+                self.assertTrue(kanban_db.unblock_task(self.connection, task_id))
+            self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+            self.assertTrue(
+                kanban_db.block_task(
+                    self.connection,
+                    task_id,
+                    reason="same unavailable capability",
+                    kind="capability",
+                )
+            )
+        self.assertEqual(self._status(task_id), "triage")
+        self.assertEqual(
+            len(kanban_db.list_runs(self.connection, task_id)),
+            kanban_db.BLOCK_RECURRENCE_LIMIT,
+        )
+        return task_id
+
+    def test_automatic_decomposer_skips_executed_triage_task(self) -> None:
+        task_id = self._route_executed_task_to_triage()
+        completions = self._client(
+            {"fanout": False, "title": "must not run", "body": "must not run"}
+        )
+        before = len(kanban_db.list_tasks(self.connection))
+
+        outcome = kanban_decompose.decompose_task(task_id, automatic=True)
+
+        self.assertFalse(outcome.ok)
+        self.assertIn("execution history", outcome.reason)
+        self.assertEqual(completions.calls, 0)
+        self.assertEqual(self._status(task_id), "triage")
+        self.assertEqual(len(kanban_db.list_tasks(self.connection)), before)
+
+    def test_manual_decomposition_remains_available_after_execution(self) -> None:
+        task_id = self._route_executed_task_to_triage()
+        completions = self._client(
+            {
+                "fanout": False,
+                "title": "operator-approved retry",
+                "body": "Retain one bounded task.",
+                "assignee": "default",
+            }
+        )
+
+        outcome = kanban_decompose.decompose_task(task_id, author="operator")
+
+        self.assertTrue(outcome.ok)
+        self.assertFalse(outcome.fanout)
+        self.assertEqual(completions.calls, 1)
+        self.assertEqual(self._status(task_id), "ready")
+
+    def test_fresh_triage_task_still_auto_promotes(self) -> None:
+        task_id = kanban_db.create_task(
+            self.connection,
+            title="fresh objective",
+            triage=True,
+        )
+        completions = self._client(
+            {
+                "fanout": False,
+                "title": "specified objective",
+                "body": "One bounded task.",
+                "assignee": "default",
+            }
+        )
+
+        outcome = kanban_decompose.decompose_task(task_id, automatic=True)
+
+        self.assertTrue(outcome.ok)
+        self.assertEqual(completions.calls, 1)
+        self.assertEqual(self._status(task_id), "ready")
+
+    def test_execution_history_gained_during_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,
+                "tasks": [
+                    {
+                        "title": "redundant child",
+                        "body": "must not be created",
+                        "assignee": "default",
+                        "parents": [],
+                    }
+                ],
+            },
+            before_response=add_run,
+        )
+        before = len(kanban_db.list_tasks(self.connection))
+
+        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")
+        self.assertEqual(len(kanban_db.list_tasks(self.connection)), before)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

dockerfiles/patch-hermes-execution-safety.py

@@ -0,0 +1,150 @@
+"""Patch upstream Hermes auto-decomposition to respect execution history."""
+
+import os
+from pathlib import Path
+
+
+def replace_once(source: str, before: str, after: str, label: str) -> str:
+    """Replace one anchored upstream fragment and fail closed on source drift."""
+    count = source.count(before)
+    if count != 1:
+        raise SystemExit(
+            f"Hermes {label} patch context changed: expected 1, found {count}"
+        )
+    return source.replace(before, after, 1)
+
+
+source_root = Path(os.environ.get("HERMES_SOURCE_ROOT", "/opt/hermes"))
+decompose_path = source_root / "hermes_cli/kanban_decompose.py"
+decompose = decompose_path.read_text(encoding="utf-8")
+
+helper_anchor = '''    if chosen not in valid_names:
+        return default_assignee
+    return chosen
+
+
+'''
+helper_replacement = helper_anchor + '''def _has_execution_history(task_id: str) -> bool:
+    """Return whether a task has ever entered a worker run."""
+    with kb.connect_closing() as conn:
+        return bool(kb.list_runs(conn, task_id))
+
+
+'''
+decompose = replace_once(
+    decompose,
+    helper_anchor,
+    helper_replacement,
+    "auto-decompose execution-history helper",
+)
+
+signature_before = '''def decompose_task(
+    task_id: str,
+    *,
+    author: Optional[str] = None,
+    timeout: Optional[int] = None,
+) -> DecomposeOutcome:
+'''
+signature_after = '''def decompose_task(
+    task_id: str,
+    *,
+    author: Optional[str] = None,
+    timeout: Optional[int] = None,
+    automatic: bool = False,
+) -> DecomposeOutcome:
+'''
+decompose = replace_once(
+    decompose,
+    signature_before,
+    signature_after,
+    "auto-decompose function signature",
+)
+
+status_before = '''    if task.status != "triage":
+        return DecomposeOutcome(
+            task_id, False, f"task is not in triage (status={task.status!r})"
+        )
+
+    cfg = _load_config()
+'''
+status_after = '''    if task.status != "triage":
+        return DecomposeOutcome(
+            task_id, False, f"task is not in triage (status={task.status!r})"
+        )
+    if automatic and _has_execution_history(task_id):
+        return DecomposeOutcome(
+            task_id,
+            False,
+            "task has execution history and requires deliberate manual triage",
+        )
+
+    cfg = _load_config()
+'''
+decompose = replace_once(
+    decompose,
+    status_before,
+    status_after,
+    "auto-decompose preflight guard",
+)
+
+single_before = '''        with kb.connect_closing() as conn:
+            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:
+            ok = kb.specify_triage_task(
+'''
+decompose = replace_once(
+    decompose,
+    single_before,
+    single_after,
+    "auto-decompose single-task commit guard",
+)
+
+fanout_before = '''    try:
+        with kb.connect_closing() as conn:
+            child_ids = kb.decompose_triage_task(
+'''
+fanout_after = '''    if automatic and _has_execution_history(task_id):
+        return DecomposeOutcome(
+            task_id,
+            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",
+)
+decompose_path.write_text(decompose, encoding="utf-8")
+
+
+watcher_path = source_root / "gateway/kanban_watchers.py"
+watcher = watcher_path.read_text(encoding="utf-8")
+call_before = '''                            outcome = _decomp.decompose_task(
+                                tid, author="auto-decomposer",
+                            )
+'''
+call_after = '''                            outcome = _decomp.decompose_task(
+                                tid,
+                                author="auto-decomposer",
+                                automatic=True,
+                            )
+'''
+watcher = replace_once(
+    watcher,
+    call_before,
+    call_after,
+    "gateway automatic-decomposition call",
+)
+watcher_path.write_text(watcher, encoding="utf-8")

services/hermes/scripts/cli_lane_runner.py

@@ -123,6 +123,10 @@ class ProcessResult:
     capacity_failure: bool
 
 
+class TerminalFinalizationPending(RuntimeError):
+    """An accepted worker result is durable but not committed to Kanban yet."""
+
+
 def utc_now() -> str:
     return datetime.now(timezone.utc).isoformat()
 
@@ -293,6 +297,115 @@ def state_path(board: str, task_id: str) -> Path:
     return STATE_ROOT / safe_board / f"{safe_task}.json"
 
 
+def _result_path(state_file: Path, run_id: Any, sequence: int) -> Path:
+    """Return a unique provider-result path; completed turns are never reused."""
+    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}.provider-{sequence}.result.json"
+    )
+
+
+def _candidate_path(state_file: Path, run_id: Any, sequence: int) -> Path:
+    """Return the durable path for one exact structured worker response."""
+    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}.candidate-{sequence}.json"
+    )
+
+
+def _terminal_path(state_file: Path, run_id: Any) -> Path:
+    """Return the replay journal path for an accepted terminal response."""
+    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")
+
+
+def _persist_candidate(
+    state: dict[str, Any],
+    state_file: Path,
+    structured: dict[str, Any],
+    *,
+    route: Route,
+    returncode: int,
+    goal_turn: int,
+) -> Path:
+    """Persist every structured response before any judge can supersede it."""
+    sequence = max(0, int(state.get("candidate_sequence", 0) or 0)) + 1
+    path = _candidate_path(state_file, state.get("run_id"), sequence)
+    while path.exists():
+        sequence += 1
+        path = _candidate_path(state_file, state.get("run_id"), sequence)
+    atomic_json(
+        path,
+        {
+            "board": state.get("board"),
+            "task_id": state.get("task_id"),
+            "expected_run_id": state.get("run_id"),
+            "goal_turn": goal_turn,
+            "provider": route.provider,
+            "model": route.model,
+            "effort": route.effort,
+            "returncode": returncode,
+            "structured": structured,
+            "recorded_at": utc_now(),
+        },
+    )
+    state["candidate_sequence"] = sequence
+    state["last_candidate_file"] = str(path)
+    atomic_json(state_file, state)
+    return path
+
+
+def _write_terminal_record(
+    state_file: Path,
+    *,
+    board: str,
+    task_id: str,
+    run_id: Any,
+    structured: dict[str, Any],
+    summary: str,
+    metadata: dict[str, Any],
+) -> tuple[Path, dict[str, Any]]:
+    """Journal an accepted result before attempting the Kanban transaction."""
+    path = _terminal_path(state_file, run_id)
+    record = {
+        "board": board,
+        "task_id": task_id,
+        "expected_run_id": run_id,
+        "result": json.dumps(structured, sort_keys=True),
+        "summary": summary,
+        "metadata": metadata,
+        "kanban_state": "pending",
+        "recorded_at": utc_now(),
+    }
+    atomic_json(path, record)
+    return path, record
+
+
+def _terminal_record_valid(record: dict[str, Any]) -> bool:
+    """Reject malformed or non-terminal replay journals without side effects."""
+    if not isinstance(record, dict):
+        return False
+    required_strings = ("board", "task_id", "result", "summary")
+    if not all(
+        isinstance(record.get(key), str) and record[key]
+        for key in required_strings
+    ):
+        return False
+    if not isinstance(record.get("expected_run_id"), int):
+        return False
+    if not isinstance(record.get("metadata"), dict):
+        return False
+    try:
+        structured = json.loads(record["result"])
+    except (TypeError, json.JSONDecodeError):
+        return False
+    return (
+        isinstance(structured, dict)
+        and structured.get("status") == "completed"
+        and not structured.get("blockers")
+    )
+
+
 def build_prompt(context: str, workspace: Path, handoff: str = "") -> str:
     """Create a bounded worker contract with an explicit machine-readable result."""
     return f"""You are a durable coding worker managed by Hermes Kanban.
@@ -667,11 +780,22 @@ def run_provider(
         state["claude_session_id"] = str(uuid.uuid4())
     state["current_route"] = asdict(route)
     state["updated_at"] = utc_now()
-    atomic_json(state_file, state)
     env = _base_env()
-    result_file = state_file.with_suffix(".result.json")
     if route.provider == "codex":
-        result_file.unlink(missing_ok=True)
+        result_sequence = max(0, int(state.get("result_sequence", 0) or 0)) + 1
+        result_file = _result_path(state_file, state.get("run_id"), result_sequence)
+        while result_file.exists():
+            result_sequence += 1
+            result_file = _result_path(
+                state_file,
+                state.get("run_id"),
+                result_sequence,
+            )
+        result_file.parent.mkdir(parents=True, exist_ok=True)
+        result_file.touch(mode=0o600, exist_ok=False)
+        state["result_sequence"] = result_sequence
+        state["current_result_file"] = str(result_file)
+        atomic_json(state_file, state)
         command = _codex_command(route, prompt, workspace, state, result_file)
         result = stream_process(
             command,
@@ -690,8 +814,23 @@ def run_provider(
             and NO_CODEX_THREAD in result.output.lower()
         ):
             state.pop("codex_thread_id", None)
+            result_sequence += 1
+            result_file = _result_path(
+                state_file,
+                state.get("run_id"),
+                result_sequence,
+            )
+            while result_file.exists():
+                result_sequence += 1
+                result_file = _result_path(
+                    state_file,
+                    state.get("run_id"),
+                    result_sequence,
+                )
+            result_file.touch(mode=0o600, exist_ok=False)
+            state["result_sequence"] = result_sequence
+            state["current_result_file"] = str(result_file)
             atomic_json(state_file, state)
-            result_file.unlink(missing_ok=True)
             result = stream_process(
                 _codex_command(route, prompt, workspace, state, result_file),
                 provider="codex",
@@ -706,8 +845,13 @@ def run_provider(
         file_result = load_json(result_file)
         if file_result.get("status") in cli_lane_goal.RESULT_STATUSES:
             result.structured = file_result
+            try:
+                result_file.chmod(0o600)
+            except OSError:
+                pass
         return result
 
+    atomic_json(state_file, state)
     resume = bool(state.get("claude_started"))
     result = stream_process(
         _claude_command(route, prompt, state, resume),
@@ -801,6 +945,95 @@ def _board_call(
     raise last_storage_error
 
 
+def _finalize_terminal_record(
+    kanban_db: Any,
+    path: Path,
+    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))
+    if not _terminal_record_valid(document):
+        return "invalid"
+    board = str(document["board"])
+    task_id = str(document["task_id"])
+    expected_run_id = int(document["expected_run_id"])
+
+    def operation(conn: Any) -> str:
+        task = kanban_db.get_task(conn, task_id)
+        if task is None:
+            return "stale"
+        status = str(_task_value(task, "status", ""))
+        if status == "done":
+            return (
+                "committed"
+                if str(_task_value(task, "result", "") or "") == document["result"]
+                else "stale"
+            )
+        if (
+            status != "running"
+            or _task_value(task, "current_run_id", None) != expected_run_id
+        ):
+            return "stale"
+        completed = kanban_db.complete_task(
+            conn,
+            task_id,
+            result=document["result"],
+            summary=document["summary"],
+            metadata=document["metadata"],
+            expected_run_id=expected_run_id,
+        )
+        return "committed" if completed else "pending"
+
+    outcome = str(_board_call(kanban_db, board, operation))
+    if outcome == "committed":
+        document["kanban_state"] = "committed"
+        document["committed_at"] = utc_now()
+        atomic_json(path, document)
+    return outcome
+
+
+def recover_pending_finalizations() -> int:
+    """Replay accepted exact-run results before scheduling more provider work."""
+    from hermes_cli import kanban_db
+
+    recovered = 0
+    for path in sorted(STATE_ROOT.glob("*/*.terminal.json")):
+        record = load_json(path)
+        if record.get("kanban_state") == "committed":
+            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(board, error)
+            continue
+        if outcome == "committed":
+            recovered += 1
+    return recovered
+
+
+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)
+    try:
+        path.stat()
+    except FileNotFoundError:
+        return False
+    except OSError:
+        return True
+    record = load_json(path)
+    if not _terminal_record_valid(record):
+        return True
+    identity = (
+        record.get("board") == board
+        and record.get("task_id") == task_id
+        and record.get("expected_run_id") == run_id
+    )
+    return not identity or record.get("kanban_state") != "committed"
+
+
 def execute_claim(board: str, task_id: str) -> None:
     """Execute one already-claimed task and commit its outcome to Kanban."""
     from hermes_cli import kanban_db
@@ -980,6 +1213,16 @@ 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,
@@ -992,6 +1235,8 @@ def execute_claim(board: str, task_id: str) -> None:
                     "goal_mode": goal_mode,
                     "goal_turn": goal_turn,
                 }
+                if candidate_file is not None:
+                    metadata["candidate_file"] = str(candidate_file)
                 if structured:
                     for key in (
                         "changed_files",
@@ -1036,18 +1281,27 @@ def execute_claim(board: str, task_id: str) -> None:
                     and result.returncode == 0
                     and completion_problem is None
                 ):
-                    _board_call(
+                    terminal_file = _terminal_path(state_file, run_id)
-                        kanban_db,
+                    metadata["terminal_record"] = str(terminal_file)
-                        board,
+                    terminal_file, terminal_record = _write_terminal_record(
-                        lambda fresh: kanban_db.complete_task(
+                        state_file,
-                            fresh,
+                        board=board,
-                            task_id,
+                        task_id=task_id,
-                            result=json.dumps(structured, sort_keys=True),
+                        run_id=run_id,
-                            summary=str(structured.get("summary") or "Completed"),
+                        structured=structured,
-                            metadata=metadata,
+                        summary=str(structured.get("summary") or "Completed"),
-                            expected_run_id=run_id,
+                        metadata=metadata,
-                        ),
                     )
+                    outcome = _finalize_terminal_record(
+                        kanban_db,
+                        terminal_file,
+                        terminal_record,
+                    )
+                    if outcome != "committed":
+                        raise TerminalFinalizationPending(
+                            "accepted worker result is durably journaled but "
+                            f"Kanban finalization is {outcome}"
+                        )
                     break
 
                 can_continue = (
@@ -1126,6 +1380,8 @@ def execute_claim(board: str, task_id: str) -> None:
                     ),
                 )
                 break
+        except TerminalFinalizationPending as error:
+            comment(str(error))
         except Exception as error:
             failure_reason = f"Direct CLI lane failed: {type(error).__name__}: {error}"
             _board_call(
@@ -1178,6 +1434,7 @@ def recover_orphans() -> None:
     """Return external running tasks to ready after a runner/pod restart."""
     from hermes_cli import kanban_db
 
+    recover_pending_finalizations()
     try:
         boards = kanban_db.list_boards(include_archived=False)
     except Exception as error:
@@ -1194,9 +1451,13 @@ def recover_orphans() -> None:
             try:
                 for task in kanban_db.list_tasks(conn):
                     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 _has_pending_finalization(board, task_id, run_id):
+                            continue
                         kanban_db.reclaim_task(
                             conn,
-                            str(_task_value(task, "id")),
+                            task_id,
                             reason="direct CLI lane restarted; provider session will resume",
                         )
                 BOARD_CORRUPTION_ERRORS.pop(board, None)
@@ -1275,6 +1536,7 @@ def main() -> int:
                     except Exception as error:
                         print(f"worker future failed: {error}", file=sys.stderr, flush=True)
                     del futures[future]
+            recover_pending_finalizations()
             active = set(futures.values())
             try:
                 newly_claimed = claim_ready(active, workers - len(futures))

testing/tests/test_hermes_cli_lanes.py

@@ -359,6 +359,66 @@ def test_codex_missing_server_thread_restarts_fresh(tmp_path: Path, monkeypatch)
     assert "codex_thread_id" not in json.loads(state_file.read_text())
 
 
+def test_codex_result_files_are_unique_and_prior_turns_are_retained(
+    tmp_path: Path,
+    monkeypatch,
+):
+    state = {"run_id": 17}
+    state_file = tmp_path / "task.json"
+    result_paths = []
+
+    def fake_stream(command, **_kwargs):
+        result_path = Path(command[command.index("-o") + 1])
+        result_path.write_text(
+            json.dumps(
+                {
+                    "status": "completed",
+                    "summary": f"turn {len(result_paths) + 1}",
+                    "changed_files": [],
+                    "tests_run": [],
+                    "artifacts": [],
+                    "findings": [],
+                    "blockers": [],
+                }
+            ),
+            encoding="utf-8",
+        )
+        result_paths.append(result_path)
+        return lanes.ProcessResult(0, "", None, False)
+
+    monkeypatch.setattr(lanes, "stream_process", fake_stream)
+    route = lanes.Route(
+        "codex", "gpt-5.6-sol", "high", "codex-high", "jetson", "vote", 1, ()
+    )
+
+    first = lanes.run_provider(
+        route,
+        "First turn.",
+        tmp_path,
+        state,
+        state_file,
+        tmp_path / "worker.log",
+        lambda _note: True,
+        60,
+    )
+    second = lanes.run_provider(
+        route,
+        "Second turn.",
+        tmp_path,
+        state,
+        state_file,
+        tmp_path / "worker.log",
+        lambda _note: True,
+        60,
+    )
+
+    assert first.structured["summary"] == "turn 1"
+    assert second.structured["summary"] == "turn 2"
+    assert len(set(result_paths)) == 2
+    assert all(path.exists() for path in result_paths)
+    assert all(path.stat().st_mode & 0o777 == 0o600 for path in result_paths)
+
+
 def test_successful_process_text_cannot_masquerade_as_capacity_failure(tmp_path: Path):
     result = lanes.stream_process(
         [sys.executable, "-c", "print('authentication work completed')"],
@@ -594,6 +654,197 @@ def test_board_call_retries_storage_faults_on_fresh_connections():
     assert all(connection.closed for connection in connections)
 
 
+def test_accepted_result_survives_failed_finalization_and_replays_exact_run(
+    tmp_path: Path,
+    monkeypatch,
+):
+    state_root = tmp_path / "cli-lanes"
+    monkeypatch.setattr(lanes, "STATE_ROOT", state_root)
+    task = SimpleNamespace(
+        id="t_terminal",
+        status="running",
+        result=None,
+        current_run_id=41,
+        assignee="cli-auto",
+        max_runtime_seconds=60,
+    )
+    completions = []
+    comments = []
+    permit_completion = {"value": False}
+
+    class Connection:
+        def close(self):
+            return None
+
+    def complete_task(_conn, _task_id, **kwargs):
+        completions.append(kwargs)
+        if not permit_completion["value"]:
+            return False
+        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_terminal"),
+        set_branch_name=lambda *_args: None,
+        set_workspace_path=lambda *_args: None,
+        build_worker_context=lambda *_args: "Finish and verify the objective.",
+        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: pytest.fail(
+            "an accepted journaled result must not be converted into a block"
+        ),
+    )
+    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", "jetson", "vote", 1, ()
+        ),
+    )
+    structured = {
+        "status": "completed",
+        "summary": "Verified exact terminal result.",
+        "changed_files": [],
+        "tests_run": ["pytest: passed"],
+        "artifacts": [],
+        "findings": [],
+        "blockers": [],
+    }
+    monkeypatch.setattr(
+        lanes,
+        "run_provider",
+        lambda *_args, **_kwargs: lanes.ProcessResult(0, "", structured, False),
+    )
+
+    lanes.execute_claim("cassandra", "t_terminal")
+
+    candidates = list((state_root / "cassandra").glob("*.candidate-*.json"))
+    terminals = list((state_root / "cassandra").glob("*.terminal.json"))
+    assert len(candidates) == 1
+    assert json.loads(candidates[0].read_text())["structured"] == structured
+    assert len(terminals) == 1
+    assert json.loads(terminals[0].read_text())["kanban_state"] == "pending"
+    assert task.status == "running"
+    assert any("durably journaled" in body for body in comments)
+
+    permit_completion["value"] = True
+    assert lanes.recover_pending_finalizations() == 1
+    assert task.status == "done"
+    assert len(completions) == 2
+    terminal = json.loads(terminals[0].read_text())
+    assert terminal["kanban_state"] == "committed"
+    assert completions[-1]["result"] == terminal["result"]
+
+
+def test_restart_does_not_reclaim_an_exact_run_awaiting_finalization(
+    tmp_path: Path,
+    monkeypatch,
+):
+    state_root = tmp_path / "cli-lanes"
+    monkeypatch.setattr(lanes, "STATE_ROOT", state_root)
+    state_file = lanes.state_path("cassandra", "t_terminal")
+    terminal_path, _record = lanes._write_terminal_record(
+        state_file,
+        board="cassandra",
+        task_id="t_terminal",
+        run_id=8,
+        structured={
+            "status": "completed",
+            "summary": "done",
+            "changed_files": [],
+            "tests_run": [],
+            "artifacts": [],
+            "findings": [],
+            "blockers": [],
+        },
+        summary="done",
+        metadata={},
+    )
+    assert terminal_path.exists()
+    task = SimpleNamespace(
+        id="t_terminal",
+        status="running",
+        assignee="cli-auto",
+        current_run_id=8,
+    )
+    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: [task],
+        reclaim_task=lambda *_args, **_kwargs: reclaimed.append(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 == []
+
+
+def test_terminal_replay_never_crosses_into_a_replacement_run(
+    tmp_path: Path,
+    monkeypatch,
+):
+    state_root = tmp_path / "cli-lanes"
+    monkeypatch.setattr(lanes, "STATE_ROOT", state_root)
+    state_file = lanes.state_path("cassandra", "t_terminal")
+    terminal_path, _record = lanes._write_terminal_record(
+        state_file,
+        board="cassandra",
+        task_id="t_terminal",
+        run_id=8,
+        structured={
+            "status": "completed",
+            "summary": "old run",
+            "changed_files": [],
+            "tests_run": [],
+            "artifacts": [],
+            "findings": [],
+            "blockers": [],
+        },
+        summary="old run",
+        metadata={},
+    )
+    replacement = SimpleNamespace(
+        id="t_terminal",
+        status="running",
+        result=None,
+        current_run_id=9,
+    )
+    completions = []
+
+    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: replacement,
+        complete_task=lambda *_args, **_kwargs: completions.append(True),
+    )
+    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
+
+    assert lanes.recover_pending_finalizations() == 0
+    assert completions == []
+    assert json.loads(terminal_path.read_text())["kanban_state"] == "pending"
+
+
 @pytest.mark.parametrize(
     ("result", "expected_action"),
     [
@@ -640,6 +891,8 @@ def test_claim_requires_structured_evidence_and_surfaces_artifacts(
 ):
     task = SimpleNamespace(
         id="t_worker",
+        status="running",
+        result=None,
         current_run_id=4,
         assignee="cli-auto",
         max_runtime_seconds=60,
@@ -677,7 +930,9 @@ def test_claim_requires_structured_evidence_and_surfaces_artifacts(
             heartbeats.append((note, expected_run_id)) or True
         ),
         add_comment=lambda *_args: None,
-        complete_task=lambda *_args, **kwargs: calls.append(("complete", kwargs)),
+        complete_task=lambda *_args, **kwargs: (
+            calls.append(("complete", kwargs)) or True
+        ),
         block_task=lambda *_args, **kwargs: calls.append(("block", kwargs)),
     )
     monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
@@ -717,6 +972,8 @@ def test_goal_card_continues_after_local_judge_rejects_progress(
 ):
     task = SimpleNamespace(
         id="t_goal",
+        status="running",
+        result=None,
         current_run_id=12,
         assignee="cli-auto",
         max_runtime_seconds=300,
@@ -741,7 +998,9 @@ def test_goal_card_continues_after_local_judge_rejects_progress(
         build_worker_context=lambda *_args: "Run tests, commit, push, and verify remote HEAD.",
         heartbeat_worker=lambda *_args, **_kwargs: True,
         add_comment=lambda _conn, _task_id, _author, body: comments.append(body),
-        complete_task=lambda *_args, **kwargs: calls.append(("complete", kwargs)),
+        complete_task=lambda *_args, **kwargs: (
+            calls.append(("complete", kwargs)) or True
+        ),
         block_task=lambda *_args, **kwargs: calls.append(("block", kwargs)),
     )
     monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
@@ -828,6 +1087,14 @@ 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"))
+    assert len(candidates) == 2
+    assert json.loads(candidates[0].read_text())["structured"]["summary"] == (
+        "Focused tests passed."
+    )
+    assert json.loads(candidates[1].read_text())["structured"]["summary"].startswith(
+        "Full tests passed"
+    )
     assert reports == []
 
 
@@ -902,7 +1169,7 @@ def test_restart_provider_change_includes_explicit_workspace_handoff(tmp_path: P
         set_workspace_path=lambda *_args: None,
         build_worker_context=lambda *_args: "resume objective",
         add_comment=lambda *_args: None,
-        complete_task=lambda *_args, **_kwargs: None,
+        complete_task=lambda *_args, **_kwargs: True,
         block_task=lambda *_args, **_kwargs: None,
     )
     monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
@@ -1256,6 +1523,27 @@ def test_agent_dashboard_reconnects_all_transient_websockets():
     assert "dashboard token rotated by a server restart" in dockerfile
 
 
+def test_agent_image_runs_execution_safety_patch_and_regressions():
+    dockerfiles = HERMES.parents[1] / "dockerfiles"
+    dockerfile = (dockerfiles / "Dockerfile.hermes-agent").read_text(encoding="utf-8")
+    dockerignore = (dockerfiles / "Dockerfile.hermes-agent.dockerignore").read_text(
+        encoding="utf-8"
+    )
+
+    for name in (
+        "patch-hermes-execution-safety.py",
+        "hermes-execution-safety-regression.py",
+    ):
+        assert f"COPY dockerfiles/{name}" in dockerfile
+        assert f"!dockerfiles/{name}" in dockerignore
+    assert "/opt/hermes/.venv/bin/python /tmp/patch-hermes-execution-safety.py" in (
+        dockerfile
+    )
+    assert "/opt/hermes/.venv/bin/python /tmp/hermes-execution-safety-regression.py" in (
+        dockerfile
+    )
+
+
 def test_agent_refreshes_routes_after_restoring_cli_logins():
     deployment = _agent_deployment()
     init_containers = {