diff --git a/services/hermes/agent-configmap.yaml b/services/hermes/agent-configmap.yaml index 91da82f6..d63951c0 100644 --- a/services/hermes/agent-configmap.yaml +++ b/services/hermes/agent-configmap.yaml @@ -87,6 +87,18 @@ data: max_in_progress_per_profile: 1 auto_decompose: true auto_decompose_per_tick: 2 + # In-pod autonomous supervisor (kanban_supervisor.py). Default false so it + # stays inert until the external codex-shepherd is retired; re-read each + # tick like auto_decompose. When true it drives the cross-card + # implement->review->repair->re-review chain by creating Kanban cards + # only (subscription lanes), never merging/approving. Bounded by the + # cycle and concurrent-chain ceilings below. + auto_supervise: false + supervise_interval_seconds: 30 + supervise_max_cycles: 5 + supervise_max_chains: 20 + supervise_review_assignee: cli-claude-xhigh + supervise_repair_assignee: cli-auto dispatch_stale_timeout_seconds: 14400 # Steady-state quota-aware routing for the direct CLI lane: below this # remaining-percent a provider stops receiving NEW cli-auto work while diff --git a/services/hermes/agent-deployment.yaml b/services/hermes/agent-deployment.yaml index 93508432..7ae26434 100644 --- a/services/hermes/agent-deployment.yaml +++ b/services/hermes/agent-deployment.yaml @@ -941,6 +941,37 @@ spec: resources: requests: {cpu: 25m, memory: 64Mi} limits: {cpu: 250m, memory: 512Mi} + - name: kanban-supervisor + image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107 + imagePullPolicy: IfNotPresent + # Autonomous cross-card review->repair->re-review driver. Reads/writes + # only the local Kanban DB under /opt/data; creates Kanban cards that + # route through the existing subscription lanes. No runtime-access + # mount and no provider client: it holds no metered/API-key path. Inert + # until kanban.auto_supervise is set true in the deployed config. + command: [/opt/hermes/.venv/bin/python, /opt/coordinator/kanban_supervisor.py] + env: + - {name: HERMES_HOME, value: /opt/data} + - {name: HOME, value: /opt/data/home} + - {name: PYTHONPATH, value: /opt/hermes} + - {name: PYTHONDONTWRITEBYTECODE, value: "1"} + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: [ALL] + readOnlyRootFilesystem: true + runAsNonRoot: true + runAsUser: 10000 + runAsGroup: 10000 + seccompProfile: + type: RuntimeDefault + volumeMounts: + - {name: home, mountPath: /opt/data} + - {name: coordinator, mountPath: /opt/coordinator, readOnly: true} + - {name: auth-patch, mountPath: /opt/hermes/hermes_cli/auth.py, subPath: auth.py} + resources: + requests: {cpu: 10m, memory: 64Mi} + limits: {cpu: 250m, memory: 256Mi} - name: credential-sync image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107 imagePullPolicy: IfNotPresent diff --git a/services/hermes/kustomization.yaml b/services/hermes/kustomization.yaml index 430ad826..7abeb694 100644 --- a/services/hermes/kustomization.yaml +++ b/services/hermes/kustomization.yaml @@ -131,6 +131,8 @@ configMapGenerator: - jenkins_build_evidence.py=scripts/jenkins_build_evidence.py - jenkins_image_build_trigger.py=scripts/jenkins_image_build_trigger.py - kanban_status_recovery.py=scripts/kanban_status_recovery.py + - kanban_supervisor.py=scripts/kanban_supervisor.py + - supervisor_policy.py=scripts/supervisor_policy.py - migrate_herdr_state.py=scripts/migrate_herdr_state.py - migrate_api_session_lineage.py=scripts/migrate_api_session_lineage.py - migrate_telegram_api_sessions.py=scripts/migrate_telegram_api_sessions.py diff --git a/services/hermes/scripts/cli_lane_execution.py b/services/hermes/scripts/cli_lane_execution.py index 8cf61981..f89ded06 100644 --- a/services/hermes/scripts/cli_lane_execution.py +++ b/services/hermes/scripts/cli_lane_execution.py @@ -166,7 +166,7 @@ def execute_claim(board: str, task_id: str) -> None: _task_value(task, "max_runtime_seconds", 0) or DEFAULT_MAX_RUNTIME ) goal_mode = bool(_task_value(task, "goal_mode", False)) - goal_max_turns = max(1, int(_task_value(task, "goal_max_turns", 1) or 1)) + goal_max_turns = max(1, int(_task_value(task, "goal_max_turns", 20) or 20)) goal_turn = max(1, int(state.get("goal_turn", 0) or 0) + 1) deadline = time.monotonic() + max_runtime while True: diff --git a/services/hermes/scripts/kanban_supervisor.py b/services/hermes/scripts/kanban_supervisor.py new file mode 100644 index 00000000..631ea76e --- /dev/null +++ b/services/hermes/scripts/kanban_supervisor.py @@ -0,0 +1,257 @@ +#!/usr/bin/env python3 +"""In-pod autonomous supervisor that sustains multi-hour unattended work. + +The deployed Hermes agent already decomposes objectives and iterates a within- +card goal loop, but nothing in-pod spawns the *cross-card* review->repair-> +re-review follow-ups after a card finishes; today that chain only exists as an +external Claude Code "codex-shepherd" session on an operator's workstation, so +unattended runs stall once the first implementation card completes. This +supervisor closes that gap from inside the pod. + +It runs a bounded poll loop that reads board state through ``hermes_cli.kanban_db`` +and, for each terminal card, applies exactly one follow-up: + +* a done implementation with a produced PR/commit and no existing review -> + create a review card (routed to a review assignee via the normal cli lane); +* a review card that returned SHIP -> mark the implementation ready for a human + to merge (a comment/flag; the supervisor never merges, approves, or clears WIP); +* a review card that returned BLOCK -> create a bounded repair card; +* a repair card that produced a new commit -> create the re-review card. + +All decision logic lives in :mod:`supervisor_policy` (pure, unit tested). This +module is only the I/O shell: config, board iteration, and turning a +:class:`supervisor_policy.Decision` into ``create_task``/``block_task``/ +``add_comment`` calls. It imports no provider client and holds no metered path; +the sole spawn is a Kanban card that routes through the existing subscription +lanes. It is inert until ``kanban.auto_supervise`` is set true in the deployed +config, re-read every tick like ``kanban.auto_decompose``. +""" + +from __future__ import annotations + +import os +import sys +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Callable + +import yaml + +import supervisor_policy as policy + +DATA_ROOT = Path(os.environ.get("HERMES_HOME", "/opt/data")) +CONFIG_PATH = DATA_ROOT / "config.yaml" + +DEFAULT_INTERVAL_SECONDS = 30 +DEFAULT_MAX_CYCLES = 5 +DEFAULT_MAX_CHAINS = 20 +DEFAULT_REVIEW_ASSIGNEE = "cli-claude-xhigh" +DEFAULT_REPAIR_ASSIGNEE = "cli-auto" + + +@dataclass(frozen=True) +class Settings: + """Per-tick configuration, re-read from the deployed config each pass.""" + + enabled: bool + interval: int + limits: policy.Limits + + +def _kanban_config() -> dict[str, Any]: + try: + document = yaml.safe_load(CONFIG_PATH.read_text(encoding="utf-8")) + except (OSError, yaml.YAMLError): + return {} + kanban = document.get("kanban") if isinstance(document, dict) else None + return kanban if isinstance(kanban, dict) else {} + + +def _bool(value: Any, default: bool) -> bool: + return value if isinstance(value, bool) else default + + +def _positive_int(value: Any, default: int) -> int: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return default + parsed = int(value) + return parsed if parsed > 0 else default + + +def _text(value: Any, default: str) -> str: + return value if isinstance(value, str) and value.strip() else default + + +def load_settings() -> Settings: + """Read the ``kanban`` config block; default to inert and conservative.""" + cfg = _kanban_config() + return Settings( + enabled=_bool(cfg.get("auto_supervise"), False), + interval=_positive_int( + cfg.get("supervise_interval_seconds"), DEFAULT_INTERVAL_SECONDS + ), + limits=policy.Limits( + max_cycles=_positive_int(cfg.get("supervise_max_cycles"), DEFAULT_MAX_CYCLES), + max_chains=_positive_int(cfg.get("supervise_max_chains"), DEFAULT_MAX_CHAINS), + review_assignee=_text( + cfg.get("supervise_review_assignee"), DEFAULT_REVIEW_ASSIGNEE + ), + repair_assignee=_text( + cfg.get("supervise_repair_assignee"), DEFAULT_REPAIR_ASSIGNEE + ), + ), + ) + + +def _board_slug(board: Any) -> str: + if isinstance(board, dict): + return str(board.get("slug") or board.get("id") or "") + return str(getattr(board, "slug", None) or getattr(board, "id", None) or board or "") + + +def _log(message: str) -> None: + print(f"kanban-supervisor: {message}", file=sys.stderr, flush=True) + + +def _iter_boards(kanban_db: Any) -> list[str]: + try: + boards = kanban_db.list_boards(include_archived=False) + except Exception as error: # noqa: BLE001 - one bad registry never stops the tick + _log(f"could not list boards: {error}") + return [] + slugs = [_board_slug(raw) for raw in boards] + return [slug for slug in slugs if slug] + + +def _comment(kanban_db: Any, conn: Any, task_id: str, body: str) -> None: + try: + kanban_db.add_comment(conn, task_id, policy.SUPERVISOR_AUTHOR, body) + except Exception as error: # noqa: BLE001 - a comment is best-effort signalling + _log(f"could not comment on {task_id}: {error}") + + +def _mark_ready_for_human(kanban_db: Any, conn: Any, task_id: str, body: str) -> None: + """Flag the implementation as human-mergeable. Never merges or clears WIP.""" + setter = getattr(kanban_db, "set_task_metadata", None) or getattr( + kanban_db, "update_task_metadata", None + ) + if callable(setter): + try: + setter(conn, task_id, {"supervisor_ready_for_human_merge": True}) + except Exception as error: # noqa: BLE001 - the comment is the durable flag + _log(f"could not set ready flag on {task_id}: {error}") + _comment(kanban_db, conn, task_id, body) + + +def _escalate(kanban_db: Any, conn: Any, task_id: str, reason: str) -> None: + body = ( + f"supervisor fail-closed escalation: {reason}. Human attention required; " + "no follow-up card was auto-created." + ) + try: + kanban_db.block_task(conn, task_id, reason=body, kind="supervisor") + except Exception as error: # noqa: BLE001 - still record the human-visible comment + _log(f"could not block {task_id}: {error}") + _comment(kanban_db, conn, task_id, body) + + +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) + _comment(kanban_db, conn, decision.target_id, f"supervisor: {decision.reason}") + + +def apply_decision( + kanban_db: Any, conn: Any, decision: policy.Decision +) -> bool: + """Execute one decision. Returns True when an action was taken.""" + if decision.action == "none": + return False + policy.assert_safe(decision) + if decision.action == "spawn": + _spawn(kanban_db, conn, decision) + return True + if decision.action == "ship": + info = decision.payload or {} + commit = info.get("commit", "") + pr = info.get("pr", "") or "branch on record" + body = ( + f"supervisor: review returned SHIP for commit {commit} ({pr}). " + "READY FOR HUMAN MERGE - a human must merge; the supervisor never " + "merges, approves, closes, or clears WIP." + ) + _mark_ready_for_human(kanban_db, conn, decision.target_id, body) + return True + if decision.action == "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 + + +def supervise_board(kanban_db: Any, board: str, limits: policy.Limits) -> int: + actions = 0 + with kanban_db.scoped_current_board(board): + conn = kanban_db.connect(board=board) + try: + tasks = list(kanban_db.list_tasks(conn)) + for task in tasks: + decision = policy.plan(task, tasks, limits) + try: + if apply_decision(kanban_db, conn, decision): + actions += 1 + except Exception as error: # noqa: BLE001 - one card never stops others + _log(f"action failed on board {board!r}: {error}") + finally: + conn.close() + return actions + + +def supervise_once(kanban_db: Any, limits: policy.Limits) -> int: + """One bounded pass over every non-archived board.""" + total = 0 + for board in _iter_boards(kanban_db): + try: + total += supervise_board(kanban_db, board, limits) + except Exception as error: # noqa: BLE001 - isolate a faulty board + _log(f"temporarily skipping board {board!r}: {error}") + return total + + +def run_forever( + kanban_db: Any, + *, + sleep: Callable[[float], None] = time.sleep, + load: Callable[[], Settings] = load_settings, + max_ticks: int | None = None, +) -> int: + """Poll loop. Inert while ``auto_supervise`` is false; re-reads config a tick.""" + ticks = 0 + while max_ticks is None or ticks < max_ticks: + settings = load() + if settings.enabled: + try: + supervise_once(kanban_db, settings.limits) + except Exception as error: # noqa: BLE001 - never let one tick kill the loop + _log(f"tick failed: {error}") + sleep(settings.interval) + ticks += 1 + return ticks + + +def main(argv: list[str] | None = None) -> int: + from hermes_cli import kanban_db + + run_forever(kanban_db) + return 0 + + +if __name__ == "__main__": # pragma: no cover - process entry point + sys.exit(main(sys.argv[1:])) diff --git a/services/hermes/scripts/supervisor_policy.py b/services/hermes/scripts/supervisor_policy.py new file mode 100644 index 00000000..800b3993 --- /dev/null +++ b/services/hermes/scripts/supervisor_policy.py @@ -0,0 +1,498 @@ +#!/usr/bin/env python3 +"""Pure decision logic for the in-pod autonomous Kanban supervisor. + +This module owns only *pure* policy: given one candidate task and a snapshot of +its board, it returns the single follow-up action the supervisor should take, a +fail-closed escalation, or nothing. It performs no I/O and imports no provider +client, so the whole cross-card implement->review->repair->re-review state +machine is unit testable and can never, by construction, emit a metered/model +call or a merge/approve/close/deploy. Every side effect (create_task, +block_task, add_comment) lives in :mod:`kanban_supervisor`. + +Safety contract enforced here (audited hard - it drives an autonomous loop): + +* FAIL CLOSED - an unparseable result, an ambiguous SHIP/BLOCK verdict, a done + implementation whose produced commit/PR cannot be identified, or a repair that + produced no new commit yields an ``escalate`` decision (human attention), + 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. +* 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. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from typing import Any + +import cli_lane_goal + +SUPERVISOR_AUTHOR = "hermes-supervisor" +STAMP_KEY = "supervisor" +REVIEW_KIND = "review" +REPAIR_KIND = "repair" +DONE_STATUSES = frozenset({"done"}) +# A supervised card whose status is one of these is no longer an in-flight chain +# link (done = finished, blocked = already escalated to a human). +INACTIVE_STATUSES = frozenset({"done", "blocked"}) + +COMMIT_KEYS = ("head_commit", "commit_sha", "commit", "sha", "revision") +PR_KEYS = ("pull_request", "pr_url", "pr", "merge_request") +BRANCH_KEYS = ("branch", "branch_name", "head_branch") + +SAFE_ACTIONS = frozenset({"none", "spawn", "ship", "escalate"}) +# Keys that would signal a privileged/metered action; a payload must never carry +# one. Enforced by :func:`assert_safe` before any card is written. +FORBIDDEN_PAYLOAD_KEYS = frozenset( + {"merge", "approve", "close", "deploy", "provider", "model", "api_key", "metered"} +) + + +@dataclass(frozen=True) +class Limits: + """Bounds and routing that gate every autonomous decision.""" + + max_cycles: int = 5 + max_chains: int = 20 + review_assignee: str = "cli-claude-xhigh" + repair_assignee: str = "cli-auto" + + +@dataclass(frozen=True) +class Decision: + """One resolved supervisor intent. ``action`` is always in SAFE_ACTIONS.""" + + action: str + reason: str = "" + target_id: str = "" + payload: dict[str, Any] | None = None + + +def assert_safe(decision: Decision) -> Decision: + """Reject any decision outside the allow-listed, non-metered action set.""" + if decision.action not in SAFE_ACTIONS: + raise ValueError(f"unsafe supervisor action: {decision.action!r}") + payload = decision.payload or {} + for key in FORBIDDEN_PAYLOAD_KEYS: + if key in payload: + raise ValueError(f"supervisor payload carries forbidden key {key!r}") + return decision + + +def field(task: Any, name: str, default: Any = None) -> Any: + if isinstance(task, dict): + return task.get(name, default) + return getattr(task, name, default) + + +def _as_dict(value: Any) -> dict[str, Any] | None: + if isinstance(value, dict): + return value + if isinstance(value, str): + try: + parsed = json.loads(value) + except (ValueError, TypeError): + return None + return parsed if isinstance(parsed, dict) else None + return None + + +def task_id(task: Any) -> str: + return str(field(task, "id", "") or "") + + +def metadata(task: Any) -> dict[str, Any]: + return _as_dict(field(task, "metadata")) or {} + + +def stamp(task: Any) -> dict[str, Any]: + value = metadata(task).get(STAMP_KEY) + return value if isinstance(value, dict) else {} + + +def supervised_kind(task: Any) -> str | None: + kind = stamp(task).get("kind") + return kind if kind in (REVIEW_KIND, REPAIR_KIND) else None + + +def is_done(task: Any) -> bool: + return str(field(task, "status", "") or "") in DONE_STATUSES + + +def objective(task: Any) -> str: + title = str(field(task, "title", "") or "") + body = str(field(task, "body", "") or "") + return f"{title}\n\n{body}".strip() + + +def parents(task: Any) -> list[str]: + raw = field(task, "parents") + if raw is None: + raw = field(task, "task_links") + if isinstance(raw, dict): + raw = raw.get("parents") + ids: list[str] = [] + if isinstance(raw, (list, tuple, set)): + for item in raw: + if isinstance(item, dict): + ident = item.get("id") or item.get("parent") or item.get("task_id") + else: + ident = item + if ident: + ids.append(str(ident)) + return ids + + +def parse_result(task: Any) -> tuple[dict[str, Any] | None, str | None]: + """Parse the terminal result, failing closed on anything non-object.""" + raw = field(task, "result") + if raw is None or raw == "" or raw == {}: + return None, "task carries no result to interpret" + parsed = _as_dict(raw) + if parsed is None: + return None, "task result is not a JSON object" + return parsed, None + + +def _first_key(source: Any, keys: tuple[str, ...]) -> str | None: + if not isinstance(source, dict): + return None + for key in keys: + value = source.get(key) + if isinstance(value, str) and value.strip(): + return value.strip() + return None + + +def _result_sources(task: Any, result: dict[str, Any] | None) -> list[dict[str, Any]]: + sources = [metadata(task)] + if isinstance(result, dict): + sources.append(result) + nested = _as_dict(result.get("metadata")) + if nested is not None: + sources.append(nested) + return sources + + +def extract_commit(task: Any, result: dict[str, Any] | None) -> str | None: + for source in _result_sources(task, result): + found = _first_key(source, COMMIT_KEYS) + if found: + return found + return None + + +def extract_pr(task: Any, result: dict[str, Any] | None) -> str | None: + for source in _result_sources(task, result): + found = _first_key(source, PR_KEYS) or _first_key(source, BRANCH_KEYS) + if found: + return found + return None + + +def _has_changes(result: dict[str, Any] | None) -> bool: + changed = (result or {}).get("changed_files") + return isinstance(changed, (list, tuple)) and len(changed) > 0 + + +def _strings(value: Any) -> list[str]: + if not isinstance(value, (list, tuple)): + return [] + return [str(item) for item in value if isinstance(item, str) and item.strip()] + + +def _int(value: Any, default: int) -> int: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return default + return int(value) + + +def _haystack(task: Any) -> str: + return "\n".join( + ( + str(field(task, "title", "") or ""), + str(field(task, "body", "") or ""), + json.dumps(metadata(task), sort_keys=True, default=str), + ) + ) + + +def _references(task: Any, ident: str) -> bool: + if not ident: + return False + if ident in parents(task): + return True + return ident in _haystack(task) + + +def _mentions_commit(task: Any, head_commit: str) -> bool: + return bool(head_commit) and head_commit in _haystack(task) + + +def _looks_like_review(task: Any) -> bool: + if metadata(task).get("task_role") == cli_lane_goal.REVIEW_ROLE: + return True + role, _ = cli_lane_goal.task_role(objective(task)) + if role == cli_lane_goal.REVIEW_ROLE: + return True + return "review" in str(field(task, "title", "") or "").lower() + + +def _looks_like_repair(task: Any) -> bool: + if metadata(task).get("task_role") == cli_lane_goal.IMPLEMENTATION_ROLE: + return True + title = str(field(task, "title", "") or "").lower() + return "repair" in title or "fix" in title + + +def existing_followup( + tasks: list[Any], kind: str, root_id: str, head_commit: str +) -> bool: + """True if any card already covers ``(root_id, head_commit)`` for ``kind``. + + Matches both supervisor-stamped cards and externally created ones (the + external shepherd), so concurrent operation never double-spawns. + """ + looks_like = _looks_like_review if kind == REVIEW_KIND else _looks_like_repair + for task in tasks: + st = stamp(task) + if ( + st.get("kind") == kind + and str(st.get("root") or "") == root_id + and str(st.get("head_commit") or "") == head_commit + ): + return True + if ( + _references(task, root_id) + and _mentions_commit(task, head_commit) + and looks_like(task) + ): + return True + return False + + +def active_chain_count(tasks: list[Any]) -> int: + """Distinct chains with an in-flight (not done/blocked) supervised card.""" + roots: set[str] = set() + for task in tasks: + if supervised_kind(task) is None: + continue + if str(field(task, "status", "") or "") in INACTIVE_STATUSES: + continue + root = str(stamp(task).get("root") or "") + if root: + roots.add(root) + return len(roots) + + +def _stamp_meta(kind: str, root: str, source: str, head_commit: str, cycle: int) -> dict: + return { + STAMP_KEY: { + "kind": kind, + "root": root, + "parent": source, + "head_commit": head_commit, + "cycle": cycle, + } + } + + +def _idempotency(kind: str, root: str, head_commit: str, cycle: int) -> str: + return f"supervisor:{kind}:{root}:{head_commit}:{cycle}" + + +def _findings_block(findings: list[str]) -> str: + if not findings: + return "" + lines = "\n".join(f"- {item}" for item in findings) + return f"\nReview findings to address:\n{lines}\n" + + +def _build_review( + root_id: str, + source_id: str, + head_commit: str, + cycle: int, + pr_ref: str, + findings: list[str], + assignee: str, +) -> dict[str, Any]: + meta = _stamp_meta(REVIEW_KIND, root_id, source_id, head_commit, cycle) + meta["task_role"] = cli_lane_goal.REVIEW_ROLE + body = ( + "Hermes-Task-Role: review\n\n" + f"Read-only review of commit {head_commit}" + f"{f' on {pr_ref}' if pr_ref else ''}. Do not modify the implementation; " + "make no code changes. Report a single SHIP or BLOCK verdict with " + "evidence." + _findings_block(findings) + ) + linked = [root_id] if source_id == root_id else [root_id, source_id] + return { + "title": f"Review commit {head_commit[:12]} (cycle {cycle})", + "body": body, + "assignee": assignee, + "parents": linked, + "metadata": meta, + "idempotency_key": _idempotency(REVIEW_KIND, root_id, head_commit, cycle), + } + + +def _build_repair( + root_id: str, + review_id: str, + head_commit: str, + cycle: int, + findings: list[str], + assignee: str, +) -> dict[str, Any]: + meta = _stamp_meta(REPAIR_KIND, root_id, review_id, head_commit, cycle) + meta["task_role"] = cli_lane_goal.IMPLEMENTATION_ROLE + body = ( + "Hermes-Task-Role: repair\n\n" + f"Repair the implementation reviewed at commit {head_commit} per the " + "review findings below, then commit and push the fix." + + _findings_block(findings) + ) + return { + "title": f"Repair review findings for {head_commit[:12]} (cycle {cycle})", + "body": body, + "assignee": assignee, + "parents": [root_id, review_id], + "metadata": meta, + "idempotency_key": _idempotency(REPAIR_KIND, root_id, head_commit, cycle), + } + + +def plan_implementation(task: Any, tasks: list[Any], limits: Limits) -> Decision: + tid = task_id(task) + result, error = parse_result(task) + role, _ = cli_lane_goal.task_role(objective(task), result or {}) + if role != cli_lane_goal.IMPLEMENTATION_ROLE: + # A done review-shaped card the supervisor did not create belongs to the + # external shepherd's chain; never drive it from the implementation side. + return Decision("none", "not a supervisor-owned implementation card") + if error is not None: + return Decision("escalate", f"implementation result unparseable: {error}", tid) + commit = extract_commit(task, result) + pr = extract_pr(task, result) + if commit is None: + if _has_changes(result): + return Decision( + "escalate", + "done implementation changed files but exposes no head commit to review", + tid, + ) + return Decision("none", "implementation produced no commit to review") + if pr is None: + return Decision( + "escalate", + "implementation head commit present but no branch/PR reference to review", + tid, + ) + if existing_followup(tasks, REVIEW_KIND, tid, commit): + return Decision("none", "a review already covers this commit") + if active_chain_count(tasks) >= limits.max_chains: + return Decision("none", "supervised-chain ceiling reached; deferring new review") + payload = _build_review(tid, tid, commit, 1, pr, [], limits.review_assignee) + return Decision("spawn", f"open review for {commit}", tid, payload) + + +def plan_review(task: Any, tasks: list[Any], limits: Limits) -> Decision: + st = stamp(task) + review_id = task_id(task) + root_id = str(st.get("root") or "") + head_commit = str(st.get("head_commit") or "") + cycle = _int(st.get("cycle"), 0) + if not root_id or not head_commit or cycle <= 0: + return Decision( + "escalate", "review card carries an incomplete supervisor stamp", review_id + ) + result, error = parse_result(task) + if error is not None: + return Decision("escalate", f"review result unparseable: {error}", review_id) + verdict, problem = cli_lane_goal.review_verdict(result) + if problem or verdict is None: + detail = problem or "no explicit SHIP or BLOCK verdict" + return Decision("escalate", f"review verdict ambiguous: {detail}", review_id) + if verdict == "SHIP": + return Decision( + "ship", + f"review shipped {head_commit}", + 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. + if cycle >= limits.max_cycles: + return Decision( + "escalate", + f"repair cycle limit ({limits.max_cycles}) reached for {root_id}; " + "escalating instead of another repair", + root_id, + ) + if existing_followup(tasks, REPAIR_KIND, root_id, head_commit): + return Decision("none", "a repair already covers this commit") + findings = _strings(result.get("findings")) + payload = _build_repair( + root_id, review_id, head_commit, cycle, findings, limits.repair_assignee + ) + return Decision("spawn", f"open repair for {head_commit}", review_id, payload) + + +def plan_repair(task: Any, tasks: list[Any], limits: Limits) -> Decision: + st = stamp(task) + repair_id = task_id(task) + root_id = str(st.get("root") or "") + old_commit = str(st.get("head_commit") or "") + cycle = _int(st.get("cycle"), 0) + if not root_id or cycle <= 0: + return Decision( + "escalate", "repair card carries an incomplete supervisor stamp", repair_id + ) + result, error = parse_result(task) + if error is not None: + return Decision("escalate", f"repair result unparseable: {error}", repair_id) + new_commit = extract_commit(task, result) + if new_commit is None: + return Decision( + "escalate", "repair completed without an identifiable new commit", repair_id + ) + if new_commit == old_commit: + return Decision( + "escalate", + "repair produced no new commit over the reviewed revision", + repair_id, + ) + next_cycle = cycle + 1 + if next_cycle > limits.max_cycles: + return Decision( + "escalate", + f"repair cycle limit ({limits.max_cycles}) reached for {root_id}; " + "escalating instead of re-review", + root_id, + ) + if existing_followup(tasks, REVIEW_KIND, root_id, new_commit): + return Decision("none", "a re-review already covers this commit") + pr = extract_pr(task, result) or "" + payload = _build_review( + root_id, repair_id, new_commit, next_cycle, pr, [], limits.review_assignee + ) + return Decision("spawn", f"open re-review for {new_commit}", repair_id, payload) + + +def plan(task: Any, tasks: list[Any], limits: Limits) -> Decision: + """Resolve the single follow-up action for one terminal card.""" + if not is_done(task): + return Decision("none", "task is not in a terminal done state") + kind = supervised_kind(task) + if kind == REVIEW_KIND: + return plan_review(task, tasks, limits) + if kind == REPAIR_KIND: + return plan_repair(task, tasks, limits) + return plan_implementation(task, tasks, limits) diff --git a/testing/quality_contract.json b/testing/quality_contract.json index c9334c7b..b5148396 100644 --- a/testing/quality_contract.json +++ b/testing/quality_contract.json @@ -45,6 +45,8 @@ "services/hermes/scripts/cli_lane_retention.py", "services/hermes/scripts/cli_lane_routing.py", "services/hermes/scripts/cli_lane_runner.py", + "services/hermes/scripts/kanban_supervisor.py", + "services/hermes/scripts/supervisor_policy.py", "testing/__init__.py", "testing/quality_contract.py", "testing/quality_docs.py", @@ -118,6 +120,8 @@ "services/hermes/scripts/cli_lane_retention.py", "services/hermes/scripts/cli_lane_routing.py", "services/hermes/scripts/cli_lane_runner.py", + "services/hermes/scripts/kanban_supervisor.py", + "services/hermes/scripts/supervisor_policy.py", "testing/tests", "testing", "services/gitea/scripts/gitea_branch_protection_check.py", @@ -222,6 +226,8 @@ "scripts/tests/**/*.py", "services/*/scripts/tests/**/*.py", "services/hermes/scripts/cli_lane_*.py", + "services/hermes/scripts/kanban_supervisor.py", + "services/hermes/scripts/supervisor_policy.py", "services/mailu/scripts/mailu_sync.py", "services/mailu/scripts/mailu_sync_listener.py", "services/gitea/scripts/gitea_branch_protection_check.py", @@ -306,6 +312,8 @@ "services/hermes/scripts/cli_lane_retention.py", "services/hermes/scripts/cli_lane_routing.py", "services/hermes/scripts/cli_lane_runner.py", + "services/hermes/scripts/kanban_supervisor.py", + "services/hermes/scripts/supervisor_policy.py", "testing/quality_contract.py", "testing/quality_docs.py", "testing/quality_hygiene.py", diff --git a/testing/tests/test_hermes_kanban_supervisor.py b/testing/tests/test_hermes_kanban_supervisor.py new file mode 100644 index 00000000..71f74a55 --- /dev/null +++ b/testing/tests/test_hermes_kanban_supervisor.py @@ -0,0 +1,370 @@ +"""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. +""" + +from __future__ import annotations + +from contextlib import nullcontext +from types import SimpleNamespace + +import yaml + +from testing.tests.test_hermes_cli_support import HERMES, _agent_deployment, _load + +supervisor = _load("kanban_supervisor") +policy = supervisor.policy + + +class RecordingDb: + """Minimal in-memory kanban_db stub capturing every mutating call.""" + + def __init__(self, tasks, boards=None, raise_on=None): + self._tasks = tasks + self._boards = boards if boards is not None else [{"slug": "cassandra"}] + self.created = [] + self.comments = [] + self.blocked = [] + self.metadata_sets = [] + self._raise_on = raise_on or set() + + def list_boards(self, include_archived=False): + if "list_boards" in self._raise_on: + raise RuntimeError("registry down") + return list(self._boards) + + def scoped_current_board(self, _board): + return nullcontext() + + def connect(self, *, board): + if "connect" in self._raise_on: + raise RuntimeError(f"cannot open {board}") + return SimpleNamespace(close=lambda: None) + + def list_tasks(self, _conn): + return list(self._tasks) + + def create_task(self, _conn, **kwargs): + if "create_task" in self._raise_on: + raise RuntimeError("write failed") + self.created.append(kwargs) + return "new-task" + + def add_comment(self, _conn, task_id, author, body): + self.comments.append((task_id, author, body)) + + def block_task(self, _conn, task_id, reason, kind): + self.blocked.append((task_id, reason, kind)) + return True + + def set_task_metadata(self, _conn, task_id, metadata): + self.metadata_sets.append((task_id, metadata)) + + +def _task(**kw): + base = { + "id": "t", + "status": "done", + "metadata": {}, + "result": None, + "parents": [], + "title": "", + "body": "", + } + base.update(kw) + return SimpleNamespace(**base) + + +def _write_config(tmp_path, monkeypatch, kanban): + path = tmp_path / "config.yaml" + path.write_text(yaml.safe_dump({"kanban": kanban}), encoding="utf-8") + monkeypatch.setattr(supervisor, "CONFIG_PATH", path) + + +# --- configuration -------------------------------------------------------- + + +def test_defaults_are_inert_and_conservative_when_config_missing(tmp_path, monkeypatch): + monkeypatch.setattr(supervisor, "CONFIG_PATH", tmp_path / "absent.yaml") + settings = supervisor.load_settings() + assert settings.enabled is False + assert settings.interval == supervisor.DEFAULT_INTERVAL_SECONDS + assert settings.limits.max_cycles == supervisor.DEFAULT_MAX_CYCLES + assert settings.limits.review_assignee == supervisor.DEFAULT_REVIEW_ASSIGNEE + + +def test_config_values_are_read_and_sanitized(tmp_path, monkeypatch): + _write_config( + tmp_path, + monkeypatch, + { + "auto_supervise": True, + "supervise_interval_seconds": 45, + "supervise_max_cycles": 0, # non-positive -> default + "supervise_max_chains": 8, + "supervise_review_assignee": " ", # blank -> default + "supervise_repair_assignee": "cli-codex-high", + }, + ) + settings = supervisor.load_settings() + assert settings.enabled is True + assert settings.interval == 45 + assert settings.limits.max_cycles == supervisor.DEFAULT_MAX_CYCLES + assert settings.limits.max_chains == 8 + assert settings.limits.review_assignee == supervisor.DEFAULT_REVIEW_ASSIGNEE + assert settings.limits.repair_assignee == "cli-codex-high" + + +def test_malformed_config_falls_back_to_defaults(tmp_path, monkeypatch): + path = tmp_path / "config.yaml" + path.write_text("kanban: [not, a, mapping]", encoding="utf-8") + monkeypatch.setattr(supervisor, "CONFIG_PATH", path) + settings = supervisor.load_settings() + assert settings.enabled is False and settings.limits.max_chains == supervisor.DEFAULT_MAX_CHAINS + + +def test_bool_int_text_helpers_reject_wrong_types(): + assert supervisor._bool("yes", False) is False + assert supervisor._positive_int(True, 5) == 5 + assert supervisor._positive_int(-3, 5) == 5 + assert supervisor._positive_int(2.0, 5) == 2 + assert supervisor._text(7, "d") == "d" + + +# --- board iteration ------------------------------------------------------ + + +def test_board_slug_handles_dict_object_and_scalar(): + assert supervisor._board_slug({"slug": "s"}) == "s" + assert supervisor._board_slug(SimpleNamespace(slug=None, id="i")) == "i" + assert supervisor._board_slug("plain") == "plain" + + +def test_iter_boards_survives_registry_failure(capsys): + db = RecordingDb([], raise_on={"list_boards"}) + assert supervisor._iter_boards(db) == [] + assert "could not list boards" in capsys.readouterr().err + + +def test_iter_boards_drops_empty_slugs(): + db = RecordingDb([], boards=[{"slug": "a"}, {"slug": ""}]) + assert supervisor._iter_boards(db) == ["a"] + + +# --- decision application ------------------------------------------------- + + +def test_impl_done_spawns_review_and_comments(): + impl = _task( + id="impl", + result={"changed_files": ["a.py"], "head_commit": "c1", "pull_request": "pr/1"}, + ) + db = RecordingDb([impl]) + assert supervisor.supervise_once(db, policy.Limits()) == 1 + assert len(db.created) == 1 + assert db.created[0]["idempotency_key"] == "supervisor:review:impl:c1:1" + assert any("supervisor:" in body for _, _, body in db.comments) + assert db.blocked == [] + + +def test_ship_marks_ready_for_human_without_merging(): + review = _task( + id="rev", + metadata={"supervisor": {"kind": "review", "root": "impl", "parent": "impl", "head_commit": "c1", "cycle": 1}}, + result={"verdict": "SHIP", "summary": "clean"}, + ) + db = RecordingDb([review]) + supervisor.supervise_once(db, policy.Limits()) + assert db.metadata_sets == [("impl", {"supervisor_ready_for_human_merge": True})] + body = db.comments[-1][2] + assert "READY FOR HUMAN MERGE" in body and "never" in body + assert db.created == [] and db.blocked == [] + + +def test_escalation_blocks_card_and_creates_no_followup(): + impl = _task(id="impl", result="unparseable") + db = RecordingDb([impl]) + supervisor.supervise_once(db, policy.Limits()) + assert db.created == [] + assert db.blocked and db.blocked[0][0] == "impl" and db.blocked[0][2] == "supervisor" + + +def test_spawn_retries_without_idempotency_key_on_typeerror(): + impl = _task( + id="impl", + result={"changed_files": ["a.py"], "head_commit": "c1", "pull_request": "pr/1"}, + ) + + class LegacyDb(RecordingDb): + def create_task(self, _conn, **kwargs): + if "idempotency_key" in kwargs: + raise TypeError("unexpected idempotency_key") + self.created.append(kwargs) + return "id" + + db = LegacyDb([impl]) + supervisor.supervise_once(db, policy.Limits()) + assert len(db.created) == 1 and "idempotency_key" not in db.created[0] + + +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 + try: + supervisor.apply_decision(db, conn, policy.Decision("teleport")) + raise AssertionError("expected rejection") + except ValueError: + pass + + +def test_ready_flag_tolerates_missing_metadata_setter_and_setter_errors(): + class NoSetterDb(RecordingDb): + set_task_metadata = None + + db = NoSetterDb([]) + conn = SimpleNamespace(close=lambda: None) + supervisor._mark_ready_for_human(db, conn, "impl", "ready") + assert db.comments[-1][0] == "impl" + + class BadSetterDb(RecordingDb): + def set_task_metadata(self, _conn, _task_id, _metadata): + raise RuntimeError("nope") + + db2 = BadSetterDb([]) + supervisor._mark_ready_for_human(db2, conn, "impl", "ready") + assert db2.comments[-1][2] == "ready" + + +def test_comment_and_block_failures_are_swallowed(capsys): + class BrokenDb(RecordingDb): + def add_comment(self, *_a, **_k): + raise RuntimeError("comment down") + + def block_task(self, *_a, **_k): + raise RuntimeError("block down") + + db = BrokenDb([]) + conn = SimpleNamespace(close=lambda: None) + supervisor._escalate(db, conn, "impl", "reason") + err = capsys.readouterr().err + assert "could not block impl" in err and "could not comment on impl" in err + + +# --- fault isolation ------------------------------------------------------ + + +def test_board_action_failure_does_not_stop_the_pass(capsys): + good = _task( + id="impl", + result={"changed_files": ["a"], "head_commit": "c1", "pull_request": "pr"}, + ) + db = RecordingDb([good], raise_on={"create_task"}) + # create_task raises -> action fails, but the pass completes without crashing. + assert supervisor.supervise_once(db, policy.Limits()) == 0 + assert "action failed on board" in capsys.readouterr().err + + +def test_non_actionable_task_is_a_clean_no_op(): + idle = _task(id="idle", status="running") + db = RecordingDb([idle]) + assert supervisor.supervise_once(db, policy.Limits()) == 0 + assert db.created == [] and db.blocked == [] and db.comments == [] + + +def test_board_connect_failure_is_isolated(capsys): + db = RecordingDb([], boards=[{"slug": "cassandra"}], raise_on={"connect"}) + assert supervisor.supervise_once(db, policy.Limits()) == 0 + assert "temporarily skipping board" in capsys.readouterr().err + + +# --- poll loop / auto_supervise gate ------------------------------------- + + +def test_loop_supervises_when_enabled(monkeypatch): + calls = [] + monkeypatch.setattr(supervisor, "supervise_once", lambda db, limits: calls.append(limits) or 0) + settings = supervisor.Settings(enabled=True, interval=0, limits=policy.Limits()) + sleeps = [] + supervisor.run_forever( + object(), sleep=sleeps.append, load=lambda: settings, max_ticks=2 + ) + assert len(calls) == 2 and sleeps == [0, 0] + + +def test_loop_is_inert_when_flag_off(monkeypatch): + calls = [] + monkeypatch.setattr(supervisor, "supervise_once", lambda *a: calls.append(a) or 0) + settings = supervisor.Settings(enabled=False, interval=0, limits=policy.Limits()) + supervisor.run_forever(object(), sleep=lambda _s: None, load=lambda: settings, max_ticks=3) + assert calls == [] + + +def test_loop_survives_a_failing_tick(monkeypatch, capsys): + def boom(_db, _limits): + raise RuntimeError("tick blew up") + + monkeypatch.setattr(supervisor, "supervise_once", boom) + settings = supervisor.Settings(enabled=True, interval=0, limits=policy.Limits()) + supervisor.run_forever(object(), sleep=lambda _s: None, load=lambda: settings, max_ticks=1) + assert "tick failed" in capsys.readouterr().err + + +def test_main_wires_the_runtime_kanban_db(monkeypatch): + seen = {} + monkeypatch.setitem( + __import__("sys").modules, "hermes_cli", SimpleNamespace(kanban_db="RUNTIME") + ) + monkeypatch.setattr(supervisor, "run_forever", lambda db: seen.setdefault("db", db)) + assert supervisor.main([]) == 0 + assert seen["db"] == "RUNTIME" + + +# --- deployment / config wiring ------------------------------------------ + + +def test_sidecar_is_deployed_with_hardened_non_metered_posture(): + spec = _agent_deployment()["spec"]["template"]["spec"] + sidecar = next(c for c in spec["containers"] if c["name"] == "kanban-supervisor") + assert sidecar["command"][-1].endswith("kanban_supervisor.py") + sec = sidecar["securityContext"] + assert sec["runAsNonRoot"] is True and sec["runAsUser"] == 10000 + assert sec["allowPrivilegeEscalation"] is False + assert sec["readOnlyRootFilesystem"] is True + assert sec["capabilities"]["drop"] == ["ALL"] + mounts = {m["name"] for m in sidecar["volumeMounts"]} + # No runtime-access mount: it holds no provider credential / metered path. + assert "runtime-access" not in mounts + assert {"home", "coordinator"} <= mounts + + +def test_auto_supervise_flag_defaults_false_in_configmap(): + documents = list( + yaml.safe_load_all((HERMES / "agent-configmap.yaml").read_text(encoding="utf-8")) + ) + config_doc = next( + doc for doc in documents if doc and doc.get("metadata", {}).get("name") == "hermes-agent-config" + ) + payload = yaml.safe_load(config_doc["data"]["config.yaml"]) + assert payload["kanban"]["auto_supervise"] is False + + +def test_supervisor_scripts_registered_in_coordinator_configmap(): + kustomization = yaml.safe_load((HERMES / "kustomization.yaml").read_text(encoding="utf-8")) + coordinator = next( + gen for gen in kustomization["configMapGenerator"] if gen["name"] == "hermes-coordinator" + ) + joined = "\n".join(coordinator["files"]) + assert "kanban_supervisor.py=scripts/kanban_supervisor.py" in joined + assert "supervisor_policy.py=scripts/supervisor_policy.py" in joined + + +# --- the confirmed NULL -> 20 goal-turn fallback fix ---------------------- + + +def test_goal_max_turns_defaults_to_twenty_not_one(): + source = (HERMES / "scripts/cli_lane_execution.py").read_text(encoding="utf-8") + assert 'int(_task_value(task, "goal_max_turns", 20) or 20)' in source + assert 'int(_task_value(task, "goal_max_turns", 1) or 1)' not in source diff --git a/testing/tests/test_hermes_kanban_supervisor_policy.py b/testing/tests/test_hermes_kanban_supervisor_policy.py new file mode 100644 index 00000000..1dde9e20 --- /dev/null +++ b/testing/tests/test_hermes_kanban_supervisor_policy.py @@ -0,0 +1,422 @@ +"""Behavioral tests for the supervisor's pure decision state machine. + +Every transition, dedup guard, fail-closed branch, and bound is exercised on +real policy logic (no mock-asserting): implementation-done->review, SHIP->ready, +BLOCK->repair, repair->re-review, per-(parent, commit) dedup including a +concurrent external shepherd, ambiguous/unparseable fail-closed, and the +cycle-limit escalation. +""" + +from __future__ import annotations + +import json + +from types import SimpleNamespace + +from testing.tests.test_hermes_cli_support import _load + +policy = _load("supervisor_policy") + +LIMITS = policy.Limits(max_cycles=3, max_chains=5) + + +def task(**kw): + base = { + "id": "t1", + "status": "done", + "assignee": "", + "title": "", + "body": "", + "result": None, + "metadata": {}, + "parents": [], + } + base.update(kw) + return SimpleNamespace(**base) + + +def review_stamp(root="impl", commit="c1", cycle=1): + return { + "supervisor": { + "kind": "review", + "root": root, + "parent": root, + "head_commit": commit, + "cycle": cycle, + } + } + + +def repair_stamp(root="impl", commit="c1", cycle=1): + return { + "supervisor": { + "kind": "repair", + "root": root, + "parent": "rev", + "head_commit": commit, + "cycle": cycle, + } + } + + +# --- primitives ----------------------------------------------------------- + + +def test_parse_result_fails_closed_on_absent_and_non_object(): + assert policy.parse_result(task(result=None))[1] + assert policy.parse_result(task(result=""))[1] + assert policy.parse_result(task(result={}))[1] + assert policy.parse_result(task(result="not json"))[1] + assert policy.parse_result(task(result="[1, 2]"))[1] + + +def test_parse_result_accepts_dict_and_json_string(): + parsed, error = policy.parse_result(task(result={"status": "completed"})) + assert error is None and parsed == {"status": "completed"} + parsed, error = policy.parse_result(task(result='{"status": "completed"}')) + assert error is None and parsed["status"] == "completed" + + +def test_metadata_parses_json_string_and_rejects_scalar(): + assert policy.metadata(task(metadata='{"a": 1}')) == {"a": 1} + assert policy.metadata(task(metadata="oops")) == {} + assert policy.metadata(task(metadata="7")) == {} + + +def test_extract_commit_and_pr_scan_metadata_result_and_nested(): + from_meta = task(metadata={"head_commit": "m"}, result={}) + assert policy.extract_commit(from_meta, {}) == "m" + nested = task(result={"metadata": {"commit": "n"}}) + parsed, _ = policy.parse_result(nested) + assert policy.extract_commit(nested, parsed) == "n" + assert policy.extract_commit(task(result={"status": "x"}), {"status": "x"}) is None + assert policy.extract_pr(task(result={"branch": "feat"}), {"branch": "feat"}) == "feat" + assert policy.extract_pr(task(result={"pr_url": "u"}), {"pr_url": "u"}) == "u" + assert policy.extract_pr(task(result={}), {}) is None + + +def test_parents_reads_lists_dicts_and_task_links_fallback(): + assert policy.parents(task(parents=["a", "b"])) == ["a", "b"] + linked = task(parents=None, task_links={"parents": [{"id": "p1"}, {"parent": "p2"}]}) + assert policy.parents(linked) == ["p1", "p2"] + assert policy.parents(task(parents={"parents": ["z"]})) == ["z"] + assert policy.parents(task(parents=None, task_links=None)) == [] + + +def test_assert_safe_rejects_unknown_action_and_forbidden_keys(): + assert policy.assert_safe(policy.Decision("none")).action == "none" + try: + policy.assert_safe(policy.Decision("merge")) + raise AssertionError("expected rejection") + except ValueError: + pass + try: + policy.assert_safe(policy.Decision("spawn", payload={"approve": True})) + raise AssertionError("expected rejection") + except ValueError: + pass + + +# --- implementation -> review -------------------------------------------- + + +def test_done_implementation_with_pr_spawns_one_review(): + impl = task( + id="impl", + result={"changed_files": ["a.py"], "head_commit": "c1", "pull_request": "pr/1"}, + ) + decision = policy.plan(impl, [impl], LIMITS) + assert decision.action == "spawn" and decision.target_id == "impl" + payload = decision.payload + assert payload["assignee"] == LIMITS.review_assignee + assert payload["parents"] == ["impl"] + assert payload["idempotency_key"] == "supervisor:review:impl:c1:1" + stamp = payload["metadata"]["supervisor"] + assert stamp == { + "kind": "review", + "root": "impl", + "parent": "impl", + "head_commit": "c1", + "cycle": 1, + } + assert "Hermes-Task-Role: review" in payload["body"] + assert payload["metadata"]["task_role"] == "review" + + +def test_non_terminal_task_is_never_acted_on(): + impl = task(id="impl", status="running", result={"head_commit": "c1"}) + assert policy.plan(impl, [impl], LIMITS).action == "none" + + +def test_unparseable_implementation_result_fails_closed(): + impl = task(id="impl", result="broken") + decision = policy.plan(impl, [impl], LIMITS) + assert decision.action == "escalate" and decision.target_id == "impl" + + +def test_done_implementation_with_changes_but_no_commit_fails_closed(): + impl = task(id="impl", result={"changed_files": ["a.py"]}) + assert policy.plan(impl, [impl], LIMITS).action == "escalate" + + +def test_done_implementation_with_no_commit_and_no_changes_is_idle(): + impl = task(id="impl", result={"status": "completed", "summary": "nothing to do"}) + assert policy.plan(impl, [impl], LIMITS).action == "none" + + +def test_commit_without_pr_fails_closed(): + impl = task(id="impl", result={"changed_files": ["a.py"], "head_commit": "c1"}) + assert policy.plan(impl, [impl], LIMITS).action == "escalate" + + +def test_review_shaped_unstamped_card_is_left_to_the_shepherd(): + card = task( + id="ext", + body="Hermes-Task-Role: review\nRead-only review, report SHIP or BLOCK.", + result={"verdict": "SHIP"}, + ) + assert policy.plan(card, [card], LIMITS).action == "none" + + +def test_existing_supervisor_review_blocks_duplicate(): + impl = task( + id="impl", + result={"changed_files": ["a.py"], "head_commit": "c1", "pull_request": "pr/1"}, + ) + review = task(id="rev", status="ready", metadata=review_stamp("impl", "c1")) + assert policy.plan(impl, [impl, review], LIMITS).action == "none" + + +def test_existing_external_shepherd_review_blocks_duplicate(): + impl = task( + id="impl", + result={"changed_files": ["a.py"], "head_commit": "c1", "pull_request": "pr/1"}, + ) + shepherd = task( + id="ext", + status="ready", + title="Review of the change", + body="Read-only review of commit c1; report SHIP or BLOCK.", + parents=["impl"], + assignee="cli-claude-xhigh", + ) + assert policy.plan(impl, [impl, shepherd], LIMITS).action == "none" + + +def test_chain_ceiling_defers_new_reviews(): + impl = task( + id="impl", + result={"changed_files": ["a.py"], "head_commit": "c1", "pull_request": "pr/1"}, + ) + active = [ + task(id=f"r{i}", status="ready", metadata=review_stamp(f"root{i}", "x")) + for i in range(LIMITS.max_chains) + ] + assert policy.plan(impl, [impl, *active], LIMITS).action == "none" + + +# --- review -> ship / repair --------------------------------------------- + + +def test_review_ship_marks_parent_ready_without_merging(): + review = task( + id="rev", + metadata=review_stamp("impl", "c1", 1), + result={"verdict": "SHIP", "summary": "clean"}, + ) + decision = policy.plan(review, [review], LIMITS) + assert decision.action == "ship" and decision.target_id == "impl" + assert decision.payload == {"commit": "c1", "pr": ""} + + +def test_review_block_spawns_bounded_repair_with_findings(): + review = task( + id="rev", + metadata=review_stamp("impl", "c1", 1), + result={"verdict": "BLOCK", "findings": ["null deref", "missing test"]}, + ) + decision = policy.plan(review, [review], LIMITS) + assert decision.action == "spawn" + payload = decision.payload + assert payload["metadata"]["supervisor"]["kind"] == "repair" + assert payload["metadata"]["supervisor"]["cycle"] == 1 + assert payload["parents"] == ["impl", "rev"] + assert payload["assignee"] == LIMITS.repair_assignee + assert "null deref" in payload["body"] and "missing test" in payload["body"] + assert payload["idempotency_key"] == "supervisor:repair:impl:c1:1" + + +def test_review_with_incomplete_stamp_fails_closed(): + review = task(id="rev", metadata={"supervisor": {"kind": "review"}}, result={"verdict": "SHIP"}) + assert policy.plan(review, [review], LIMITS).action == "escalate" + + +def test_review_unparseable_result_fails_closed(): + review = task(id="rev", metadata=review_stamp(), result="nope") + assert policy.plan(review, [review], LIMITS).action == "escalate" + + +def test_review_ambiguous_verdict_fails_closed(): + review = task( + id="rev", + metadata=review_stamp(), + result={"status": "completed", "summary": "looked at it"}, + ) + assert policy.plan(review, [review], LIMITS).action == "escalate" + + +def test_review_block_at_cycle_limit_escalates_instead_of_repairing(): + 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" + + +def test_review_block_skips_when_repair_already_exists(): + review = task( + id="rev", + metadata=review_stamp("impl", "c1", 1), + result={"verdict": "BLOCK", "findings": ["x"]}, + ) + repair = task(id="rep", status="ready", metadata=repair_stamp("impl", "c1", 1)) + assert policy.plan(review, [review, repair], LIMITS).action == "none" + + +# --- repair -> re-review -------------------------------------------------- + + +def test_repair_with_new_commit_spawns_re_review_at_next_cycle(): + repair = task( + id="rep", + metadata=repair_stamp("impl", "c1", 1), + result={"changed_files": ["a.py"], "head_commit": "c2", "branch": "feat"}, + ) + decision = policy.plan(repair, [repair], LIMITS) + assert decision.action == "spawn" + payload = decision.payload + stamp = payload["metadata"]["supervisor"] + assert stamp["kind"] == "review" and stamp["cycle"] == 2 and stamp["head_commit"] == "c2" + assert payload["parents"] == ["impl", "rep"] + assert payload["idempotency_key"] == "supervisor:review:impl:c2:2" + + +def test_repair_incomplete_stamp_fails_closed(): + repair = task(id="rep", metadata={"supervisor": {"kind": "repair"}}, result={"head_commit": "c2"}) + assert policy.plan(repair, [repair], LIMITS).action == "escalate" + + +def test_repair_unparseable_result_fails_closed(): + repair = task(id="rep", metadata=repair_stamp(), result="broken") + assert policy.plan(repair, [repair], LIMITS).action == "escalate" + + +def test_repair_without_new_commit_fails_closed(): + repair = task(id="rep", metadata=repair_stamp("impl", "c1", 1), result={"status": "completed"}) + assert policy.plan(repair, [repair], LIMITS).action == "escalate" + + +def test_repair_producing_same_commit_fails_closed(): + repair = task( + id="rep", + metadata=repair_stamp("impl", "c1", 1), + result={"head_commit": "c1"}, + ) + assert policy.plan(repair, [repair], LIMITS).action == "escalate" + + +def test_repair_at_cycle_limit_escalates_instead_of_re_reviewing(): + 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" + + +def test_repair_skips_when_re_review_already_exists(): + repair = task( + id="rep", + metadata=repair_stamp("impl", "c1", 1), + result={"head_commit": "c2"}, + ) + existing = task(id="rev2", status="ready", metadata=review_stamp("impl", "c2", 2)) + assert policy.plan(repair, [repair, existing], LIMITS).action == "none" + + +# --- dedup / classification helpers -------------------------------------- + + +def test_existing_followup_matches_external_repair_by_title_and_commit(): + repair = task( + id="ext-rep", + title="Repair the regression", + body="fixes commit c1", + parents=["impl"], + ) + assert policy.existing_followup([repair], "repair", "impl", "c1") is True + assert policy.existing_followup([repair], "repair", "impl", "other") is False + + +def test_active_chain_count_only_counts_in_flight_supervised_cards(): + tasks = [ + task(id="a", status="ready", metadata=review_stamp("r1", "c")), + task(id="b", status="done", metadata=review_stamp("r2", "c")), + task(id="c", status="blocked", metadata=repair_stamp("r3", "c")), + task(id="d", status="ready", metadata={}), + ] + assert policy.active_chain_count(tasks) == 1 + + +def test_looks_like_helpers_cover_role_metadata_and_titles(): + assert policy._looks_like_review(task(metadata={"task_role": "review"})) + assert policy._looks_like_review(task(title="Review pass")) + assert policy._looks_like_repair(task(metadata={"task_role": "implementation"})) + assert policy._looks_like_repair(task(title="fix the bug")) + assert not policy._looks_like_repair(task(title="ship it")) + + +def test_helper_edge_branches_are_defensive(): + assert policy.field({"id": "d"}, "id") == "d" + assert policy.parents(task(parents=["ok", "", {"id": None}])) == ["ok"] + assert policy._first_key(None, policy.COMMIT_KEYS) is None + assert policy._result_sources(task(), None) == [{}] + assert policy._strings("scalar") == [] + assert policy._references(task(), "") is False + assert policy.existing_followup([], "review", "impl", "c1") is False + + +def test_active_chain_count_ignores_supervised_card_without_root(): + tasks = [task(id="a", status="ready", metadata={"supervisor": {"kind": "review"}})] + assert policy.active_chain_count(tasks) == 0 + + +def test_review_block_without_findings_still_spawns_repair(): + review = task( + id="rev", + metadata=review_stamp("impl", "c1", 1), + result={"verdict": "BLOCK", "summary": "must not ship BLOCK"}, + ) + decision = policy.plan(review, [review], LIMITS) + assert decision.action == "spawn" + assert "findings to address" not in decision.payload["body"] + + +def test_no_decision_ever_yields_a_metered_or_merge_action(): + corpus = [ + task(id="impl", result={"changed_files": ["a"], "head_commit": "c1", "pull_request": "p"}), + task(id="rev", metadata=review_stamp(), result={"verdict": "SHIP"}), + task(id="rev2", metadata=review_stamp(), result={"verdict": "BLOCK", "findings": ["x"]}), + task(id="rep", metadata=repair_stamp(), result={"head_commit": "c9"}), + task(id="bad", metadata=review_stamp(), result="garbage"), + ] + for candidate in corpus: + decision = policy.plan(candidate, corpus, LIMITS) + assert decision.action in policy.SAFE_ACTIONS + payload = json.dumps(decision.payload or {}) + for forbidden in ("merge", "approve", "provider", "api_key", "deploy"): + assert forbidden not in payload + policy.assert_safe(decision)