hermes: bind terminal commit to journal inode

Reviewed-on: atlas/titan-iac#13

dockerfiles/Dockerfile.hermes-agent

@@ -629,16 +629,125 @@ source = source[:runs_start] + runs_source
 path.write_text(source)
 PY
 
-# A manual evidence-based close must not expose a parked task to the dispatcher
-# between separate `unblock` and `complete` commands. Upstream already permits
-# direct completion from `blocked`; extend the same atomic path to `scheduled`
-# and make that contract visible in CLI help so agents do not create a ready
-# window that can launch redundant work.
+# ``create_task(initial_status="blocked")`` is an explicit operator park, but
+# upstream only treats later block_task() events as sticky. Recognize the
+# existing created(status=blocked) event too. This also protects tasks created
+# before this image patch without making dependency or circuit-breaker blocks
+# depend on a new persisted field.
 RUN python - <<'PY'
 from pathlib import Path
 
 db_path = Path("/opt/hermes/hermes_cli/kanban_db.py")
 db_source = db_path.read_text()
+
+# The rewritten sticky-block gate below classifies a task from the ``created``
+# event payload that ``create_task`` writes. That producer is upstream code we
+# do not own, so anchor the exact contract the consumer depends on instead of
+# trusting it: an ``initial_status="blocked"`` park must resolve ``task_status``
+# to ``"blocked"``, every other creation must resolve it to something else, and
+# the ``created`` event must carry that same variable under the ``"status"``
+# key. If upstream renames the key, drops it, hardcodes it, or stops deriving it
+# from ``initial_status``, fail the build here rather than ship a consumer that
+# silently mis-classifies every task it reads. These assert only -- they
+# deliberately do not rewrite the producer. The regression suite copied in below
+# is the second net: it proves the same contract against the real create_task,
+# for the drift these textual anchors cannot see.
+producer_park_anchor = '''                if initial_status == "blocked":
+                    task_status = "blocked"
+'''
+if db_source.count(producer_park_anchor) != 1:
+    raise SystemExit(
+        "Hermes Kanban create_task park semantics changed: expected 1, "
+        f"found {db_source.count(producer_park_anchor)}"
+    )
+# The mirror of the park branch: every non-park creation must resolve to a
+# status the gate does *not* read as a park. Without this, drift that widened
+# the ladder to park ordinary work would leave both other anchors intact.
+producer_unparked_anchor = '''                elif triage:
+                    task_status = "triage"
+                else:
+                    task_status = "ready"
+'''
+if db_source.count(producer_unparked_anchor) != 1:
+    raise SystemExit(
+        "Hermes Kanban create_task non-park status ladder changed: expected 1, "
+        f"found {db_source.count(producer_unparked_anchor)}"
+    )
+producer_event_anchor = '''                _append_event(
+                    conn,
+                    task_id,
+                    "created",
+                    {
+                        "assignee": assignee,
+                        "status": task_status,
+'''
+if db_source.count(producer_event_anchor) != 1:
+    raise SystemExit(
+        "Hermes Kanban created-event status payload changed: expected 1, "
+        f"found {db_source.count(producer_event_anchor)}"
+    )
+
+sticky_doc_before = '''    The cheapest signal that distinguishes the two is the most recent
+    ``"blocked"`` / ``"unblocked"`` event for the task.  If the most
+    recent one is ``"blocked"`` (or there is a ``"blocked"`` event and
+    no ``"unblocked"`` event has fired since), the task is sticky and
+    ``recompute_ready`` must *not* auto-promote it.
+
+    Returns ``False`` when there is no such event at all (e.g. the task
+'''
+sticky_doc_after = '''    The cheapest signal that distinguishes the two is the most recent
+    ``"created"`` / ``"blocked"`` / ``"unblocked"`` event for the task.
+    A ``"created"`` event whose payload has ``status="blocked"`` is the
+    explicit park requested by ``create_task(initial_status="blocked")``.
+    A later ``"unblocked"`` event clears either kind of sticky block.
+
+    Returns ``False`` when there is no such event at all (e.g. the task
+'''
+if db_source.count(sticky_doc_before) != 1:
+    raise SystemExit(
+        "Hermes Kanban sticky-block documentation changed: expected 1, "
+        f"found {db_source.count(sticky_doc_before)}"
+    )
+db_source = db_source.replace(sticky_doc_before, sticky_doc_after, 1)
+
+sticky_before = '''    row = conn.execute(
+        "SELECT kind FROM task_events "
+        "WHERE task_id = ? AND kind IN ('blocked', 'unblocked') "
+        "ORDER BY id DESC LIMIT 1",
+        (task_id,),
+    ).fetchone()
+    return bool(row) and row["kind"] == "blocked"
+'''
+sticky_after = '''    row = conn.execute(
+        "SELECT kind, payload FROM task_events "
+        "WHERE task_id = ? AND kind IN ('created', 'blocked', 'unblocked') "
+        "ORDER BY id DESC LIMIT 1",
+        (task_id,),
+    ).fetchone()
+    if not row or row["kind"] == "unblocked":
+        return False
+    if row["kind"] == "blocked":
+        return True
+    try:
+        payload = json.loads(row["payload"] or "{}")
+    except (TypeError, ValueError):
+        return False
+    if not isinstance(payload, dict):
+        return False
+    return payload.get("status") == "blocked"
+'''
+if db_source.count(sticky_before) != 1:
+    raise SystemExit(
+        "Hermes Kanban sticky-block gate changed: expected 1, "
+        f"found {db_source.count(sticky_before)}"
+    )
+db_source = db_source.replace(sticky_before, sticky_after, 1)
+
+# A manual evidence-based close must not expose a parked task to the dispatcher
+# between separate `unblock` and `complete` commands. Upstream already permits
+# direct completion from `blocked`; extend the same atomic path to `scheduled`
+# and make that contract visible in CLI help so agents do not create a ready
+# window that can launch redundant work.
 state_before = '''                 WHERE id = ?
                    AND status IN ('running', 'ready', 'blocked')
 '''
@@ -671,6 +780,10 @@ if cli_source.count(help_before) != 1:
 cli_path.write_text(cli_source.replace(help_before, help_after, 1))
 PY
 
+COPY dockerfiles/hermes-kanban-blocked-regression.py /tmp/hermes-kanban-blocked-regression.py
+RUN /opt/hermes/.venv/bin/python /tmp/hermes-kanban-blocked-regression.py \
+    && rm /tmp/hermes-kanban-blocked-regression.py
+
 COPY dockerfiles/hermes-python-sandbox-tool.py /opt/hermes/tools/python_sandbox_tool.py
 COPY dockerfiles/hermes-public-extract/__init__.py /opt/hermes/plugins/web/public_extract/__init__.py
 COPY dockerfiles/hermes-public-extract/plugin.yaml /opt/hermes/plugins/web/public_extract/plugin.yaml
@@ -1431,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 \
@@ -1489,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

@@ -1,6 +1,9 @@
 **
+!dockerfiles/hermes-kanban-blocked-regression.py
 !dockerfiles/hermes-python-sandbox-tool.py
 !dockerfiles/hermes-public-extract/
 !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,564 @@
+"""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_unchanged_triage_can_replay_its_exact_latest_ended_run(self) -> None:
+        task_id = self._route_executed_task_to_triage()
+        runs = kanban_db.list_runs(self.connection, task_id)
+        run_id = runs[-1].id
+
+        self.assertTrue(
+            kanban_db.complete_task(
+                self.connection,
+                task_id,
+                result="accepted before the loop breaker routed triage",
+                summary="accepted before the loop breaker routed triage",
+                replay_ended_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)
+        updated_runs = kanban_db.list_runs(self.connection, task_id)
+        self.assertEqual(len(updated_runs), kanban_db.BLOCK_RECURRENCE_LIMIT)
+        self.assertEqual(updated_runs[-1].id, run_id)
+        self.assertEqual(updated_runs[-1].outcome, "completed")
+
+    def test_triage_status_transition_prevents_stale_exact_run_replay(self) -> None:
+        task_id = self._route_executed_task_to_triage()
+        run_id = kanban_db.list_runs(self.connection, task_id)[-1].id
+        self.assertTrue(
+            kanban_db.specify_triage_task(
+                self.connection,
+                task_id,
+                title="operator changed the objective",
+                author="operator",
+            )
+        )
+
+        self.assertFalse(
+            kanban_db.complete_task(
+                self.connection,
+                task_id,
+                result="stale accepted result",
+                replay_ended_run_id=run_id,
+            )
+        )
+        self.assertNotEqual(self._status(task_id), "done")
+
+    def test_decomposition_lineage_prevents_replay_even_if_status_returns(self) -> None:
+        task_id = self._route_executed_task_to_triage()
+        run_id = kanban_db.list_runs(self.connection, task_id)[-1].id
+        children = kanban_db.decompose_triage_task(
+            self.connection,
+            task_id,
+            root_assignee="default",
+            children=[{"title": "new child objective"}],
+            author="operator",
+            auto_promote=False,
+        )
+        self.assertEqual(len(children or []), 1)
+        with kanban_db.write_txn(self.connection):
+            self.connection.execute(
+                "UPDATE tasks SET status = 'ready' WHERE id = ?",
+                (task_id,),
+            )
+
+        self.assertFalse(
+            kanban_db.complete_task(
+                self.connection,
+                task_id,
+                result="pre-decomposition accepted result",
+                replay_ended_run_id=run_id,
+            )
+        )
+        self.assertEqual(self._status(task_id), "ready")
+
+    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/hermes-kanban-blocked-regression.py

@@ -0,0 +1,249 @@
+"""Regression tests for Hermes Kanban blocked-task scheduling semantics."""
+
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import unittest
+from pathlib import Path
+
+
+_HERMES_HOME = tempfile.TemporaryDirectory(prefix="hermes-kanban-test-home-")
+os.environ["HERMES_HOME"] = _HERMES_HOME.name
+
+from hermes_cli import kanban_db  # noqa: E402
+
+
+class BlockedTaskSchedulingTests(unittest.TestCase):
+    """Exercise the real upstream database API against isolated databases."""
+
+    def setUp(self) -> None:
+        self._database_dir = tempfile.TemporaryDirectory(
+            prefix="hermes-kanban-test-db-"
+        )
+        self.connection = kanban_db.connect(
+            Path(self._database_dir.name) / "kanban.db"
+        )
+
+    def tearDown(self) -> None:
+        self.connection.close()
+        self._database_dir.cleanup()
+
+    def create_task(self, title: str, **kwargs: object) -> str:
+        return kanban_db.create_task(self.connection, title=title, **kwargs)
+
+    def status(self, task_id: str) -> str:
+        task = kanban_db.get_task(self.connection, task_id)
+        self.assertIsNotNone(task)
+        return task.status
+
+    def created_payload(self, task_id: str) -> dict:
+        """Return the decoded payload of the task's ``created`` event."""
+        row = self.connection.execute(
+            "SELECT payload FROM task_events "
+            "WHERE task_id = ? AND kind = 'created' ORDER BY id LIMIT 1",
+            (task_id,),
+        ).fetchone()
+        self.assertIsNotNone(row, "create_task emitted no 'created' event")
+        return json.loads(row["payload"] or "{}")
+
+    def set_created_payload(self, task_id: str, payload: object) -> None:
+        """Overwrite a ``created`` event payload to simulate producer drift."""
+        with self.connection:
+            self.connection.execute(
+                "UPDATE task_events SET payload = ? "
+                "WHERE task_id = ? AND kind = 'created'",
+                (payload, task_id),
+            )
+
+    def block_via_circuit_breaker(self, title: str) -> str:
+        """Create + claim a normal task, then trip the breaker once on it.
+
+        The resulting task is ``blocked`` with exactly one recorded failure and
+        -- critically -- no ``blocked`` event, because the breaker emits
+        ``gave_up``. Its most recent created/blocked/unblocked event is
+        therefore its own ``created`` event, which is what makes it the right
+        probe for the created-event branch of the sticky gate.
+        """
+        task_id = self.create_task(title)
+        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+        self.assertTrue(
+            kanban_db._record_spawn_failure(
+                self.connection,
+                task_id,
+                "worker failed",
+                failure_limit=1,
+            )
+        )
+        self.assertEqual(self.status(task_id), "blocked")
+        return task_id
+
+    def test_initial_block_without_parents_is_sticky(self) -> None:
+        task_id = self.create_task("operator parked", initial_status="blocked")
+
+        self.assertEqual(kanban_db.recompute_ready(self.connection), 0)
+        self.assertEqual(self.status(task_id), "blocked")
+
+    def test_initial_block_with_complete_parents_is_sticky(self) -> None:
+        parent_id = self.create_task("completed prerequisite")
+        self.assertTrue(kanban_db.complete_task(self.connection, parent_id))
+        task_id = self.create_task(
+            "operator parked after prerequisite",
+            initial_status="blocked",
+            parents=[parent_id],
+        )
+
+        self.assertEqual(kanban_db.recompute_ready(self.connection), 0)
+        self.assertEqual(self.status(task_id), "blocked")
+
+    def test_initial_block_remains_sticky_when_parent_later_completes(self) -> None:
+        parent_id = self.create_task("incomplete prerequisite")
+        task_id = self.create_task(
+            "operator parked before prerequisite completes",
+            initial_status="blocked",
+            parents=[parent_id],
+        )
+
+        self.assertEqual(self.status(task_id), "blocked")
+        self.assertTrue(kanban_db.complete_task(self.connection, parent_id))
+        self.assertEqual(self.status(task_id), "blocked")
+
+    def test_initial_block_is_released_by_an_explicit_unblock(self) -> None:
+        task_id = self.create_task("operator parked", initial_status="blocked")
+
+        self.assertTrue(kanban_db.unblock_task(self.connection, task_id))
+        self.assertEqual(self.status(task_id), "ready")
+
+    def test_dependency_block_promotes_after_parent_completes(self) -> None:
+        parent_id = self.create_task("incomplete prerequisite")
+        task_id = self.create_task("dependency gated", parents=[parent_id])
+        self.assertEqual(self.status(task_id), "todo")
+
+        self.assertTrue(kanban_db.complete_task(self.connection, parent_id))
+        self.assertEqual(self.status(task_id), "ready")
+
+    def test_explicit_block_and_unblock_remain_sticky_and_reversible(self) -> None:
+        task_id = self.create_task("worker handoff")
+
+        self.assertTrue(
+            kanban_db.block_task(
+                self.connection,
+                task_id,
+                reason="human review required",
+            )
+        )
+        self.assertEqual(kanban_db.recompute_ready(self.connection), 0)
+        self.assertEqual(self.status(task_id), "blocked")
+        self.assertTrue(kanban_db.unblock_task(self.connection, task_id))
+        self.assertEqual(self.status(task_id), "ready")
+
+    def test_circuit_breaker_block_is_not_auto_promoted(self) -> None:
+        task_id = self.create_task("repeated worker failure")
+        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
+
+        # Upstream has no public circuit-breaker entry point; exercise the
+        # dispatcher helper that owns its persisted failure-limit semantics.
+        self.assertTrue(
+            kanban_db._record_spawn_failure(
+                self.connection,
+                task_id,
+                "worker failed",
+                failure_limit=1,
+            )
+        )
+        self.assertEqual(self.status(task_id), "blocked")
+        self.assertEqual(
+            kanban_db.recompute_ready(self.connection, failure_limit=1),
+            0,
+        )
+        self.assertEqual(self.status(task_id), "blocked")
+
+    def test_circuit_breaker_block_is_promoted_below_the_failure_limit(self) -> None:
+        """Non-sticky oracle for the created-event branch of the sticky gate.
+
+        A circuit-breaker block must still auto-recover once the effective
+        failure limit rises above the recorded failure count -- that is the
+        pre-existing upstream contract the sticky-block patch must not widen.
+
+        This task's most recent created/blocked/unblocked event is its own
+        ``created`` event carrying ``status="ready"``, so it is exactly the
+        case that a mis-classifying gate strands. It fails if the gate treats
+        any ``created`` event as a park (over-broad consumer) and it fails if
+        ``create_task`` drifts to labelling every ``created`` event
+        ``"blocked"`` (over-broad producer), while the
+        circuit-breaker-at-current-limit case above still holds.
+        """
+        task_id = self.block_via_circuit_breaker("recoverable worker failure")
+
+        self.assertFalse(
+            kanban_db._has_sticky_block(self.connection, task_id),
+            "a circuit-breaker block must never be classified as a park",
+        )
+        self.assertEqual(
+            kanban_db.recompute_ready(self.connection, failure_limit=2),
+            1,
+        )
+        self.assertEqual(self.status(task_id), "ready")
+
+    def test_created_event_carries_the_status_the_sticky_gate_reads(self) -> None:
+        """Producer contract asserted directly against the real create_task.
+
+        The sticky gate reads ``created`` payload ``status``. Pin both poles of
+        that contract so producer drift is a test failure here as well as a
+        build failure at the Dockerfile source anchor.
+        """
+        parked_id = self.create_task("operator parked", initial_status="blocked")
+        normal_id = self.create_task("ordinary work")
+
+        self.assertEqual(self.created_payload(parked_id).get("status"), "blocked")
+        self.assertEqual(self.created_payload(normal_id).get("status"), "ready")
+
+    def test_non_blocked_created_payloads_are_not_treated_as_a_park(self) -> None:
+        """Only ``status="blocked"`` parks; every other value must promote."""
+        for payload_status in ("ready", "todo", "running", "triage"):
+            with self.subTest(payload_status=payload_status):
+                task_id = self.block_via_circuit_breaker(
+                    f"breaker trip with {payload_status} created payload"
+                )
+                self.set_created_payload(
+                    task_id, json.dumps({"status": payload_status})
+                )
+
+                self.assertFalse(
+                    kanban_db._has_sticky_block(self.connection, task_id)
+                )
+                self.assertEqual(
+                    kanban_db.recompute_ready(self.connection, failure_limit=2),
+                    1,
+                )
+                self.assertEqual(self.status(task_id), "ready")
+
+    def test_malformed_created_payloads_fail_closed_to_not_parked(self) -> None:
+        """Undecodable or non-object payloads must not park -- or crash.
+
+        ``recompute_ready`` walks every blocked task in one pass, so an
+        exception escaping the gate would stall promotion board-wide. Falling
+        back to "not parked" preserves upstream's pre-patch auto-recover
+        default for rows the gate cannot interpret.
+        """
+        malformed = (None, "", "   ", "{not json", "[]", "null", '"blocked"', "17")
+        for payload in malformed:
+            with self.subTest(payload=payload):
+                task_id = self.block_via_circuit_breaker(
+                    f"breaker trip with payload {payload!r}"
+                )
+                self.set_created_payload(task_id, payload)
+
+                self.assertFalse(
+                    kanban_db._has_sticky_block(self.connection, task_id)
+                )
+                self.assertEqual(
+                    kanban_db.recompute_ready(self.connection, failure_limit=2),
+                    1,
+                )
+                self.assertEqual(self.status(task_id), "ready")
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

dockerfiles/patch-hermes-execution-safety.py

@@ -0,0 +1,737 @@
+"""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)
+            # Replay is permitted only while the exact run-ending transition
+            # remains the latest status mutation. Comments and diagnostics do
+            # not invalidate durable accepted work, but an unblock, explicit
+            # triage specification, decomposition, reclaim, or newer attempt
+            # does. BEGIN IMMEDIATE keeps this lineage check and the guarded
+            # task/run updates in one transaction.
+            last_transition = conn.execute(
+                """
+                SELECT kind, run_id
+                  FROM task_events
+                 WHERE task_id = ?
+                   AND kind IN (
+                       'archived', 'block_loop_detected', 'blocked', 'claimed',
+                       'completed', 'decomposed', 'dependency_wait', 'gave_up',
+                       'promoted', 'promoted_manual', 'reclaimed', 'scheduled',
+                       'specified', 'stale', 'timed_out', 'unblocked'
+                   )
+                 ORDER BY id DESC
+                 LIMIT 1
+                """,
+                (task_id,),
+            ).fetchone()
+            if (
+                last_transition is None
+                or last_transition["run_id"] != replayed_run_id
+                or last_transition["kind"] not in {
+                    "block_loop_detected",
+                    "blocked",
+                    "reclaimed",
+                    "stale",
+                    "timed_out",
+                    "gave_up",
+                }
+            ):
+                return False
+            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', 'triage')
+                   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/agent-deployment.yaml

@@ -91,6 +91,7 @@ spec:
         vault.hashicorp.com/agent-limits-mem: 128Mi
     spec:
       serviceAccountName: hermes-agent
+      enableServiceLinks: false
       automountServiceAccountToken: true
       securityContext:
         fsGroup: 10000
@@ -110,7 +111,7 @@ spec:
                     values: ["true"]
                   - key: kubernetes.io/hostname
                     operator: NotIn
-                    values: [titan-08, titan-13, titan-14, titan-17, titan-18]
+                    values: [titan-04, titan-08, titan-13, titan-14, titan-17, titan-18, titan-19]
           preferredDuringSchedulingIgnoredDuringExecution:
             - weight: 100
               preference:
@@ -689,7 +690,7 @@ spec:
             seccompProfile:
               type: RuntimeDefault
           resources:
-            requests: {cpu: 350m, memory: 768Mi}
+            requests: {cpu: 300m, memory: 768Mi}
             limits: {cpu: "3", memory: 6Gi}
         - name: oauth2-proxy
           image: quay.io/oauth2-proxy/oauth2-proxy:v7.15.3@sha256:10a1165743a192e1940b4708fb9647027185ce11a681a1c5519b442ff7f1f561
@@ -873,7 +874,7 @@ spec:
             - {name: CLAUDE_CONFIG_DIR, value: /runtime-access/claude}
             - {name: KUBECONFIG, value: /opt/data/home/.kube/config}
             - {name: PYTHONPATH, value: /opt/hermes}
-            - {name: HERMES_CLI_LANE_CONCURRENCY, value: "4"}
+            - {name: HERMES_CLI_LANE_CONCURRENCY, value: "2"}
             - {name: HERMES_AUTO_ROUTER_PROFILE, value: agent}
             - {name: PATH, value: /opt/coordinator:/opt/data/tools/bin:/opt/data/home/.local/bin:/opt/hermes/.venv/bin:/usr/local/bin:/usr/bin:/bin}
           securityContext:
@@ -896,7 +897,7 @@ spec:
             - {name: tmp, mountPath: /tmp}
           resources:
             requests: {cpu: 100m, memory: 256Mi}
-            limits: {cpu: "3", memory: 6Gi}
+            limits: {cpu: "2", memory: 6Gi}
         - name: model-steward
           image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107
           imagePullPolicy: IfNotPresent

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(

services/maintenance/apps/metis-configmap.yaml

@@ -15,7 +15,7 @@ data:
   METIS_MAX_DEVICE_BYTES: "1000000000000"
   METIS_NAMESPACE: maintenance
   METIS_REMOTE_POD_TIMEOUT_SEC: "14400"
-  METIS_RUNNER_IMAGE_AMD64: registry.bstein.dev/bstein/metis:0.1.0-295-amd64 # {"$imagepolicy": "maintenance:metis-amd64"}
+  METIS_RUNNER_IMAGE_AMD64: registry.bstein.dev/bstein/metis:0.1.0-296-amd64 # {"$imagepolicy": "maintenance:metis-amd64"}
   METIS_RUNNER_IMAGE_ARM64: registry.bstein.dev/bstein/metis:0.1.0-295-arm64 # {"$imagepolicy": "maintenance:metis-arm64"}
   METIS_HARBOR_REGISTRY: registry.bstein.dev
   METIS_HARBOR_PROJECT: metis

services/maintenance/kustomization.yaml

@@ -11,7 +11,7 @@ resources:
   - networking
 images:
   - name: registry.bstein.dev/bstein/ariadne
-    newTag: 0.1.0-460 # {"$imagepolicy": "maintenance:ariadne:tag"}
+    newTag: 0.1.0-461 # {"$imagepolicy": "maintenance:ariadne:tag"}
   - name: registry.bstein.dev/bstein/metis
     newTag: 0.1.0-295-arm64 # {"$imagepolicy": "maintenance:metis-arm64:tag"}
   - name: registry.bstein.dev/bstein/soteria

services/maintenance/node-ops/metis-sentinel-amd64-daemonset.yaml

@@ -32,7 +32,7 @@ spec:
         kubernetes.io/arch: amd64
       containers:
         - name: metis-sentinel
-          image: registry.bstein.dev/bstein/metis-sentinel:0.1.0-295-amd64 # {"$imagepolicy": "maintenance:metis-sentinel-amd64"}
+          image: registry.bstein.dev/bstein/metis-sentinel:0.1.0-296-amd64 # {"$imagepolicy": "maintenance:metis-sentinel-amd64"}
           imagePullPolicy: Always
           envFrom:
             - configMapRef:

testing/tests/test_hermes_chat_quality.py

@@ -120,6 +120,9 @@ def test_agent_image_completes_parked_kanban_tasks_atomically():
         "Atomically mark running, ready, blocked, or scheduled tasks done"
         in dockerfile
     )
+    assert "kind IN ('created', 'blocked', 'unblocked')" in dockerfile
+    assert "payload.get(\"status\") == \"blocked\"" in dockerfile
+    assert "hermes-kanban-blocked-regression.py" in dockerfile
 
 
 def test_sandbox_shares_only_the_tenant_workspace_without_credentials():

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 == [
         (