diff --git a/hermes_cli/commands.py b/hermes_cli/commands.py index 10f8fbf046b8..7fe287aa51c8 100644 --- a/hermes_cli/commands.py +++ b/hermes_cli/commands.py @@ -209,7 +209,8 @@ class CommandDef: "claim", "comment", "complete", "edit", "block", "unblock", "archive", "tail", "dispatch", "stats", "notify-subscribe", "notify-list", "notify-unsubscribe", "log", "runs", - "heartbeat", "assignees", "context", "specify", "gc")), + "heartbeat", "assignees", "context", "specify", "decompose", + "resolve-fanin", "gc")), CommandDef("reload", "Reload .env variables into the running session", "Tools & Skills", cli_only=True), CommandDef("reload-mcp", "Reload MCP servers from config", "Tools & Skills", diff --git a/hermes_cli/kanban.py b/hermes_cli/kanban.py index 7fc7bf948950..2f01bdc292aa 100644 --- a/hermes_cli/kanban.py +++ b/hermes_cli/kanban.py @@ -859,6 +859,31 @@ def build_parser(parent_subparsers: argparse._SubParsersAction) -> argparse.Argu p_gc.add_argument("--log-retention-days", type=int, default=30, help="Delete worker log files older than N days (default: 30)") + # --- resolve-fanin --- + p_rf = sub.add_parser( + "resolve-fanin", + help=( + "Classify a final review task and, in --apply mode, create a " + "deduped fix + fix-review remediation pair for BLOCK/NEED_MORE." + ), + ) + p_rf.add_argument("final_task_id") + mode = p_rf.add_mutually_exclusive_group() + mode.add_argument("--dry-run", dest="rf_dry_run", action="store_true", + help="Classification only — never write (default)") + mode.add_argument("--apply", dest="rf_apply", action="store_true", + help="Create deduped fix + fix-review cards when REQUIRED") + p_rf.add_argument("--fix-assignee", default=None, + help="Profile assigned to the auto-created fix card") + p_rf.add_argument("--review-assignee", default=None, + help="Profile assigned to the auto-created fix-review card") + p_rf.add_argument("--reporter-assignee", default=None, + help="Optional fan-in reporter profile gated behind the fix-review") + p_rf.add_argument("--max-fan-in-threshold", type=int, default=32, + help="Maximum transitive graph size eligible for --apply auto-remediation (default: 32; raise explicitly for larger fan-ins)") + p_rf.add_argument("--json", action="store_true", + help="Emit the machine-readable ledger JSON (always JSON-shaped output)") + kanban_parser.set_defaults(_kanban_parser=kanban_parser) return kanban_parser @@ -974,6 +999,7 @@ def kanban_command(args: argparse.Namespace) -> int: "specify": _cmd_specify, "decompose": _cmd_decompose, "gc": _cmd_gc, + "resolve-fanin": _cmd_resolve_fanin, } handler = handlers.get(action) if not handler: @@ -2740,6 +2766,37 @@ def _cmd_gc(args: argparse.Namespace) -> int: return 0 +def _cmd_resolve_fanin(args: argparse.Namespace) -> int: + """Classify a final review task and optionally create deduped + remediation cards. See ``hermes_cli.kanban_resolver`` for the + classification logic and safety boundaries. + + Always emits a JSON ledger so downstream tooling (and the test + suite) can parse the outcome regardless of ``--json``. + """ + from hermes_cli import kanban_resolver as kr + + apply = bool(getattr(args, "rf_apply", False)) + # dry-run is the safe default whenever --apply isn't set, even + # without the explicit --dry-run flag. + with kb.connect() as conn: + try: + ledger = kr.resolve_fanin( + conn, + args.final_task_id, + apply=apply, + fix_assignee=getattr(args, "fix_assignee", None), + review_assignee=getattr(args, "review_assignee", None), + reporter_assignee=getattr(args, "reporter_assignee", None), + max_fan_in_threshold=getattr(args, "max_fan_in_threshold", None), + ) + except ValueError as exc: + print(json.dumps({"error": str(exc)}, indent=2)) + return 1 + print(json.dumps(ledger, indent=2, ensure_ascii=False)) + return 0 + + # --------------------------------------------------------------------------- # Slash-command entry point (used by /kanban from CLI and gateway) # --------------------------------------------------------------------------- diff --git a/hermes_cli/kanban_resolver.py b/hermes_cli/kanban_resolver.py new file mode 100644 index 000000000000..5969a71c00f6 --- /dev/null +++ b/hermes_cli/kanban_resolver.py @@ -0,0 +1,490 @@ +"""Kanban fan-in resolver — root-fix slice for review BLOCK/NEED_MORE. + +When an implementation or review graph finishes at a final review task +with verdict ``BLOCK`` or ``NEED_MORE``, the originating operator can +be left waiting: ``notify-subscribe`` is a passive terminal delivery, +not an active fan-in/remediation resolver. This module classifies the +final task into three independent dimensions and, in apply mode, +creates at most one deduped fix card + one fix-review card so a +runnable remediation graph exists on the board. The fix card is +kept dependency-free (fix ready -> fix-review todo -> reporter todo) +so it dispatches even when the final review is a sticky ``blocked`` +verdict; the link back to the final task is recorded as an audit +comment rather than a dependency edge. + +Classification heuristics +------------------------- +* ``task_verdict``: ``GO`` | ``BLOCK`` | ``NEED_MORE`` — parsed from + the final review task's summary / result / body / latest comments. +* ``ack_status``: ``PENDING`` | ``SENT`` | ``FAILED`` | + ``WATCHDOG_FALLBACK`` | ``MANUAL_RELAYED`` — pulled from a comment + whose body starts with ``ack-status:`` (the documented surface that + gateway watchdogs / manual relays write). Default ``PENDING``. +* ``remediation_status``: ``NONE`` | ``REQUIRED`` | ``CREATED`` | + ``BLOCKED``. ``BLOCKED`` is reserved for unsafe sentinel categories + the resolver must never auto-fix. + +Safety +------ +The resolver only auto-creates remediation cards when the final review +task's combined text: + +* contains a ``Verdict: BLOCK`` or ``Verdict: NEED_MORE`` line, AND +* references a ``review-required:`` or ``handoff:`` sentinel, AND +* stays within the caller's ``max_fan_in_threshold`` graph-size bound, + when one is supplied, AND +* does NOT mention any unsafe blocker category — secret/token/key, + destructive data loss, auth/credential/login, user input required, + live money / trading / execution. + +Card bodies are sanitised: only the verdict, a short reason summary, +and the originating task id are included. Raw transcript paths, +secrets, and full review text never reach the new task body. +""" + +from __future__ import annotations + +import json +import re +import sqlite3 +from typing import Any, Iterable, Optional + +from hermes_cli import kanban_db as kb + + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +VALID_VERDICTS = ("GO", "BLOCK", "NEED_MORE") +VALID_ACK_STATUSES = ( + "PENDING", "SENT", "FAILED", "WATCHDOG_FALLBACK", "MANUAL_RELAYED", +) +VALID_REMEDIATION = ("NONE", "REQUIRED", "CREATED", "BLOCKED") + +_VERDICT_RE = re.compile(r"verdict\s*[:=]\s*(GO|BLOCK|NEED_MORE)\b", re.IGNORECASE) +_ACK_RE = re.compile( + r"ack[-_ ]?status\s*[:=]\s*(PENDING|SENT|FAILED|WATCHDOG_FALLBACK|MANUAL_RELAYED)\b", + re.IGNORECASE, +) + +# Sentinels that mean "an operator/worker explicitly handed back for human +# review/handoff" — the only categories we are allowed to auto-resolve. +_AUTORESOLVE_SENTINELS = ( + re.compile(r"review[-_]required", re.IGNORECASE), + re.compile(r"\bhandoff\b", re.IGNORECASE), +) + +# Unsafe blocker categories. Matching any one of these in the combined +# review text forces ``remediation_status=BLOCKED`` — no card creation, +# even in apply mode. +_UNSAFE_PATTERNS: tuple[re.Pattern[str], ...] = ( + re.compile(r"\b(secret|secrets|token|api[-_ ]?key|password|credential|cred)\b", re.IGNORECASE), + re.compile(r"\bleak(ed|ing)?\b", re.IGNORECASE), + re.compile(r"\b(rm\s+-rf|drop\s+table|truncate\s+table|destructive|data[-_ ]?loss|wipe)\b", re.IGNORECASE), + re.compile(r"\b(auth|authentication|login|oauth|sso)\b", re.IGNORECASE), + re.compile(r"\b(user[-_ ]?input|needs?\s+user\s+input|prompt\s+the\s+user)\b", re.IGNORECASE), + re.compile(r"\b(live[-_ ]?(money|trading|exec(ution)?))\b", re.IGNORECASE), + re.compile(r"\b(real[-_ ]?money|production[-_ ]?order|execute\s+trade)\b", re.IGNORECASE), +) + +_IDEMPOTENCY_FIX_PREFIX = "fanin-fix:" +_IDEMPOTENCY_REVIEW_PREFIX = "fanin-fix-review:" +_IDEMPOTENCY_REPORTER_PREFIX = "fanin-reporter:" + + +# --------------------------------------------------------------------------- +# Classification +# --------------------------------------------------------------------------- + +def _gather_review_text(conn: sqlite3.Connection, task_id: str) -> str: + """Concatenate body / result / latest comments for parsing. + + Never include free-form payload bodies — only the text fields the + operator can reasonably reason about. Cap each comment to its first + 400 chars so a runaway log dump can't blow up the regex pass. + """ + row = conn.execute( + "SELECT body, result FROM tasks WHERE id = ?", (task_id,) + ).fetchone() + if row is None: + return "" + parts: list[str] = [] + for field in ("body", "result"): + v = row[field] + if v: + parts.append(str(v)) + # Worker completions usually store the operator-facing Verdict in + # task_runs.summary, not tasks.result. Include recent run summaries + # and completed-event summaries so `kanban_complete(summary=...)` + # handoffs classify correctly even when result is empty. + for r in conn.execute( + "SELECT summary, metadata, error FROM task_runs " + "WHERE task_id = ? ORDER BY id DESC LIMIT 5", + (task_id,), + ).fetchall(): + for field in ("summary", "metadata", "error"): + v = r[field] + if v: + parts.append(str(v)[:800]) + for ev in conn.execute( + "SELECT payload FROM task_events WHERE task_id = ? " + "ORDER BY id DESC LIMIT 10", + (task_id,), + ).fetchall(): + payload = ev["payload"] + if not payload: + continue + try: + data = json.loads(payload) + except (TypeError, json.JSONDecodeError): + parts.append(str(payload)[:400]) + continue + summary = data.get("summary") if isinstance(data, dict) else None + if summary: + parts.append(str(summary)[:400]) + for c in kb.list_comments(conn, task_id): + if c.body: + parts.append(c.body[:400]) + return "\n".join(parts) + + +def _parse_verdict(text: str) -> str: + m = _VERDICT_RE.search(text or "") + if not m: + # Default to GO when no verdict marker is present — callers can + # override with an explicit verdict in the review summary. + return "GO" + return m.group(1).upper() + + +def _parse_ack_status(text: str) -> str: + m = _ACK_RE.search(text or "") + if not m: + return "PENDING" + return m.group(1).upper() + + +def _is_unsafe(text: str) -> Optional[str]: + """Return the matching unsafe-category snippet, or ``None`` if safe.""" + for pat in _UNSAFE_PATTERNS: + m = pat.search(text or "") + if m: + return m.group(0) + return None + + +def _has_autoresolve_sentinel(text: str) -> bool: + return any(p.search(text or "") for p in _AUTORESOLVE_SENTINELS) + + +def _walk_ancestors(conn: sqlite3.Connection, task_id: str) -> list[str]: + """Return all transitive ancestors of ``task_id`` (excluding it) + plus the task itself, deduped & sorted for determinism.""" + seen: set[str] = set() + stack: list[str] = [task_id] + while stack: + cur = stack.pop() + if cur in seen: + continue + seen.add(cur) + for pid in kb.parent_ids(conn, cur): + if pid not in seen: + stack.append(pid) + return sorted(seen) + + +def _root_ancestor(conn: sqlite3.Connection, task_id: str) -> Optional[str]: + """Return the deterministic root ancestor (oldest, then min id). + + Used as the ``origin_return`` field — the originator the ledger + points back to. When the task has multiple roots, prefer the + earliest ``created_at`` then lowest id for stability. + """ + ancestors = _walk_ancestors(conn, task_id) + roots: list[str] = [] + for aid in ancestors: + if not kb.parent_ids(conn, aid): + roots.append(aid) + if not roots: + return None + rows = conn.execute( + "SELECT id, created_at FROM tasks WHERE id IN (" + + ",".join("?" * len(roots)) + ")", + roots, + ).fetchall() + rows = sorted(rows, key=lambda r: (r["created_at"] or 0, r["id"])) + return rows[0]["id"] if rows else None + + +def _sanitize_snippet(text: str, *, max_len: int = 240) -> str: + """Strip suspicious tokens (paths starting with ``/``, secret-like + runs of hex) and clip to ``max_len`` characters. Always operates on + the first non-empty line so we never include multi-line payloads. + """ + if not text: + return "" + line = next((ln for ln in text.splitlines() if ln.strip()), "") + # Scrub anything that looks like an absolute path. + line = re.sub(r"(?:[a-zA-Z]:)?(?:/[\w.\-]+){2,}", "", line) + # Scrub long hex/base64-ish runs (likely tokens). + line = re.sub(r"[A-Za-z0-9_\-]{32,}", "", line) + line = line.strip() + if len(line) > max_len: + line = line[: max_len - 1].rstrip() + "…" + return line + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + +def classify(conn: sqlite3.Connection, final_task_id: str) -> dict[str, Any]: + """Classify the final review task without mutating any state.""" + task = kb.get_task(conn, final_task_id) + if task is None: + raise ValueError(f"unknown task {final_task_id}") + text = _gather_review_text(conn, final_task_id) + verdict = _parse_verdict(text) + ack = _parse_ack_status(text) + + if verdict == "GO": + remediation = "NONE" + unsafe_hit: Optional[str] = None + else: + unsafe_hit = _is_unsafe(text) + if unsafe_hit: + remediation = "BLOCKED" + elif _has_autoresolve_sentinel(text): + remediation = "REQUIRED" + else: + # No autoresolve sentinel + non-GO verdict => still + # remediation-required, but we err on the side of safety + # and refuse to auto-fix without an explicit handoff marker. + remediation = "BLOCKED" + unsafe_hit = "missing review-required/handoff sentinel" + + return { + "task_verdict": verdict, + "ack_status": ack, + "remediation_status": remediation, + "blocked_reason": unsafe_hit, + "final_task": final_task_id, + "final_status": task.status, + } + + +def _existing_remediation_ids( + conn: sqlite3.Connection, final_task_id: str +) -> tuple[Optional[str], Optional[str]]: + """Look up previously created fix / fix-review cards for a final task + by idempotency_key. Returns ``(fix_id, fix_review_id)`` (each may be + ``None``). + """ + fix_key = _IDEMPOTENCY_FIX_PREFIX + final_task_id + review_key = _IDEMPOTENCY_REVIEW_PREFIX + final_task_id + rows = conn.execute( + "SELECT id, idempotency_key FROM tasks " + "WHERE idempotency_key IN (?, ?) AND status != 'archived'", + (fix_key, review_key), + ).fetchall() + fix_id: Optional[str] = None + review_id: Optional[str] = None + for r in rows: + if r["idempotency_key"] == fix_key: + fix_id = r["id"] + elif r["idempotency_key"] == review_key: + review_id = r["id"] + return fix_id, review_id + + +def _existing_reporter_id(conn: sqlite3.Connection, final_task_id: str) -> Optional[str]: + """Return the deduped final fan-in reporter card for a final task.""" + row = conn.execute( + "SELECT id FROM tasks WHERE idempotency_key = ? " + "AND status != 'archived' ORDER BY created_at DESC LIMIT 1", + (_IDEMPOTENCY_REPORTER_PREFIX + final_task_id,), + ).fetchone() + return row["id"] if row else None + + +def resolve_fanin( + conn: sqlite3.Connection, + final_task_id: str, + *, + apply: bool = False, + fix_assignee: Optional[str] = None, + review_assignee: Optional[str] = None, + reporter_assignee: Optional[str] = None, + board: Optional[str] = None, + max_fan_in_threshold: Optional[int] = None, +) -> dict[str, Any]: + """Classify and (optionally) create deduped remediation cards. + + ``apply=False`` is dry-run: classification only, no writes. + + When ``apply=True`` and ``remediation_status == 'REQUIRED'``, create + at most one fix card + one fix-review card. Both carry deterministic + ``idempotency_key`` values, so re-running the resolver on the same + final task is a safe no-op. + + ``max_fan_in_threshold`` bounds automatic remediation by graph size. + If the final task's transitive dependency graph is larger than this + positive threshold, the resolver returns ``BLOCKED`` and writes + nothing. Operators can raise the threshold explicitly after reviewing + the graph. + + Returns a machine-readable ledger: + + .. code-block:: json + + { + "board": "default", + "origin_return": "t_…", + "graph_task_ids": ["t_…", ...], + "final_task": "t_…", + "task_verdict": "BLOCK", + "ack_status": "PENDING", + "remediation_status": "CREATED", + "remediation_task_ids": {"fix": "t_…", "fix_review": "t_…"}, + "dry_run": false, + "blocked_reason": null + } + """ + if max_fan_in_threshold is not None and max_fan_in_threshold < 1: + raise ValueError("max_fan_in_threshold must be >= 1") + + base = classify(conn, final_task_id) + verdict = base["task_verdict"] + remediation = base["remediation_status"] + graph_task_ids = _walk_ancestors(conn, final_task_id) + graph_task_count = len(graph_task_ids) + + threshold_blocked_reason: Optional[str] = None + if ( + max_fan_in_threshold is not None + and graph_task_count > max_fan_in_threshold + and remediation == "REQUIRED" + ): + remediation = "BLOCKED" + threshold_blocked_reason = ( + f"fan-in graph size {graph_task_count} exceeds threshold " + f"{max_fan_in_threshold}" + ) + + # Existing cards (if any) — surface them even on dry-run so the + # operator can tell the resolver already ran. + existing_fix, existing_review = _existing_remediation_ids(conn, final_task_id) + if existing_fix and existing_review: + # Treat a fully-formed prior run as the canonical CREATED state. + remediation = "CREATED" + + rem_ids: dict[str, Optional[str]] = {"fix": existing_fix, "fix_review": existing_review} + + ledger = { + "board": kb.get_current_board() if board is None else board, + "origin_return": _root_ancestor(conn, final_task_id), + "graph_task_ids": graph_task_ids, + "graph_task_count": graph_task_count, + "max_fan_in_threshold": max_fan_in_threshold, + "final_task": final_task_id, + "task_verdict": verdict, + "ack_status": base["ack_status"], + "remediation_status": remediation, + "remediation_task_ids": rem_ids, + "reporter_task_id": _existing_reporter_id(conn, final_task_id), + "dry_run": not apply, + "blocked_reason": threshold_blocked_reason or base["blocked_reason"], + } + + if not apply: + return ledger + if remediation != "REQUIRED": + # Either NONE / BLOCKED / already CREATED — nothing to do. + return ledger + + # ----- Build sanitised card bodies. ----- + text = _gather_review_text(conn, final_task_id) + verdict_summary = _sanitize_snippet(text) + fix_body = ( + f"Auto-created from {final_task_id} (Verdict: {verdict}).\n" + f"Summary: {verdict_summary or '(no summary)'}\n" + f"Apply minimal remediation. See the parent review card for context." + ) + review_body = ( + f"Auto-created fix review for {final_task_id} (Verdict: {verdict}).\n" + f"Re-run reviewer flow against the fix output." + ) + + fix_key = _IDEMPOTENCY_FIX_PREFIX + final_task_id + review_key = _IDEMPOTENCY_REVIEW_PREFIX + final_task_id + + # Do NOT parent the fix card to the final review task. The final may be + # a sticky ``blocked`` review-required verdict we must preserve as-is; + # parenting the fix under it would leave the fix stuck in ``todo`` (a + # child can only reach ``ready`` once every parent is ``done``), so the + # dispatcher — which only claims ``ready`` tasks — could never run it. + # The remediation graph is kept self-runnable (fix ready -> fix-review + # todo -> reporter todo) and the link back to the final review is + # preserved as an audit comment + the fix body, not a dependency edge. + fresh_fix = existing_fix is None + fix_id = existing_fix or kb.create_task( + conn, + title=f"fix: remediate {final_task_id} ({verdict})", + body=fix_body, + assignee=fix_assignee, + created_by="fanin-resolver", + parents=[], + idempotency_key=fix_key, + initial_status="running", + ) + review_id = existing_review or kb.create_task( + conn, + title=f"fix-review: verify {final_task_id} remediation", + body=review_body, + assignee=review_assignee, + created_by="fanin-resolver", + parents=[fix_id], + idempotency_key=review_key, + initial_status="running", + ) + + # Audit trail: because we deliberately do not add a dependency edge from + # the (possibly still-``blocked``) final review to the fix card, record + # the created remediation ids as a comment on the final task so the + # verdict/handoff history points forward to the remediation graph. Only + # write it on a fresh creation so re-runs stay idempotent (no duplicate + # comment). The final task's own verdict/status is never mutated. + if fresh_fix: + kb.add_comment( + conn, + final_task_id, + author="fanin-resolver", + body=( + f"remediation-created: fix={fix_id} fix_review={review_id} " + f"(verdict {verdict}; final review left untouched)" + ), + ) + + # Final fan-in reporter (optional): if the caller named one and we + # have no existing reporter card, gate it behind the fix-review. + # Treated as best-effort linkage; we never duplicate it. + if reporter_assignee: + reporter_key = _IDEMPOTENCY_REPORTER_PREFIX + final_task_id + reporter_id = _existing_reporter_id(conn, final_task_id) + if reporter_id is None: + reporter_id = kb.create_task( + conn, + title=f"fanin-report: {final_task_id}", + body=f"Report back to originator once {review_id} clears.", + assignee=reporter_assignee, + created_by="fanin-resolver", + parents=[review_id], + idempotency_key=reporter_key, + initial_status="running", + ) + ledger["reporter_task_id"] = reporter_id + + ledger["remediation_task_ids"] = {"fix": fix_id, "fix_review": review_id} + ledger["remediation_status"] = "CREATED" + return ledger diff --git a/tests/hermes_cli/test_kanban_resolve_fanin.py b/tests/hermes_cli/test_kanban_resolve_fanin.py new file mode 100644 index 000000000000..65ed9ef8f8b5 --- /dev/null +++ b/tests/hermes_cli/test_kanban_resolve_fanin.py @@ -0,0 +1,468 @@ +"""Tests for ``hermes kanban resolve-fanin`` — the Kanban root-fix slice +that classifies a final review task's verdict / ack / remediation and, +in apply mode, creates one deduped fix card + one fix-review card. + +Bounded scope: see /tmp/t_330e19c2_claude_prompt.md. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from hermes_cli import kanban as kc +from hermes_cli import kanban_db as kb + + +@pytest.fixture +def kanban_home(tmp_path, monkeypatch): + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setattr(Path, "home", lambda: tmp_path) + kb.init_db() + return home + + +# --------------------------------------------------------------------------- +# Helpers — build a small graph: origin -> impl -> final_review (done). +# --------------------------------------------------------------------------- + +def _make_graph(*, verdict: str, ack_status: str | None = None, + unsafe_phrase: str | None = None): + """Create origin → impl → final_review and mark final_review done with + a ``Verdict: `` summary. Returns ``(origin, impl, final)``. + + When ``unsafe_phrase`` is supplied, it is appended to the final + review's result so the safety guard fires. + """ + conn = kb.connect() + try: + origin = kb.create_task(conn, title="origin work") + kb.complete_task(conn, origin, result="origin done") + impl = kb.create_task(conn, title="impl", + parents=[origin]) + kb.complete_task(conn, impl, result="impl done") + final = kb.create_task( + conn, + title="final review", + parents=[impl], + assignee="ccreviewer", + ) + # ready -> done with the verdict baked into summary. + body = f"Verdict: {verdict}\nreview-required: parent worker asked for human review." + if unsafe_phrase: + body += f"\n{unsafe_phrase}" + ok = kb.complete_task(conn, final, result=body, summary=body) + assert ok, "fixture: final review task did not transition to done" + + if ack_status: + # Record ack outcome as a comment with the documented prefix. + kb.add_comment(conn, final, author="gateway-watchdog", + body=f"ack-status: {ack_status}") + finally: + conn.close() + return origin, impl, final + + +# --------------------------------------------------------------------------- +# Classification +# --------------------------------------------------------------------------- + +def test_classify_go_with_ack_failed_is_not_need_more(kanban_home): + """GO verdict + ACK failure must classify as task_verdict=GO, + ack_status=FAILED, remediation_status=NONE. The ack failure is an + operational/delivery issue — it must NOT be misreported as a code + NEED_MORE that triggers a fix card.""" + _, _, final = _make_graph(verdict="GO", ack_status="FAILED") + + out = kc.run_slash(f"resolve-fanin {final} --dry-run --json") + payload = _extract_json(out) + + assert payload["task_verdict"] == "GO" + assert payload["ack_status"] == "FAILED" + assert payload["remediation_status"] == "NONE" + assert payload["remediation_task_ids"] in (None, {}, {"fix": None, "fix_review": None}) + + +def test_classify_reads_verdict_from_run_summary_when_result_empty(kanban_home): + """kanban_complete commonly stores the final ACK text in run + summary/event payload while tasks.result is empty; resolver must + still see the Verdict marker.""" + conn = kb.connect() + try: + origin = kb.create_task(conn, title="origin work") + kb.complete_task(conn, origin, result="origin done") + final = kb.create_task(conn, title="final review", parents=[origin]) + summary = "Verdict: BLOCK\nreview-required: summary-only blocker" + assert kb.complete_task(conn, final, result=None, summary=summary) + finally: + conn.close() + + out = kc.run_slash(f"resolve-fanin {final} --dry-run --json") + payload = _extract_json(out) + assert payload["task_verdict"] == "BLOCK" + assert payload["remediation_status"] == "REQUIRED" + + +def test_dry_run_writes_nothing_on_block(kanban_home): + """A BLOCK verdict in dry-run reports REQUIRED but creates no cards.""" + _, _, final = _make_graph(verdict="BLOCK") + + before_n = _board_task_count() + out = kc.run_slash(f"resolve-fanin {final} --dry-run --json") + after_n = _board_task_count() + + payload = _extract_json(out) + assert payload["task_verdict"] == "BLOCK" + assert payload["remediation_status"] == "REQUIRED" + assert payload["dry_run"] is True + assert payload["remediation_task_ids"] in (None, {"fix": None, "fix_review": None}) + assert after_n == before_n, "dry-run must not insert any cards" + + +def test_apply_block_creates_dedup_fix_and_fix_review(kanban_home): + """BLOCK verdict + --apply creates exactly one fix card and one + fix-review card, both linked to the final task. A second run must + not duplicate either card.""" + _, _, final = _make_graph(verdict="BLOCK") + + out1 = kc.run_slash( + f"resolve-fanin {final} --apply --json " + f"--fix-assignee ccsupervisor --review-assignee ccreviewer" + ) + payload1 = _extract_json(out1) + assert payload1["task_verdict"] == "BLOCK" + assert payload1["remediation_status"] == "CREATED" + fix_id = payload1["remediation_task_ids"]["fix"] + rev_id = payload1["remediation_task_ids"]["fix_review"] + assert fix_id and rev_id and fix_id != rev_id + + conn = kb.connect() + try: + fix = kb.get_task(conn, fix_id) + rev = kb.get_task(conn, rev_id) + assert fix is not None and rev is not None + # Fix-review depends on fix. + assert fix_id in kb.parent_ids(conn, rev_id) + # Sanitization: no raw "review-required:" sentinel leak in bodies. + assert "review-required:" not in (fix.body or "") + assert "review-required:" not in (rev.body or "") + # Assignees flowed through. + assert fix.assignee == "ccsupervisor" + assert rev.assignee == "ccreviewer" + finally: + conn.close() + + # Second run is a no-op for card creation. + out2 = kc.run_slash( + f"resolve-fanin {final} --apply --json " + f"--fix-assignee ccsupervisor --review-assignee ccreviewer" + ) + payload2 = _extract_json(out2) + assert payload2["remediation_task_ids"]["fix"] == fix_id + assert payload2["remediation_task_ids"]["fix_review"] == rev_id + # No third card appeared. + conn = kb.connect() + try: + rows = conn.execute( + "SELECT id FROM tasks WHERE id IN (?, ?)", (fix_id, rev_id) + ).fetchall() + assert len(rows) == 2 + # Total task count grew by exactly 2 over the original graph. + # origin + impl + final + fix + fix_review = 5. + total = conn.execute("SELECT COUNT(*) AS n FROM tasks").fetchone()["n"] + assert total == 5 + finally: + conn.close() + + +def test_apply_unsafe_blocker_refuses_card_creation(kanban_home): + """A BLOCK verdict whose review body mentions an unsafe sentinel + category (secret leak, destructive data loss, auth/credential, user + input required, live money/trading) must classify as + remediation_status=BLOCKED and create no cards even in --apply.""" + _, _, final = _make_graph( + verdict="BLOCK", + unsafe_phrase="reason: leaked SECRET_KEY in commit", + ) + + before_n = _board_task_count() + out = kc.run_slash(f"resolve-fanin {final} --apply --json") + after_n = _board_task_count() + payload = _extract_json(out) + + assert payload["task_verdict"] == "BLOCK" + assert payload["remediation_status"] == "BLOCKED" + assert payload["remediation_task_ids"] in (None, {"fix": None, "fix_review": None}) + assert after_n == before_n + + +def test_apply_need_more_creates_fix_with_need_more_marker(kanban_home): + """NEED_MORE verdict also drives remediation; created cards must + reflect that this came from NEED_MORE (not generic BLOCK).""" + _, _, final = _make_graph(verdict="NEED_MORE") + + out = kc.run_slash(f"resolve-fanin {final} --apply --json") + payload = _extract_json(out) + + assert payload["task_verdict"] == "NEED_MORE" + assert payload["remediation_status"] == "CREATED" + assert payload["remediation_task_ids"]["fix"] + assert payload["remediation_task_ids"]["fix_review"] + + +def test_apply_can_gate_final_reporter_behind_fix_review(kanban_home): + """When requested, apply mode creates one deduped final fan-in + reporter gated behind the fix-review card and reports it in the + machine-readable ledger.""" + _, _, final = _make_graph(verdict="BLOCK") + + out1 = kc.run_slash( + f"resolve-fanin {final} --apply --json " + f"--fix-assignee ccsupervisor --review-assignee ccreviewer " + f"--reporter-assignee ccsupervisor" + ) + payload1 = _extract_json(out1) + review_id = payload1["remediation_task_ids"]["fix_review"] + reporter_id = payload1["reporter_task_id"] + assert reporter_id + + conn = kb.connect() + try: + reporter = kb.get_task(conn, reporter_id) + assert reporter is not None + assert reporter.assignee == "ccsupervisor" + assert review_id in kb.parent_ids(conn, reporter_id) + finally: + conn.close() + + out2 = kc.run_slash( + f"resolve-fanin {final} --apply --json " + f"--fix-assignee ccsupervisor --review-assignee ccreviewer " + f"--reporter-assignee ccsupervisor" + ) + payload2 = _extract_json(out2) + assert payload2["reporter_task_id"] == reporter_id + assert _board_task_count() == 6 + + +def test_apply_respects_max_fan_in_threshold(kanban_home): + """The apply path is bounded by transitive graph size so a single + command cannot silently spawn remediation for an unexpectedly large + fan-in. This uses the real temp kanban DB fixture and verifies no + cards are inserted when the bound is exceeded.""" + _, _, final = _make_graph(verdict="BLOCK") + + before_n = _board_task_count() + out = kc.run_slash( + f"resolve-fanin {final} --apply --json --max-fan-in-threshold 2" + ) + after_n = _board_task_count() + payload = _extract_json(out) + + assert payload["graph_task_count"] == 3 + assert payload["max_fan_in_threshold"] == 2 + assert payload["remediation_status"] == "BLOCKED" + assert "exceeds threshold 2" in payload["blocked_reason"] + assert payload["remediation_task_ids"] in ( + None, + {"fix": None, "fix_review": None}, + ) + assert after_n == before_n + + +def test_apply_allows_explicitly_raised_fan_in_threshold(kanban_home): + """Operators can raise the threshold after inspecting a larger graph; + the real DB apply path then creates the deduped remediation pair.""" + _, _, final = _make_graph(verdict="BLOCK") + + out = kc.run_slash( + f"resolve-fanin {final} --apply --json --max-fan-in-threshold 3" + ) + payload = _extract_json(out) + + assert payload["graph_task_count"] == 3 + assert payload["max_fan_in_threshold"] == 3 + assert payload["remediation_status"] == "CREATED" + assert payload["remediation_task_ids"]["fix"] + assert payload["remediation_task_ids"]["fix_review"] + + +def test_invalid_max_fan_in_threshold_fails_before_writes(kanban_home): + _, _, final = _make_graph(verdict="BLOCK") + + before_n = _board_task_count() + out = kc.run_slash( + f"resolve-fanin {final} --apply --json --max-fan-in-threshold 0" + ) + after_n = _board_task_count() + payload = _extract_json(out) + + assert "max_fan_in_threshold must be >= 1" in payload["error"] + assert after_n == before_n + + +# --------------------------------------------------------------------------- +# Internals — ledger shape +# --------------------------------------------------------------------------- + +def test_ledger_carries_graph_and_origin_return(kanban_home): + origin, impl, final = _make_graph(verdict="GO") + out = kc.run_slash(f"resolve-fanin {final} --dry-run --json") + payload = _extract_json(out) + assert payload["final_task"] == final + assert payload["board"] # current board slug + assert set(payload["graph_task_ids"]) >= {origin, impl, final} + assert payload["origin_return"] == origin + + +# --------------------------------------------------------------------------- +# Blocked review-required final — remediation must stay runnable +# --------------------------------------------------------------------------- + +def _make_blocked_graph(*, verdict: str = "BLOCK"): + """origin(done) -> impl(done) -> final(**blocked** review-required). + + Mirrors a sticky ``blocked`` review-required verdict the resolver must + preserve. Returns ``(origin, impl, final)``. + """ + conn = kb.connect() + try: + origin = kb.create_task(conn, title="origin work") + kb.complete_task(conn, origin, result="origin done") + impl = kb.create_task(conn, title="impl", parents=[origin]) + kb.complete_task(conn, impl, result="impl done") + final = kb.create_task( + conn, title="final review", parents=[impl], assignee="ccreviewer" + ) + reason = ( + f"Verdict: {verdict}\n" + "review-required: worker handed back for human review." + ) + assert kb.block_task(conn, final, reason=reason), ( + "fixture: final review task did not transition to blocked" + ) + assert kb.get_task(conn, final).status == "blocked" + finally: + conn.close() + return origin, impl, final + + +def test_apply_on_blocked_final_creates_runnable_remediation( + kanban_home, monkeypatch +): + """Regression: when the final review is a sticky ``blocked`` + review-required verdict, ``resolve-fanin --apply`` must create a + *runnable* remediation graph. + + Previously the fix card was parented to the blocked final, so it stuck + in ``todo`` (a child cannot reach ``ready`` until every parent is + ``done``) and the dispatcher — which only claims ``ready`` tasks — + could never run it. The blocked final's verdict/status must stay + untouched; the link back to it is preserved as an audit comment, not a + dependency edge. + """ + origin, impl, final = _make_blocked_graph(verdict="BLOCK") + + out = kc.run_slash( + f"resolve-fanin {final} --apply --json " + f"--fix-assignee ccsupervisor --review-assignee ccreviewer" + ) + payload = _extract_json(out) + assert payload["task_verdict"] == "BLOCK" + assert payload["remediation_status"] == "CREATED" + fix_id = payload["remediation_task_ids"]["fix"] + rev_id = payload["remediation_task_ids"]["fix_review"] + assert fix_id and rev_id and fix_id != rev_id + + conn = kb.connect() + try: + fix = kb.get_task(conn, fix_id) + rev = kb.get_task(conn, rev_id) + # The core of the fix: the fix card is dispatchable (``ready``) and + # does NOT depend on the blocked final. + assert fix.status == "ready", ( + f"fix card must be dispatchable, got {fix.status!r}" + ) + assert final not in kb.parent_ids(conn, fix_id) + # fix-review is still gated behind the fix card. + assert kb.parent_ids(conn, rev_id) == [fix_id] + assert rev.status == "todo" + # Audit trail preserved as a comment on the final task. + comments = [c.body for c in kb.list_comments(conn, final)] + assert any(fix_id in b and rev_id in b for b in comments), ( + f"expected a remediation-created audit comment, got {comments!r}" + ) + # The blocked final verdict/status is untouched. + assert kb.get_task(conn, final).status == "blocked" + finally: + conn.close() + + # ---- Dispatcher/claim semantics: the remediation graph actually runs. ---- + import hermes_cli.profiles as _profiles + + monkeypatch.setattr(_profiles, "profile_exists", lambda name: True) + claimed: list[str] = [] + + def _spawn(task, workspace_path, board): + claimed.append(task.id) + return 4321 # fake worker pid + + conn = kb.connect() + try: + kb.dispatch_once(conn, spawn_fn=_spawn) + assert fix_id in claimed, "dispatcher must claim the ready fix card" + assert kb.get_task(conn, fix_id).status == "running" + + # Complete the fix -> the fix-review promotes todo -> ready. + assert kb.complete_task( + conn, fix_id, result="fix applied", summary="Verdict: GO" + ) + assert kb.get_task(conn, rev_id).status == "ready" + + # Second tick claims the now-ready fix-review; complete it. + claimed.clear() + kb.dispatch_once(conn, spawn_fn=_spawn) + assert rev_id in claimed, "dispatcher must claim the ready fix-review" + assert kb.complete_task( + conn, rev_id, result="fix verified", summary="Verdict: GO" + ) + assert kb.get_task(conn, rev_id).status == "done" + + # Semantic resolution: the remediation pair ran to completion while + # the original blocked final verdict/status is unchanged. + assert kb.get_task(conn, final).status == "blocked" + finally: + conn.close() + + # Re-classifying the final still reports BLOCK — its verdict was never + # mutated to GO/done by the remediation flow. + payload2 = _extract_json(kc.run_slash(f"resolve-fanin {final} --dry-run --json")) + assert payload2["task_verdict"] == "BLOCK" + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _extract_json(out: str) -> dict: + # ``run_slash`` returns captured stdout — the CLI emits one JSON + # document when ``--json`` is set. + out = out.strip() + # Tolerate a trailing newline / extra noise: find the first ``{`` and + # last ``}``. + start = out.find("{") + end = out.rfind("}") + assert start != -1 and end != -1, f"no JSON in output: {out!r}" + return json.loads(out[start:end + 1]) + + +def _board_task_count() -> int: + conn = kb.connect() + try: + return conn.execute("SELECT COUNT(*) AS n FROM tasks").fetchone()["n"] + finally: + conn.close()