diff --git a/services/hermes/scripts/kanban_supervisor.py b/services/hermes/scripts/kanban_supervisor.py index 631ea76e..6d02f30c 100644 --- a/services/hermes/scripts/kanban_supervisor.py +++ b/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 diff --git a/services/hermes/scripts/supervisor_policy.py b/services/hermes/scripts/supervisor_policy.py index 800b3993..cc19c986 100644 --- a/services/hermes/scripts/supervisor_policy.py +++ b/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") diff --git a/testing/tests/test_hermes_kanban_supervisor.py b/testing/tests/test_hermes_kanban_supervisor.py index 71f74a55..b1ddde09 100644 --- a/testing/tests/test_hermes_kanban_supervisor.py +++ b/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 ------------------------------------------------------ diff --git a/testing/tests/test_hermes_kanban_supervisor_policy.py b/testing/tests/test_hermes_kanban_supervisor_policy.py index 1dde9e20..79973ac4 100644 --- a/testing/tests/test_hermes_kanban_supervisor_policy.py +++ b/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():