hermes: bind terminal commit to journal inode

dockerfiles/Dockerfile.hermes-agent

@@ -1431,6 +1431,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 \
@@ -1489,6 +1496,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

@@ -4,3 +4,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,491 @@
+"""Build-time regressions for Hermes automatic decomposition safety."""
+
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import threading
+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_latest_ended_run_can_replay_a_durable_completion(self) -> None:
+        task_id = kanban_db.create_task(self.connection, title="journaled result")
+        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+        run_id = kanban_db.get_task(self.connection, task_id).current_run_id
+        self.assertTrue(
+            kanban_db.block_task(
+                self.connection,
+                task_id,
+                reason="legacy post-journal failure",
+                kind="capability",
+                expected_run_id=run_id,
+            )
+        )
+
+        with mock.patch.object(
+            kanban_db, "_fire_kanban_lifecycle_hook"
+        ) as lifecycle:
+            self.assertTrue(
+                kanban_db.complete_task(
+                    self.connection,
+                    task_id,
+                    result="durable terminal result",
+                    summary="durable terminal result",
+                    replay_ended_run_id=run_id,
+                )
+            )
+        lifecycle.assert_called_once()
+        self.assertEqual(lifecycle.call_args.args[:2], ("kanban_task_completed", task_id))
+        self.assertEqual(lifecycle.call_args.kwargs["run_id"], run_id)
+        task = kanban_db.get_task(self.connection, task_id)
+        self.assertEqual(task.status, "done")
+        self.assertEqual(task.completed_run_id, run_id)
+        runs = kanban_db.list_runs(self.connection, task_id)
+        self.assertEqual(len(runs), 1)
+        self.assertEqual(runs[0].id, run_id)
+        self.assertEqual(runs[0].status, "done")
+        self.assertEqual(runs[0].outcome, "completed")
+        self.assertEqual(runs[0].summary, "durable terminal result")
+        completed_events = self.connection.execute(
+            "SELECT run_id FROM task_events "
+            "WHERE task_id = ? AND kind = 'completed' ORDER BY id",
+            (task_id,),
+        ).fetchall()
+        self.assertEqual([row["run_id"] for row in completed_events], [run_id])
+
+    def test_older_ended_run_cannot_complete_over_a_replacement(self) -> None:
+        task_id = kanban_db.create_task(self.connection, title="replacement guard")
+        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+        old_run = kanban_db.get_task(self.connection, task_id).current_run_id
+        self.assertTrue(
+            kanban_db.block_task(
+                self.connection,
+                task_id,
+                reason="first run",
+                kind="capability",
+                expected_run_id=old_run,
+            )
+        )
+        self.assertTrue(kanban_db.unblock_task(self.connection, task_id))
+        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+        replacement_run = kanban_db.get_task(self.connection, task_id).current_run_id
+        self.assertNotEqual(old_run, replacement_run)
+        self.assertTrue(
+            kanban_db.block_task(
+                self.connection,
+                task_id,
+                reason="replacement run",
+                kind="capability",
+                expected_run_id=replacement_run,
+            )
+        )
+
+        self.assertFalse(
+            kanban_db.complete_task(
+                self.connection,
+                task_id,
+                result="stale result",
+                replay_ended_run_id=old_run,
+            )
+        )
+        self.assertIn(self._status(task_id), {"blocked", "triage"})
+
+    def test_equal_result_bytes_remain_bound_to_the_completing_run(self) -> None:
+        task_id = kanban_db.create_task(self.connection, title="same result retries")
+        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+        first_run = kanban_db.get_task(self.connection, task_id).current_run_id
+        self.assertTrue(
+            kanban_db.block_task(
+                self.connection,
+                task_id,
+                reason="first attempt ended",
+                expected_run_id=first_run,
+            )
+        )
+        self.assertTrue(kanban_db.unblock_task(self.connection, task_id))
+        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+        second_run = kanban_db.get_task(self.connection, task_id).current_run_id
+        self.assertTrue(
+            kanban_db.complete_task(
+                self.connection,
+                task_id,
+                result="identical result bytes",
+                summary="second run wins",
+                expected_run_id=second_run,
+            )
+        )
+
+        task = kanban_db.get_task(self.connection, task_id)
+        self.assertEqual(task.completed_run_id, second_run)
+        self.assertNotEqual(task.completed_run_id, first_run)
+        runs = {run.id: run for run in kanban_db.list_runs(self.connection, task_id)}
+        self.assertEqual(runs[first_run].outcome, "blocked")
+        self.assertEqual(runs[second_run].outcome, "completed")
+
+    def test_exact_run_reclaim_succeeds_for_the_authoritative_run(self) -> None:
+        task_id = kanban_db.create_task(self.connection, title="recover exact run")
+        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+        run_id = kanban_db.get_task(self.connection, task_id).current_run_id
+
+        self.assertTrue(
+            kanban_db.reclaim_task(
+                self.connection,
+                task_id,
+                reason="invalid exact journal",
+                expected_run_id=run_id,
+            )
+        )
+        task = kanban_db.get_task(self.connection, task_id)
+        self.assertEqual(task.status, "ready")
+        self.assertIsNone(task.current_run_id)
+
+    def test_stale_reclaim_before_transaction_preserves_replacement_run(self) -> None:
+        task_id = kanban_db.create_task(self.connection, title="replacement before txn")
+        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+        old_run = kanban_db.get_task(self.connection, task_id).current_run_id
+        self.assertTrue(
+            kanban_db.block_task(
+                self.connection,
+                task_id,
+                reason="old recovery run",
+                expected_run_id=old_run,
+            )
+        )
+        self.assertTrue(kanban_db.unblock_task(self.connection, task_id))
+        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+        replacement = kanban_db.get_task(self.connection, task_id)
+        replacement_run = replacement.current_run_id
+        self.assertNotEqual(old_run, replacement_run)
+        signals = []
+
+        self.assertFalse(
+            kanban_db.reclaim_task(
+                self.connection,
+                task_id,
+                reason="stale journal",
+                expected_run_id=old_run,
+                signal_fn=lambda *args: signals.append(args),
+            )
+        )
+        latest = kanban_db.get_task(self.connection, task_id)
+        self.assertEqual(latest.status, "running")
+        self.assertEqual(latest.current_run_id, replacement_run)
+        self.assertEqual(latest.claim_lock, replacement.claim_lock)
+        self.assertEqual(signals, [])
+
+    def test_reclaim_update_guard_preserves_run_changed_inside_transaction(self) -> None:
+        task_id = kanban_db.create_task(self.connection, title="replacement in txn")
+        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+        old_run = kanban_db.get_task(self.connection, task_id).current_run_id
+        replacement = {}
+
+        def install_replacement(*_args, **_kwargs):
+            cursor = self.connection.execute(
+                "INSERT INTO task_runs (task_id, status, claim_lock, started_at) "
+                "VALUES (?, 'running', ?, strftime('%s','now'))",
+                (task_id, "replacement-lock"),
+            )
+            replacement["run_id"] = int(cursor.lastrowid)
+            self.connection.execute(
+                "UPDATE tasks SET current_run_id = ?, claim_lock = ? WHERE id = ?",
+                (replacement["run_id"], "replacement-lock", task_id),
+            )
+            return {}
+
+        with mock.patch.object(
+            kanban_db,
+            "_terminate_reclaimed_worker",
+            side_effect=install_replacement,
+        ):
+            self.assertFalse(
+                kanban_db.reclaim_task(
+                    self.connection,
+                    task_id,
+                    reason="journal for old run",
+                    expected_run_id=old_run,
+                )
+            )
+
+        latest = kanban_db.get_task(self.connection, task_id)
+        self.assertEqual(latest.status, "running")
+        self.assertEqual(latest.current_run_id, replacement["run_id"])
+        self.assertEqual(latest.claim_lock, "replacement-lock")
+
+    def test_concurrent_exact_reclaim_and_completion_have_one_winner(self) -> None:
+        task_id = kanban_db.create_task(self.connection, title="concurrent finalizer")
+        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+        run_id = kanban_db.get_task(self.connection, task_id).current_run_id
+        barrier = threading.Barrier(2)
+        outcomes = {}
+        errors = []
+
+        def reclaim() -> None:
+            try:
+                with kanban_db.connect_closing() as connection:
+                    barrier.wait()
+                    outcomes["reclaim"] = kanban_db.reclaim_task(
+                        connection,
+                        task_id,
+                        reason="concurrent invalid journal",
+                        expected_run_id=run_id,
+                    )
+            except BaseException as error:  # pragma: no cover - assertion relay
+                errors.append(error)
+
+        def finalize() -> None:
+            try:
+                with kanban_db.connect_closing() as connection:
+                    barrier.wait()
+                    outcomes["complete"] = kanban_db.complete_task(
+                        connection,
+                        task_id,
+                        result="durable winner",
+                        summary="durable winner",
+                        expected_run_id=run_id,
+                    )
+            except BaseException as error:  # pragma: no cover - assertion relay
+                errors.append(error)
+
+        threads = [threading.Thread(target=reclaim), threading.Thread(target=finalize)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join(timeout=10)
+            self.assertFalse(thread.is_alive())
+
+        self.assertEqual(errors, [])
+        self.assertEqual(set(outcomes), {"reclaim", "complete"})
+        self.assertEqual(sum(bool(value) for value in outcomes.values()), 1)
+        latest = kanban_db.get_task(self.connection, task_id)
+        self.assertIn(latest.status, {"done", "ready"})
+        self.assertIsNone(latest.current_run_id)
+        runs = kanban_db.list_runs(self.connection, task_id)
+        self.assertEqual(len(runs), 1)
+        self.assertIsNotNone(runs[0].ended_at)
+
+    def test_fresh_triage_task_still_auto_promotes(self) -> None:
+        task_id = kanban_db.create_task(
+            self.connection,
+            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 _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,
+        )
+
+        completions = self._client(
+            {
+                "fanout": True,
+                "tasks": [
+                    {
+                        "title": "redundant child",
+                        "body": "must not be created",
+                        "assignee": "default",
+                        "parents": [],
+                    }
+                ],
+            },
+            before_response=lambda: self._insert_run(task_id),
+        )
+        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,702 @@
+"""Patch upstream Hermes exact-run completion and decomposition safety."""
+
+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"))
+db_path = source_root / "hermes_cli/kanban_db.py"
+db = db_path.read_text(encoding="utf-8")
+
+complete_signature_before = '''def complete_task(
+    conn: sqlite3.Connection,
+    task_id: str,
+    *,
+    result: Optional[str] = None,
+    summary: Optional[str] = None,
+    metadata: Optional[dict] = None,
+    created_cards: Optional[Iterable[str]] = None,
+    expected_run_id: Optional[int] = None,
+) -> bool:
+'''
+complete_signature_after = '''def complete_task(
+    conn: sqlite3.Connection,
+    task_id: str,
+    *,
+    result: Optional[str] = None,
+    summary: Optional[str] = None,
+    metadata: Optional[dict] = None,
+    created_cards: Optional[Iterable[str]] = None,
+    expected_run_id: Optional[int] = None,
+    replay_ended_run_id: Optional[int] = None,
+) -> bool:
+'''
+db = replace_once(
+    db,
+    complete_signature_before,
+    complete_signature_after,
+    "exact ended-run completion signature",
+)
+
+complete_branch_before = '''    with write_txn(conn):
+        if expected_run_id is None:
+            cur = conn.execute(
+                """
+                UPDATE tasks
+                   SET status       = 'done',
+                       result       = ?,
+                       completed_at = ?,
+                       claim_lock   = NULL,
+                       claim_expires= NULL,
+                       worker_pid   = NULL,
+                       block_kind   = NULL,
+                       block_recurrences = 0
+                 WHERE id = ?
+                   AND status IN ('running', 'ready', 'blocked', 'scheduled')
+                """,
+                (result, now, task_id),
+            )
+        else:
+'''
+complete_branch_after = '''    # Journal recovery may arrive after a legacy error path ended the
+    # accepted run. Verify that it is still the latest run in this same
+    # transaction; a replacement attempt must make the replay fail closed.
+    with write_txn(conn):
+        replayed_run_id = None
+        if expected_run_id is not None and replay_ended_run_id is not None:
+            raise ValueError("expected_run_id and replay_ended_run_id are mutually exclusive")
+        if replay_ended_run_id is not None:
+            replayed_run_id = int(replay_ended_run_id)
+            cur = conn.execute(
+                """
+                UPDATE tasks
+                   SET status       = 'done',
+                       result       = ?,
+                       completed_at = ?,
+                       claim_lock   = NULL,
+                       claim_expires= NULL,
+                       worker_pid   = NULL,
+                       block_kind   = NULL,
+                       block_recurrences = 0
+                 WHERE id = ?
+                   AND status IN ('ready', 'blocked')
+                   AND current_run_id IS NULL
+                   AND ? = (
+                       SELECT id FROM task_runs
+                        WHERE task_id = ?
+                        ORDER BY id DESC
+                        LIMIT 1
+                   )
+                   AND EXISTS (
+                       SELECT 1 FROM task_runs
+                        WHERE id = ?
+                          AND task_id = ?
+                          AND ended_at IS NOT NULL
+                   )
+                """,
+                (
+                    result,
+                    now,
+                    task_id,
+                    replayed_run_id,
+                    task_id,
+                    replayed_run_id,
+                    task_id,
+                ),
+            )
+        elif expected_run_id is None:
+            cur = conn.execute(
+                """
+                UPDATE tasks
+                   SET status       = 'done',
+                       result       = ?,
+                       completed_at = ?,
+                       claim_lock   = NULL,
+                       claim_expires= NULL,
+                       worker_pid   = NULL,
+                       block_kind   = NULL,
+                       block_recurrences = 0
+                 WHERE id = ?
+                   AND status IN ('running', 'ready', 'blocked', 'scheduled')
+                """,
+                (result, now, task_id),
+            )
+        else:
+'''
+db = replace_once(
+    db,
+    complete_branch_before,
+    complete_branch_after,
+    "transactional exact ended-run completion",
+)
+
+run_completion_before = '''        run_id = _end_run(
+            conn, task_id,
+            outcome="completed", status="done",
+            summary=summary if summary is not None else result,
+            metadata=metadata,
+        )
+        # If complete_task was called on a never-claimed task (ready or
+        # blocked → done with no run in flight), synthesize a
+        # zero-duration run so the handoff fields are persisted in
+        # attempt history instead of silently lost.
+        if run_id is None and (summary or metadata or result):
+            run_id = _synthesize_ended_run(
+                conn, task_id,
+                outcome="completed",
+                summary=summary if summary is not None else result,
+                metadata=metadata,
+            )
+'''
+run_completion_after = '''        if replayed_run_id is None:
+            run_id = _end_run(
+                conn, task_id,
+                outcome="completed", status="done",
+                summary=summary if summary is not None else result,
+                metadata=metadata,
+            )
+            # If complete_task was called on a never-claimed task (ready or
+            # blocked → done with no run in flight), synthesize a
+            # zero-duration run so the handoff fields are persisted in
+            # attempt history instead of silently lost.
+            if run_id is None and (summary or metadata or result):
+                run_id = _synthesize_ended_run(
+                    conn, task_id,
+                    outcome="completed",
+                    summary=summary if summary is not None else result,
+                    metadata=metadata,
+                )
+        else:
+            run_id = replayed_run_id
+            replayed = conn.execute(
+                """
+                UPDATE task_runs
+                   SET status = 'done',
+                       outcome = 'completed',
+                       summary = ?,
+                       error = NULL,
+                       metadata = ?,
+                       claim_lock = NULL,
+                       claim_expires = NULL,
+                       worker_pid = NULL
+                 WHERE id = ?
+                   AND task_id = ?
+                   AND ended_at IS NOT NULL
+                """,
+                (
+                    summary if summary is not None else result,
+                    json.dumps(metadata, ensure_ascii=False) if metadata else None,
+                    run_id,
+                    task_id,
+                ),
+            )
+            if replayed.rowcount != 1:
+                raise RuntimeError("exact ended-run completion lost its run identity")
+        conn.execute(
+            "UPDATE tasks SET completed_run_id = ? WHERE id = ?",
+            (run_id, task_id),
+        )
+'''
+db = replace_once(
+    db,
+    run_completion_before,
+    run_completion_after,
+    "exact completion run provenance",
+)
+
+reclaim_before = '''def reclaim_task(
+    conn: sqlite3.Connection,
+    task_id: str,
+    *,
+    reason: Optional[str] = None,
+    signal_fn=None,
+) -> bool:
+    """Operator-driven reclaim: release the claim and reset to ``ready``.
+
+    Unlike :func:`release_stale_claims` which only acts on tasks whose
+    ``claim_expires`` has passed, this function reclaims immediately
+    regardless of TTL. Intended for the dashboard/CLI recovery flow
+    when an operator wants to abort a running worker without waiting
+    for the TTL to expire (e.g. after seeing a hallucination warning).
+
+    Returns True if a reclaim happened, False if the task isn't in a
+    reclaimable state (not running, or doesn't exist).
+    """
+    row = conn.execute(
+        "SELECT status, claim_lock, worker_pid FROM tasks WHERE id = ?",
+        (task_id,),
+    ).fetchone()
+    if not row:
+        return False
+    if row["status"] != "running" and row["claim_lock"] is None:
+        # Nothing to reclaim — already ready / blocked / done.
+        return False
+    prev_lock = row["claim_lock"]
+    termination = _terminate_reclaimed_worker(
+        row["worker_pid"], prev_lock, signal_fn=signal_fn,
+    )
+    with write_txn(conn):
+        cur = conn.execute(
+            "UPDATE tasks SET status = 'ready', claim_lock = NULL, "
+            "claim_expires = NULL, worker_pid = NULL "
+            "WHERE id = ? AND status IN ('running', 'ready', 'blocked') "
+            "AND claim_lock IS ?",
+            (task_id, prev_lock),
+        )
+        if cur.rowcount != 1:
+            return False
+        run_id = _end_run(
+            conn, task_id,
+            outcome="reclaimed", status="reclaimed",
+            error=(
+                f"manual_reclaim: {reason}" if reason
+                else f"manual_reclaim lock={prev_lock}"
+            ),
+            metadata=termination,
+        )
+        payload = {
+            "manual": True,
+            "reason": reason,
+            "prev_lock": prev_lock,
+        }
+        payload.update(termination)
+        _append_event(
+            conn, task_id, "reclaimed",
+            payload,
+            run_id=run_id,
+        )
+    # Operator intervention — they've looked at the task, so the
+    # consecutive-failures counter is now stale. Give the next retry
+    # a fresh budget. (_clear_failure_counter opens its own write_txn,
+    # so it runs after the enclosing one commits.)
+    _clear_failure_counter(conn, task_id)
+    return True
+'''
+reclaim_after = '''def reclaim_task(
+    conn: sqlite3.Connection,
+    task_id: str,
+    *,
+    reason: Optional[str] = None,
+    signal_fn=None,
+    expected_run_id: Optional[int] = None,
+) -> bool:
+    """Operator-driven reclaim: release the claim and reset to ``ready``.
+
+    Unlike :func:`release_stale_claims` which only acts on tasks whose
+    ``claim_expires`` has passed, this function reclaims immediately
+    regardless of TTL. Intended for the dashboard/CLI recovery flow
+    when an operator wants to abort a running worker without waiting
+    for the TTL to expire (e.g. after seeing a hallucination warning).
+
+    When ``expected_run_id`` is supplied, the run identity is checked after
+    ``BEGIN IMMEDIATE`` and included in the guarded update. A replacement run
+    is therefore neither signalled nor reclaimed by stale recovery evidence.
+
+    Returns True if a reclaim happened, False if the task isn't in a
+    reclaimable state (not running, or doesn't exist), or its run changed.
+    """
+    with write_txn(conn):
+        row = conn.execute(
+            "SELECT status, claim_lock, worker_pid, current_run_id "
+            "FROM tasks WHERE id = ?",
+            (task_id,),
+        ).fetchone()
+        if not row:
+            return False
+        if expected_run_id is not None and row["current_run_id"] != expected_run_id:
+            return False
+        if row["status"] != "running" and row["claim_lock"] is None:
+            # Nothing to reclaim — already ready / blocked / done.
+            return False
+        prev_lock = row["claim_lock"]
+        termination = _terminate_reclaimed_worker(
+            row["worker_pid"], prev_lock, signal_fn=signal_fn,
+        )
+        if expected_run_id is None:
+            cur = conn.execute(
+                "UPDATE tasks SET status = 'ready', claim_lock = NULL, "
+                "claim_expires = NULL, worker_pid = NULL "
+                "WHERE id = ? AND status IN ('running', 'ready', 'blocked') "
+                "AND claim_lock IS ?",
+                (task_id, prev_lock),
+            )
+        else:
+            cur = conn.execute(
+                "UPDATE tasks SET status = 'ready', claim_lock = NULL, "
+                "claim_expires = NULL, worker_pid = NULL "
+                "WHERE id = ? AND status IN ('running', 'ready', 'blocked') "
+                "AND claim_lock IS ? AND current_run_id = ?",
+                (task_id, prev_lock, int(expected_run_id)),
+            )
+        if cur.rowcount != 1:
+            return False
+        run_id = _end_run(
+            conn, task_id,
+            outcome="reclaimed", status="reclaimed",
+            error=(
+                f"manual_reclaim: {reason}" if reason
+                else f"manual_reclaim lock={prev_lock}"
+            ),
+            metadata=termination,
+        )
+        payload = {
+            "manual": True,
+            "reason": reason,
+            "prev_lock": prev_lock,
+        }
+        payload.update(termination)
+        _append_event(
+            conn, task_id, "reclaimed",
+            payload,
+            run_id=run_id,
+        )
+    # Operator intervention — they've looked at the task, so the
+    # consecutive-failures counter is now stale. Give the next retry
+    # a fresh budget. (_clear_failure_counter opens its own write_txn,
+    # so it runs after the enclosing one commits.)
+    _clear_failure_counter(conn, task_id)
+    return True
+'''
+db = replace_once(
+    db,
+    reclaim_before,
+    reclaim_after,
+    "transactional exact-run reclaim",
+)
+
+task_field_before = '''    current_run_id: Optional[int] = None
+    workflow_template_id: Optional[str] = None
+'''
+task_field_after = '''    current_run_id: Optional[int] = None
+    # Durable provenance for the run that authoritatively completed this task.
+    completed_run_id: Optional[int] = None
+    workflow_template_id: Optional[str] = None
+'''
+db = replace_once(
+    db,
+    task_field_before,
+    task_field_after,
+    "completed-run task field",
+)
+
+task_row_before = '''            current_run_id=(
+                row["current_run_id"] if "current_run_id" in keys else None
+            ),
+            workflow_template_id=(
+'''
+task_row_after = '''            current_run_id=(
+                row["current_run_id"] if "current_run_id" in keys else None
+            ),
+            completed_run_id=(
+                row["completed_run_id"] if "completed_run_id" in keys else None
+            ),
+            workflow_template_id=(
+'''
+db = replace_once(
+    db,
+    task_row_before,
+    task_row_after,
+    "completed-run row mapping",
+)
+
+task_schema_before = '''    -- run is in-flight). Denormalised for cheap reads.
+    current_run_id       INTEGER,
+    -- Forward-compat for v2 workflow routing. In v1 the kernel writes
+'''
+task_schema_after = '''    -- run is in-flight). Denormalised for cheap reads.
+    current_run_id       INTEGER,
+    -- Immutable provenance for the task result's authoritative run.
+    completed_run_id     INTEGER,
+    -- Forward-compat for v2 workflow routing. In v1 the kernel writes
+'''
+db = replace_once(
+    db,
+    task_schema_before,
+    task_schema_after,
+    "completed-run schema",
+)
+
+task_migration_before = '''    if "current_run_id" not in cols:
+        _add_column_if_missing(
+            conn, "tasks", "current_run_id", "current_run_id INTEGER"
+        )
+    if "workflow_template_id" not in cols:
+'''
+task_migration_after = '''    if "current_run_id" not in cols:
+        _add_column_if_missing(
+            conn, "tasks", "current_run_id", "current_run_id INTEGER"
+        )
+    if "completed_run_id" not in cols:
+        _add_column_if_missing(
+            conn, "tasks", "completed_run_id", "completed_run_id INTEGER"
+        )
+    if "workflow_template_id" not in cols:
+'''
+db = replace_once(
+    db,
+    task_migration_before,
+    task_migration_after,
+    "completed-run migration",
+)
+
+specify_signature_before = '''def specify_triage_task(
+    conn: sqlite3.Connection,
+    task_id: str,
+    *,
+    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")
+
+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 = '''                author=audit_author,
+            )
+        if not ok:
+'''
+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,
+    single_before,
+    single_after,
+    "auto-decompose single-task commit guard",
+)
+
+fanout_before = '''                author=audit_author,
+                auto_promote=auto_promote,
+            )
+'''
+fanout_after = '''                author=audit_author,
+                auto_promote=auto_promote,
+                require_no_runs=automatic,
+            )
+'''
+decompose = replace_once(
+    decompose,
+    fanout_before,
+    fanout_after,
+    "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")
+
+
+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/recover_cassandra_workers.py

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

testing/tests/test_hermes_worker_recovery.py

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