hermes: make supervisor terminal emissions idempotent across ticks

Fix reviewer BLOCK: the stateless poll loop re-planned every terminal card each
tick, so SHIP and the two cycle-limit escalations - whose source cards stay in
the done state - re-fired their comment/block on every pass (~2880/day/chain).

- Add a persistent emission Ledger (/opt/data/supervisor/emitted.json): SHIP and
  escalate perform their action only when the (action, target) key is absent,
  then record it, so each fires at most once and survives a pod restart. Spawns
  remain self-limiting via the existing dedup scan.
- Cycle-limit escalations now target the SOURCE review/repair card (not the impl
  root), so the card also leaves the done state and is skipped next tick even if
  block_task round-trips imperfectly.
- Harden the _spawn TypeError fallback: re-run the dedup scan before retrying so
  a post-insert TypeError can never double-insert.
- Add across-ticks idempotency tests (10x supervise_once -> exactly one
  comment/flag/block) plus ledger persistence/corruption and spawn-guard tests.

Both modules stay 100% line+branch, <500 LOC. Stacks on the merge train
(base 5f27e50c).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

services/hermes/scripts/kanban_supervisor.py

@@ -29,6 +29,7 @@ config, re-read every tick like ``kanban.auto_decompose``.
 
 from __future__ import annotations
 
+import json
 import os
 import sys
 import time
@@ -42,6 +43,9 @@ import supervisor_policy as policy
 
 DATA_ROOT = Path(os.environ.get("HERMES_HOME", "/opt/data"))
 CONFIG_PATH = DATA_ROOT / "config.yaml"
+# Persistent record of terminal (ship / escalate) emissions so each fires at
+# most once across the stateless poll loop, and survives a pod restart.
+LEDGER_PATH = DATA_ROOT / "supervisor" / "emitted.json"
 
 DEFAULT_INTERVAL_SECONDS = 30
 DEFAULT_MAX_CYCLES = 5
@@ -114,6 +118,43 @@ def _log(message: str) -> None:
     print(f"kanban-supervisor: {message}", file=sys.stderr, flush=True)
 
 
+class Ledger:
+    """Durable set of emission keys the supervisor itself has already fired.
+
+    The poll loop is stateless and re-plans every terminal card each tick, so a
+    SHIP or a cycle-limit escalation would otherwise re-comment its target on
+    every pass. This ledger is the authoritative read-before-write guard: an
+    emission is performed only when its key is absent, then recorded, so each
+    fires at most once regardless of whether the underlying ``kanban_db`` moves
+    the source card out of its done state.
+    """
+
+    def __init__(self, path: Path):
+        self._path = path
+        self._keys = self._load(path)
+
+    @staticmethod
+    def _load(path: Path) -> set[str]:
+        try:
+            data = json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, ValueError):
+            return set()
+        return {str(key) for key in data} if isinstance(data, list) else set()
+
+    def has(self, key: str) -> bool:
+        return key in self._keys
+
+    def record(self, key: str) -> None:
+        self._keys.add(key)
+        try:
+            self._path.parent.mkdir(parents=True, exist_ok=True)
+            tmp = self._path.with_suffix(".tmp")
+            tmp.write_text(json.dumps(sorted(self._keys)), encoding="utf-8")
+            os.replace(tmp, self._path)
+        except OSError as error:
+            _log(f"could not persist emission ledger: {error}")
+
+
 def _iter_boards(kanban_db: Any) -> list[str]:
     try:
         boards = kanban_db.list_boards(include_archived=False)
@@ -156,28 +197,55 @@ def _escalate(kanban_db: Any, conn: Any, task_id: str, reason: str) -> None:
     _comment(kanban_db, conn, task_id, body)
 
 
+def _already_created(kanban_db: Any, conn: Any, payload: dict[str, Any]) -> bool:
+    """True if a card matching this spawn payload's stamp already exists."""
+    supervisor_stamp = (payload.get("metadata") or {}).get("supervisor") or {}
+    kind = supervisor_stamp.get("kind")
+    root = str(supervisor_stamp.get("root") or "")
+    commit = str(supervisor_stamp.get("head_commit") or "")
+    if not kind or not root:
+        return False
+    try:
+        tasks = list(kanban_db.list_tasks(conn))
+    except Exception as error:  # noqa: BLE001 - if we cannot confirm, do not skip
+        _log(f"could not re-scan before spawn retry: {error}")
+        return False
+    return policy.existing_followup(tasks, kind, root, commit)
+
+
 def _spawn(kanban_db: Any, conn: Any, decision: policy.Decision) -> None:
     payload = dict(decision.payload or {})
     try:
         kanban_db.create_task(conn, **payload)
     except TypeError:
-        # Older runtimes may not accept idempotency_key; the pre-spawn dedup scan
-        # in supervisor_policy still prevents a duplicate, so drop it and retry.
-        payload.pop("idempotency_key", None)
-        kanban_db.create_task(conn, **payload)
+        # create_task may reject idempotency_key on older runtimes. That is
+        # raised at call binding, before any insert, but a post-insert TypeError
+        # is also possible, so re-run the dedup scan first: only retry the create
+        # when no matching card exists, so a partial insert is never doubled.
+        if not _already_created(kanban_db, conn, payload):
+            payload.pop("idempotency_key", None)
+            kanban_db.create_task(conn, **payload)
     _comment(kanban_db, conn, decision.target_id, f"supervisor: {decision.reason}")
 
 
 def apply_decision(
-    kanban_db: Any, conn: Any, decision: policy.Decision
+    kanban_db: Any, conn: Any, decision: policy.Decision, ledger: Ledger
 ) -> bool:
-    """Execute one decision. Returns True when an action was taken."""
+    """Execute one decision. Returns True when an action was taken.
+
+    Spawns are self-limiting across ticks (the created card fails the next dedup
+    scan). SHIP and escalate targets stay in their done state, so they are gated
+    on the persistent ledger and fire at most once per source card.
+    """
     if decision.action == "none":
         return False
     policy.assert_safe(decision)
     if decision.action == "spawn":
         _spawn(kanban_db, conn, decision)
         return True
+    key = f"{decision.action}:{decision.target_id}"
+    if ledger.has(key):
+        return False  # already emitted on a prior tick; never re-emit or re-block
     if decision.action == "ship":
         info = decision.payload or {}
         commit = info.get("commit", "")
@@ -188,15 +256,15 @@ def apply_decision(
             "merges, approves, closes, or clears WIP."
         )
         _mark_ready_for_human(kanban_db, conn, decision.target_id, body)
-        return True
-    if decision.action == "escalate":
+    else:  # escalate
         _escalate(kanban_db, conn, decision.target_id, decision.reason)
-        return True
-    # Unreachable given assert_safe, but fail closed on any future action name.
-    raise ValueError(f"unhandled supervisor action: {decision.action!r}")  # pragma: no cover
+    ledger.record(key)
+    return True
 
 
-def supervise_board(kanban_db: Any, board: str, limits: policy.Limits) -> int:
+def supervise_board(
+    kanban_db: Any, board: str, limits: policy.Limits, ledger: Ledger
+) -> int:
     actions = 0
     with kanban_db.scoped_current_board(board):
         conn = kanban_db.connect(board=board)
@@ -205,7 +273,7 @@ def supervise_board(kanban_db: Any, board: str, limits: policy.Limits) -> int:
             for task in tasks:
                 decision = policy.plan(task, tasks, limits)
                 try:
-                    if apply_decision(kanban_db, conn, decision):
+                    if apply_decision(kanban_db, conn, decision, ledger):
                         actions += 1
                 except Exception as error:  # noqa: BLE001 - one card never stops others
                     _log(f"action failed on board {board!r}: {error}")
@@ -214,12 +282,16 @@ def supervise_board(kanban_db: Any, board: str, limits: policy.Limits) -> int:
     return actions
 
 
-def supervise_once(kanban_db: Any, limits: policy.Limits) -> int:
+def supervise_once(
+    kanban_db: Any, limits: policy.Limits, ledger: Ledger | None = None
+) -> int:
     """One bounded pass over every non-archived board."""
+    if ledger is None:
+        ledger = Ledger(LEDGER_PATH)
     total = 0
     for board in _iter_boards(kanban_db):
         try:
-            total += supervise_board(kanban_db, board, limits)
+            total += supervise_board(kanban_db, board, limits, ledger)
         except Exception as error:  # noqa: BLE001 - isolate a faulty board
             _log(f"temporarily skipping board {board!r}: {error}")
     return total

services/hermes/scripts/supervisor_policy.py

@@ -17,12 +17,11 @@ Safety contract enforced here (audited hard - it drives an autonomous loop):
   never an auto-spawn and never an inferred SHIP.
 * NO DUPLICATE - a review or repair is proposed only when no card already covers
   the same ``(root, head_commit)``, detected across supervisor-stamped *and*
-  externally created cards (e.g. the external codex-shepherd), so it is safe to
-  run concurrently without double-spawning.
+  externally created cards (the codex-shepherd), so it is concurrency-safe.
 * BOUNDED - a per-chain review<->repair cycle limit and a max-concurrent-chains
   ceiling turn runaway loops into human escalations instead of capacity burn.
-* SUBSCRIPTION ONLY - the only spawn is a Kanban card routed through the
-  existing cli-* subscription lanes; no provider/API-key field is ever produced.
+* SUBSCRIPTION ONLY - the only spawn is a Kanban card on the existing cli-*
+  subscription lanes; no provider/API-key field is ever produced.
 """
 
 from __future__ import annotations
@@ -427,14 +426,15 @@ def plan_review(task: Any, tasks: list[Any], limits: Limits) -> Decision:
             root_id,
             {"commit": head_commit, "pr": extract_pr(task, result) or ""},
         )
-    # BLOCK verdict -> spawn a bounded repair, unless one already exists or the
-    # per-chain cycle budget is exhausted.
+    # BLOCK -> spawn a bounded repair, unless one exists or the budget is spent.
     if cycle >= limits.max_cycles:
+        # Escalate the review card itself, not root_id: the impl stays
+        # legitimately done, so re-targeting it re-fires every tick.
         return Decision(
             "escalate",
-            f"repair cycle limit ({limits.max_cycles}) reached for {root_id}; "
-            "escalating instead of another repair",
-            root_id,
+            f"repair cycle limit ({limits.max_cycles}) reached for chain {root_id}; "
+            "escalating this review instead of spawning another repair",
+            review_id,
         )
     if existing_followup(tasks, REPAIR_KIND, root_id, head_commit):
         return Decision("none", "a repair already covers this commit")
@@ -471,11 +471,12 @@ def plan_repair(task: Any, tasks: list[Any], limits: Limits) -> Decision:
         )
     next_cycle = cycle + 1
     if next_cycle > limits.max_cycles:
+        # Escalate the repair card itself, not root_id (see plan_review).
         return Decision(
             "escalate",
-            f"repair cycle limit ({limits.max_cycles}) reached for {root_id}; "
-            "escalating instead of re-review",
-            root_id,
+            f"repair cycle limit ({limits.max_cycles}) reached for chain {root_id}; "
+            "escalating this repair instead of spawning a re-review",
+            repair_id,
         )
     if existing_followup(tasks, REVIEW_KIND, root_id, new_commit):
         return Decision("none", "a re-review already covers this commit")

testing/tests/test_hermes_kanban_supervisor.py

@@ -1,8 +1,6 @@
 """Behavioral tests for the supervisor I/O shell: config, board iteration,
-decision application, the poll loop, the auto_supervise gate, and the deployment
-wiring. Real logic is exercised against a stubbed ``hermes_cli.kanban_db`` built
-the way the existing cli-lane tests build theirs, plus the confirmed
-NULL->20 goal-turn fallback fix in cli_lane_execution.
+decision application, across-ticks idempotency, the poll loop, the
+auto_supervise gate, deployment wiring, and the NULL->20 goal-turn fix.
 """
 
 from __future__ import annotations
@@ -10,6 +8,7 @@ from __future__ import annotations
 from contextlib import nullcontext
 from types import SimpleNamespace
 
+import pytest
 import yaml
 
 from testing.tests.test_hermes_cli_support import HERMES, _agent_deployment, _load
@@ -18,6 +17,12 @@ supervisor = _load("kanban_supervisor")
 policy = supervisor.policy
 
 
+@pytest.fixture(autouse=True)
+def _isolate_ledger(tmp_path, monkeypatch):
+    """Keep each test's emission ledger on an isolated, writable path."""
+    monkeypatch.setattr(supervisor, "LEDGER_PATH", tmp_path / "emitted.json")
+
+
 class RecordingDb:
     """Minimal in-memory kanban_db stub capturing every mutating call."""
 
@@ -212,9 +217,10 @@ def test_spawn_retries_without_idempotency_key_on_typeerror():
 def test_apply_decision_ignores_none_and_rejects_unsafe():
     db = RecordingDb([])
     conn = SimpleNamespace(close=lambda: None)
-    assert supervisor.apply_decision(db, conn, policy.Decision("none")) is False
+    ledger = supervisor.Ledger(supervisor.LEDGER_PATH)
+    assert supervisor.apply_decision(db, conn, policy.Decision("none"), ledger) is False
     try:
-        supervisor.apply_decision(db, conn, policy.Decision("teleport"))
+        supervisor.apply_decision(db, conn, policy.Decision("teleport"), ledger)
         raise AssertionError("expected rejection")
     except ValueError:
         pass
@@ -253,6 +259,128 @@ def test_comment_and_block_failures_are_swallowed(capsys):
     assert "could not block impl" in err and "could not comment on impl" in err
 
 
+# --- across-ticks idempotency (the re-emission bug) -----------------------
+
+
+def _ship_review(commit="c1"):
+    return _task(
+        id="rev",
+        metadata={"supervisor": {"kind": "review", "root": "impl", "parent": "impl", "head_commit": commit, "cycle": 1}},
+        result={"verdict": "SHIP", "summary": "clean"},
+    )
+
+
+def _cycle_exhausted_review():
+    return _task(
+        id="rev",
+        metadata={"supervisor": {"kind": "review", "root": "impl", "parent": "impl", "head_commit": "c1", "cycle": 5}},
+        result={"verdict": "BLOCK", "findings": ["still broken"]},
+    )
+
+
+def _cycle_exhausted_repair():
+    return _task(
+        id="rep",
+        metadata={"supervisor": {"kind": "repair", "root": "impl", "parent": "rev", "head_commit": "c1", "cycle": 5}},
+        result={"head_commit": "c2"},
+    )
+
+
+def test_ship_marks_ready_exactly_once_across_ten_ticks():
+    review = _ship_review()
+    db = RecordingDb([review])
+    for _ in range(10):
+        supervisor.supervise_once(db, policy.Limits())
+    ready_comments = [c for c in db.comments if "READY FOR HUMAN MERGE" in c[2]]
+    assert len(ready_comments) == 1
+    assert len(db.metadata_sets) == 1
+    assert db.created == [] and db.blocked == []
+
+
+def test_review_cycle_limit_escalates_exactly_once_across_ten_ticks():
+    # block_task does not move the card out of done here, so the ledger - not a
+    # status transition - must be what stops re-emission.
+    db = RecordingDb([_cycle_exhausted_review()])
+    for _ in range(10):
+        supervisor.supervise_once(db, policy.Limits())
+    assert len(db.blocked) == 1 and db.blocked[0][0] == "rev"
+    assert len([c for c in db.comments if c[0] == "rev"]) == 1
+    assert db.created == []
+
+
+def test_repair_cycle_limit_escalates_exactly_once_across_ten_ticks():
+    db = RecordingDb([_cycle_exhausted_repair()])
+    for _ in range(10):
+        supervisor.supervise_once(db, policy.Limits())
+    assert len(db.blocked) == 1 and db.blocked[0][0] == "rep"
+    assert len([c for c in db.comments if c[0] == "rep"]) == 1
+
+
+def test_supervise_once_accepts_an_injected_ledger():
+    review = _ship_review()
+    db = RecordingDb([review])
+    ledger = supervisor.Ledger(supervisor.LEDGER_PATH)
+    supervisor.supervise_once(db, policy.Limits(), ledger)
+    supervisor.supervise_once(db, policy.Limits(), ledger)
+    assert len([c for c in db.comments if "READY FOR HUMAN MERGE" in c[2]]) == 1
+
+
+def test_ledger_survives_reload_from_disk():
+    ledger = supervisor.Ledger(supervisor.LEDGER_PATH)
+    assert ledger.has("ship:impl") is False
+    ledger.record("ship:impl")
+    reloaded = supervisor.Ledger(supervisor.LEDGER_PATH)
+    assert reloaded.has("ship:impl") is True
+
+
+def test_ledger_tolerates_corrupt_or_unwritable_state(monkeypatch, tmp_path, capsys):
+    corrupt = tmp_path / "corrupt.json"
+    corrupt.write_text("{not json", encoding="utf-8")
+    assert supervisor.Ledger(corrupt).has("x") is False
+    # A record failure is swallowed and logged, never raised into the tick.
+    monkeypatch.setattr(supervisor, "LEDGER_PATH", tmp_path / "missing" / "x")
+    bad = supervisor.Ledger(supervisor.LEDGER_PATH)
+    monkeypatch.setattr(supervisor.os, "replace", lambda *_a: (_ for _ in ()).throw(OSError("ro")))
+    bad.record("k")
+    assert "could not persist emission ledger" in capsys.readouterr().err
+
+
+def test_spawn_typeerror_does_not_double_insert_when_row_already_exists():
+    review = _task(
+        id="rev",
+        metadata={"supervisor": {"kind": "review", "root": "impl", "parent": "impl", "head_commit": "c1", "cycle": 1}},
+        status="ready",
+    )
+
+    class InsertThenTypeErrorDb(RecordingDb):
+        def create_task(self, _conn, **kwargs):
+            # Simulate an insert that lands but still raises TypeError.
+            self.created.append(kwargs)
+            self._tasks.append(review)
+            raise TypeError("post-insert boom")
+
+    impl = _task(
+        id="impl",
+        result={"changed_files": ["a"], "head_commit": "c1", "pull_request": "pr"},
+    )
+    db = InsertThenTypeErrorDb([impl])
+    supervisor.supervise_once(db, policy.Limits())
+    assert len(db.created) == 1  # retry suppressed by the re-scan guard
+
+
+def test_already_created_returns_false_without_stamp_or_on_scan_error():
+    db = RecordingDb([])
+    conn = SimpleNamespace(close=lambda: None)
+    assert supervisor._already_created(db, conn, {"metadata": {}}) is False
+
+    class BrokenScanDb(RecordingDb):
+        def list_tasks(self, _conn):
+            raise RuntimeError("scan down")
+
+    payload = {"metadata": {"supervisor": {"kind": "review", "root": "impl", "head_commit": "c1"}}}
+    assert supervisor._already_created(BrokenScanDb([]), conn, payload) is False
+
+
 # --- fault isolation ------------------------------------------------------
 
 

testing/tests/test_hermes_kanban_supervisor_policy.py

@@ -265,14 +265,17 @@ def test_review_ambiguous_verdict_fails_closed():
     assert policy.plan(review, [review], LIMITS).action == "escalate"
 
 
-def test_review_block_at_cycle_limit_escalates_instead_of_repairing():
+def test_review_block_at_cycle_limit_escalates_the_source_review_not_impl():
     review = task(
         id="rev",
         metadata=review_stamp("impl", "c1", LIMITS.max_cycles),
         result={"verdict": "BLOCK", "findings": ["x"]},
     )
     decision = policy.plan(review, [review], LIMITS)
-    assert decision.action == "escalate" and decision.target_id == "impl"
+    # Targets the review card (source), not root_id, so it leaves the done state
+    # and is not re-planned into the same escalation next tick.
+    assert decision.action == "escalate" and decision.target_id == "rev"
+    assert "impl" in decision.reason
 
 
 def test_review_block_skips_when_repair_already_exists():
@@ -327,14 +330,15 @@ def test_repair_producing_same_commit_fails_closed():
     assert policy.plan(repair, [repair], LIMITS).action == "escalate"
 
 
-def test_repair_at_cycle_limit_escalates_instead_of_re_reviewing():
+def test_repair_at_cycle_limit_escalates_the_source_repair_not_impl():
     repair = task(
         id="rep",
         metadata=repair_stamp("impl", "c1", LIMITS.max_cycles),
         result={"head_commit": "c2"},
     )
     decision = policy.plan(repair, [repair], LIMITS)
-    assert decision.action == "escalate" and decision.target_id == "impl"
+    assert decision.action == "escalate" and decision.target_id == "rep"
+    assert "impl" in decision.reason
 
 
 def test_repair_skips_when_re_review_already_exists():