diff --git a/hooks/hooks.json b/hooks/hooks.json index 977e8c4..288c07f 100644 --- a/hooks/hooks.json +++ b/hooks/hooks.json @@ -10,6 +10,16 @@ } ] } + ], + "Stop": [ + { + "hooks": [ + { + "type": "command", + "command": "python \"${CLAUDE_PLUGIN_ROOT}/hooks/stop_gate.py\"" + } + ] + } ] } } diff --git a/hooks/stop_gate.py b/hooks/stop_gate.py new file mode 100644 index 0000000..d7a7269 --- /dev/null +++ b/hooks/stop_gate.py @@ -0,0 +1,401 @@ +#!/usr/bin/env python3 +"""Stop-gate hook — refuse to finish while UI changes are unverified. + +This is the load-bearing correctness of Phase 3 (see ``planning/PHASE_3.md``). +A ``Stop`` hook fires on *every* turn end. On each firing it reads the +session-scoped state the ``PostToolUse`` flagger writes and the verification +result (if any) and routes the three outcomes: + +* **UI touched, no covering verify-result yet** -> *block*, naming the files to + verify. Claude gets the reason and keeps working. +* **``outcome == "fail"``** -> *block*, reason = the expected-vs-actual + ``detail`` so the agent self-corrects, then re-verifies. +* **``outcome == "pass"`` covering the touched set** -> *allow*. +* **``outcome == "abstain"`` covering the touched set** -> *allow, and surface + the abstain ``detail`` to the user*. **This is the single most important + rule.** Abstain is a legitimate stopping point that escalates -- an + unverifiable change must not be blocked. If it were, the change would thrash + into Claude Code's 8-consecutive-block cap and then false-pass anyway, which + is the exact failure this whole phase exists to prevent. + +The gate is satisfied only when ``covered`` covers ``ui_touched``; a later edit +to a *new* UI file re-opens it (block again, naming just the new file). + +Constraints honoured from the hook contract: + +* **Idempotent / never nag.** A turn with nothing to verify allows silently. +* **``stop_hook_active``-aware.** Re-entry is expected; the bounded retry below + guarantees the block budget is never spun. +* **8-block-cap-safe.** A correct->verify cycle that cannot be satisfied is + capped at :data:`DEFAULT_RETRY_CAP` blocks (well under 8); on exhaustion the + gate *escalates* -- allows the stop and surfaces a clear "could not satisfy + after N attempts" message -- rather than looping invisibly. + +Deliberately **stdlib-only and self-contained**: it must run as +``python ${CLAUDE_PLUGIN_ROOT}/hooks/stop_gate.py`` with no dependency on the +``cyclaudes`` package being importable from the hook process. The core is +factored into :func:`decide` so every branch is unit-testable against +``tmp_path`` state files; ``__main__`` is a thin stdin/stdout/exit shim. +""" + +from __future__ import annotations + +import json +import os +import sys + +#: Root of the git-ignored session state, relative to the project directory. +#: Must match the FROZEN INTERFACE in planning/PHASE_3.md and what issue A writes. +STATE_DIRNAME = ".cyclaudes" +PENDING_SUBDIR = "pending-ui" +RESULT_SUBDIR = "verify-result" +#: Our own bookkeeping (retry counters, audit counters); not part of the frozen +#: cross-issue contract -- only this hook reads and writes it. +GATE_SUBDIR = "gate-state" +COUNTERS_FILE = "counters.json" + +#: Default cap on correct->verify cycles before the gate escalates instead of +#: blocking again. Kept well under Claude Code's 8-consecutive-block override so +#: an unsatisfiable check escalates *loudly* rather than being force-passed by +#: the platform. Overridable via ``CYCLAUDES_RETRY_CAP`` for tuning. +DEFAULT_RETRY_CAP = 3 + + +# --------------------------------------------------------------------------- +# state IO (stdlib-only, defensive: a malformed/absent file reads as "nothing") +# --------------------------------------------------------------------------- + + +def _state_dir(project_dir: str) -> str: + return os.path.join(project_dir, STATE_DIRNAME) + + +def _read_json(path: str): + try: + with open(path, encoding="utf-8") as fh: + return json.load(fh) + except (OSError, ValueError): + return None + + +def _write_json(path: str, data) -> None: + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as fh: + json.dump(data, fh, indent=2, sort_keys=True) + + +def _norm(relpath: str) -> str: + """Normalise a state-file relpath for set comparison (separators only). + + Both sides of the coverage check come from the same producers writing + repo-relative paths, so normalising ``\\`` -> ``/`` and trimming is enough; + case is left untouched to avoid guessing the filesystem's semantics. + """ + return str(relpath).replace("\\", "/").strip() + + +def read_pending(project_dir: str, session_id: str) -> set[str]: + """The de-duplicated UI-touched set for *session_id* (empty if none).""" + if not session_id: + return set() + data = _read_json( + os.path.join(_state_dir(project_dir), PENDING_SUBDIR, f"{session_id}.json") + ) + if not isinstance(data, dict): + return set() + touched = data.get("ui_touched") or [] + return {_norm(p) for p in touched if str(p).strip()} + + +def read_verify_result(project_dir: str, session_id: str): + """The verify-result record for *session_id*, or ``None`` if absent/foreign. + + Validates the ``session_id`` inside the file against the one keying the + path, so a stray or mismatched record can never satisfy the wrong session's + gate. + """ + if not session_id: + return None + data = _read_json( + os.path.join(_state_dir(project_dir), RESULT_SUBDIR, f"{session_id}.json") + ) + if not isinstance(data, dict): + return None + if data.get("session_id") not in (None, session_id): + return None + return data + + +# --------------------------------------------------------------------------- +# our own gate bookkeeping (retry budget + audit counters) +# --------------------------------------------------------------------------- + + +def _gate_path(project_dir: str, session_id: str) -> str: + return os.path.join(_state_dir(project_dir), GATE_SUBDIR, f"{session_id}.json") + + +def load_gate_state(project_dir: str, session_id: str) -> dict: + data = _read_json(_gate_path(project_dir, session_id)) + if not isinstance(data, dict): + return {"session_id": session_id, "block_count": 0, "last_counted_at": None} + data.setdefault("session_id", session_id) + data.setdefault("block_count", 0) + data.setdefault("last_counted_at", None) + return data + + +def save_gate_state(project_dir: str, session_id: str, state: dict) -> None: + _write_json(_gate_path(project_dir, session_id), state) + + +def _reset_block_count(project_dir: str, session_id: str) -> None: + state = load_gate_state(project_dir, session_id) + if state.get("block_count"): + state["block_count"] = 0 + save_gate_state(project_dir, session_id, state) + + +def bump_counter(project_dir: str, name: str) -> None: + """Persist a pass/fail/abstain/block/escalate tally for later audit. + + The point (planning/PHASE_3.md, "Key risk") is that early runs can be + checked for *swallowed abstentions*: if a project accrued unverifiable UI + changes but the ``abstain`` tally is zero, something quietly reclassified + them. Counters live at ``.cyclaudes/counters.json`` (project-wide). + """ + path = os.path.join(_state_dir(project_dir), COUNTERS_FILE) + data = _read_json(path) + if not isinstance(data, dict): + data = {} + data[name] = int(data.get(name, 0)) + 1 + _write_json(path, data) + + +def _count_outcome(project_dir: str, session_id: str, outcome: str, at) -> None: + """Tally a *verification outcome* once, de-duped by its ``at`` timestamp. + + The Stop hook re-reads the same verify-result on every turn end; without + de-duping, a single fail blocked across three turns would inflate the audit + counters threefold. Keyed on the result's ``at`` so each distinct + verification is counted exactly once. + """ + state = load_gate_state(project_dir, session_id) + if at is not None and state.get("last_counted_at") == at: + return + state["last_counted_at"] = at + save_gate_state(project_dir, session_id, state) + bump_counter(project_dir, outcome) + + +# --------------------------------------------------------------------------- +# decision core +# --------------------------------------------------------------------------- + + +def _retry_cap() -> int: + try: + cap = int(os.environ.get("CYCLAUDES_RETRY_CAP", DEFAULT_RETRY_CAP)) + except (TypeError, ValueError): + return DEFAULT_RETRY_CAP + return cap if cap > 0 else DEFAULT_RETRY_CAP + + +def _allow(reason=None, system_message=None, escalated=False) -> dict: + return { + "decision": "allow", + "reason": reason, + "system_message": system_message, + "escalated": escalated, + } + + +def _unverified_reason(files: list[str], *, have_result: bool) -> str: + listed = ", ".join(files) + lead = ( + "UI changes have not been verified yet." + if not have_result + else "New UI changes are not covered by the last verification." + ) + return ( + f"{lead} These edited UI file(s) need a verification run before you can " + f"stop: {listed}. Run the cyclaudes UI checks that cover them (e.g. " + f"`cyclaudes verify`), or invoke the verify-ui skill to author and run " + f"the checks. If the change genuinely cannot be verified, record an " + f"abstain outcome (CannotVerify) -- that is a legitimate stopping point, " + f"not a reason to guess." + ) + + +def _fail_reason(detail: str) -> str: + return ( + "UI verification FAILED. Self-correct, then re-verify. " + f"Expected-vs-actual:\n{detail}" + ) + + +def _abstain_message(covered: list[str], detail: str) -> str: + listed = ", ".join(covered) if covered else "the touched UI files" + return ( + "UI verification ABSTAINED (could not verify) for " + f"{listed}. This is a legitimate stopping point that needs a human's " + f"eyes -- it is NOT a pass and NOT a failure. What the check could not " + f"determine:\n{detail}" + ) + + +def _exhaustion_message(cap: int, last_reason: str) -> str: + return ( + f"Could not satisfy the UI checks after {cap} attempts. Escalating and " + f"allowing the stop rather than blocking further (the block budget is " + f"bounded on purpose). Last blocking reason was:\n{last_reason}" + ) + + +def _gate_block( + project_dir: str, + session_id: str, + *, + reason: str, + result_at=None, + fail_at=None, +) -> dict: + """Emit a block, unless the bounded-retry budget is exhausted -> escalate. + + ``fail_at`` (the verify-result ``at`` of a *fail* being blocked on) lets the + audit counter tally the failure exactly once even while it is blocked across + several turns. + """ + if fail_at is not None: + _count_outcome(project_dir, session_id, "fail", fail_at) + + cap = _retry_cap() + state = load_gate_state(project_dir, session_id) + count = int(state.get("block_count", 0)) + + if count >= cap: + # Exhausted. Escalate rather than burn another block -- this is what + # keeps an unsatisfiable check from thrashing into the 8-block cap and + # false-passing there. Reset so a genuinely new change starts fresh. + state["block_count"] = 0 + save_gate_state(project_dir, session_id, state) + bump_counter(project_dir, "escalate") + return _allow(system_message=_exhaustion_message(cap, reason), escalated=True) + + state["block_count"] = count + 1 + save_gate_state(project_dir, session_id, state) + bump_counter(project_dir, "block") + return { + "decision": "block", + "reason": reason, + "system_message": None, + "escalated": False, + } + + +def decide(payload: dict, project_dir: str) -> dict: + """Route one Stop event. Pure w.r.t. its inputs + the ``.cyclaudes`` state. + + :param payload: the Stop hook stdin JSON -- ``session_id``, + ``stop_hook_active`` (and ``cwd``, used by ``main`` to locate the + project). Re-entry (``stop_hook_active``) needs no special branch: the + bounded retry below already guarantees the block budget cannot spin. + :param project_dir: the project root that owns ``.cyclaudes/``. + :returns: ``{"decision": "block"|"allow", "reason", "system_message", + "escalated"}``. ``reason`` is Claude-facing (the block channel); + ``system_message`` is the user-facing escalation surfaced on an allow. + """ + payload = payload or {} + session_id = payload.get("session_id") or "" + + ui_touched = read_pending(project_dir, session_id) + + # Nothing to verify -> never nag. The idempotent, cheap-to-satisfy path that + # keeps non-UI turns (and re-entries with nothing pending) instant. + if not ui_touched: + _reset_block_count(project_dir, session_id) + return _allow() + + result = read_verify_result(project_dir, session_id) + outcome = (result or {}).get("outcome") + covered = {_norm(p) for p in ((result or {}).get("covered") or [])} + uncovered = ui_touched - covered + + # A fail outranks everything: the agent must self-correct before stopping, + # even if it has meanwhile touched further files. Reason carries the diff. + if result and outcome == "fail": + detail = result.get("detail") or "UI verification failed (no detail recorded)." + return _gate_block( + project_dir, + session_id, + reason=_fail_reason(detail), + fail_at=result.get("at"), + ) + + # No result yet, or a newly-edited UI file the last verification did not + # cover -> block, naming exactly what still needs verifying (re-opens gate). + if uncovered: + return _gate_block( + project_dir, + session_id, + reason=_unverified_reason(sorted(uncovered), have_result=bool(result)), + ) + + # Fully covered by the latest verification. + if outcome == "abstain": + # THE most important rule: abstain satisfies the gate AND escalates. + _reset_block_count(project_dir, session_id) + _count_outcome(project_dir, session_id, "abstain", result.get("at")) + detail = result.get("detail") or "(no reason recorded)" + return _allow( + system_message=_abstain_message(sorted(covered), detail), escalated=True + ) + + if outcome == "pass": + _reset_block_count(project_dir, session_id) + _count_outcome(project_dir, session_id, "pass", result.get("at")) + return _allow() + + # Defensive: a result with full coverage but an outcome we don't recognise. + # Treat as unverified rather than assuming success. + return _gate_block( + project_dir, + session_id, + reason=_unverified_reason(sorted(ui_touched), have_result=bool(result)), + ) + + +# --------------------------------------------------------------------------- +# __main__ -- thin stdin / stdout-JSON / exit-code shim +# --------------------------------------------------------------------------- + + +def main(argv=None) -> int: + try: + payload = json.load(sys.stdin) + except (ValueError, OSError): + payload = {} + if not isinstance(payload, dict): + payload = {} + + project_dir = payload.get("cwd") or os.getcwd() + outcome = decide(payload, project_dir) + + if outcome["decision"] == "block": + # JSON decision mode: Claude receives `reason` and keeps working. + print(json.dumps({"decision": "block", "reason": outcome["reason"]})) + return 0 + + # Allow. Surface any escalation (abstain / retry-exhaustion) to the user via + # `systemMessage` -- the Stop hook cannot inject additionalContext, so this + # is the escalation channel. + body: dict = {} + if outcome.get("system_message"): + body["systemMessage"] = outcome["system_message"] + if body: + print(json.dumps(body)) + return 0 + + +if __name__ == "__main__": # pragma: no cover - exercised via subprocess/tests + sys.exit(main()) diff --git a/planning/TODO.md b/planning/TODO.md index b8b1875..910734b 100644 --- a/planning/TODO.md +++ b/planning/TODO.md @@ -112,9 +112,9 @@ Tightly coupled; best done by **one agent**, not fanned out. repo-relative path to `.cyclaudes/pending-ui/.json` per the frozen schema (`planning/PHASE_3.md`) the Stop hook (issue #32) reads. No-op on a non-UI path; never blocks the tool call. `tests/test_flag_ui_change.py`) -- [ ] Loop integration: pass → continue; fail → actionable diff + self-correct; abstain → +- [x] Loop integration: pass → continue; fail → actionable diff + self-correct; abstain → escalate with specifics -- [ ] Bounded retry — cap correct→verify cycles, escalate on exhaustion +- [x] Bounded retry — cap correct→verify cycles, escalate on exhaustion - [ ] Success criterion: a full issue resolution completes with zero Cameron input ## Phase 4 — Vision fallback → Phase 3 diff --git a/pyproject.toml b/pyproject.toml index f08818f..8788e7a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,6 +15,11 @@ dependencies = ["touchpoint-py>=0.3.0", "pytest>=8.0"] cyclaudes = "cyclaudes.pytest_plugin" cyclaudes_ui = "cyclaudes.pytest_ui" +# `cyclaudes verify` records verify-result/.json when UI checks run +# (Phase 3, deliverable B) — the input the Stop-gate hook routes on. +[project.scripts] +cyclaudes = "cyclaudes.verify_result:main" + [build-system] requires = ["hatchling"] build-backend = "hatchling.build" diff --git a/src/cyclaudes/verify_result.py b/src/cyclaudes/verify_result.py new file mode 100644 index 0000000..e2591ad --- /dev/null +++ b/src/cyclaudes/verify_result.py @@ -0,0 +1,283 @@ +"""Write ``verify-result/.json`` when UI checks run. + +The Stop gate (``hooks/stop_gate.py``) can only route the three outcomes if +*something* records them per session, against the FROZEN INTERFACE in +``planning/PHASE_3.md``:: + + { "session_id": "...", "outcome": "pass|fail|abstain", + "covered": ["relpath/one.tsx"], + "detail": "expected-vs-actual, or the abstain reason", + "at": "" } + +This module is that writer, delivered as a thin ``cyclaudes verify`` CLI +(chosen over a pytest-``sessionfinish`` hook so an *ordinary* pytest run never +writes a result -- only an explicit verification does, which keeps the gate's +input unambiguous). It runs the project's cyclaudes checks with ``pytest`` and +maps the run to an outcome. + +The outcome mapping is **exactly** the three-outcome semantics already defined +in :mod:`cyclaudes.abstain` / :mod:`cyclaudes.pytest_plugin` -- it reuses their +process exit codes rather than re-deriving them, so a real failure can never be +reclassified as an abstention or vice versa: + +* any UI-check failure -> ``fail`` (pytest exit ``1``) +* any abstention, no failure -> ``abstain`` (pytest exit ``EXIT_ABSTAINED`` = 12) +* all checks pass -> ``pass`` (pytest exit ``0``) + +The writing and outcome-mapping are factored into importable, side-effect-free +functions (:func:`map_exit_code`, :func:`classify`, :func:`write_result`) so +they are testable without a subprocess or an editable install; the CLI is a +thin driver over them. +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +from datetime import datetime, timezone + +from .abstain import EXIT_ABSTAINED + +STATE_DIRNAME = ".cyclaudes" +PENDING_SUBDIR = "pending-ui" +RESULT_SUBDIR = "verify-result" + +#: The only valid outcomes, matching the frozen schema and the three-outcome +#: discipline the rest of the package enforces. +OUTCOMES = ("pass", "fail", "abstain") + + +# --------------------------------------------------------------------------- +# pure outcome mapping (mirrors cyclaudes.pytest_plugin's exit-code decision) +# --------------------------------------------------------------------------- + + +def map_exit_code(exit_code: int) -> str: + """Map a cyclaudes pytest run's exit code to a frozen-schema outcome. + + Reuses the exact codes :mod:`cyclaudes.pytest_plugin` produces: ``0`` is an + all-pass, :data:`~cyclaudes.abstain.EXIT_ABSTAINED` (12) is "only + abstentions, nothing broken", and everything else (a real failure ``1``, a + collection/usage error, ...) is a ``fail``. Erring toward ``fail`` for the + odd exit codes is the safe direction: a verification that did not cleanly + pass or cleanly abstain has not verified anything. + """ + if exit_code == 0: + return "pass" + if exit_code == EXIT_ABSTAINED: + return "abstain" + return "fail" + + +def classify(failures, abstentions) -> tuple[str, str]: + """Derive ``(outcome, detail)`` from collected failures and abstentions. + + Failure outranks abstention (a known-broken result is the more urgent + signal -- the same precedence :mod:`cyclaudes.pytest_plugin` uses for the + exit code), so the mapping is: any *failure* -> ``fail`` with the + expected-vs-actual text; else any *abstention* -> ``abstain`` with the + reasons; else ``pass``. + + :param failures: iterable of ``(nodeid, detail_text)``. + :param abstentions: iterable of ``(nodeid, reason)``. + """ + failures = list(failures) + abstentions = list(abstentions) + if failures: + detail = "\n\n".join( + f"{nodeid}\n{text}".rstrip() for nodeid, text in failures + ) + return "fail", detail + if abstentions: + detail = "; ".join(f"{nodeid}: {reason}" for nodeid, reason in abstentions) + return "abstain", detail + return "pass", "" + + +# --------------------------------------------------------------------------- +# the writer +# --------------------------------------------------------------------------- + + +def _state_dir(project_dir: str) -> str: + return os.path.join(project_dir, STATE_DIRNAME) + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +def write_result( + project_dir: str, + session_id: str, + outcome: str, + covered, + detail: str, + *, + at: str | None = None, +) -> str: + """Write the frozen-schema ``verify-result/.json``; return path. + + :raises ValueError: if *outcome* is not one of :data:`OUTCOMES`. Refusing an + unknown outcome here keeps a typo from writing a record the gate would + then have to guess about. + """ + if outcome not in OUTCOMES: + raise ValueError(f"outcome must be one of {OUTCOMES}, got {outcome!r}") + record = { + "session_id": session_id, + "outcome": outcome, + "covered": list(covered), + "detail": detail, + "at": at or _now_iso(), + } + path = os.path.join(_state_dir(project_dir), RESULT_SUBDIR, f"{session_id}.json") + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as fh: + json.dump(record, fh, indent=2, sort_keys=True) + return path + + +def read_covered_from_pending(project_dir: str, session_id: str) -> list[str]: + """The UI files this session touched -- the default ``covered`` set. + + A verification run addresses whatever UI files are pending for the session, + so absent an explicit ``--covered`` the touched set is exactly what the run + covers. Returns ``[]`` if there is no pending-ui record. + """ + path = os.path.join( + _state_dir(project_dir), PENDING_SUBDIR, f"{session_id}.json" + ) + try: + with open(path, encoding="utf-8") as fh: + data = json.load(fh) + except (OSError, ValueError): + return [] + if not isinstance(data, dict): + return [] + return [str(p) for p in (data.get("ui_touched") or []) if str(p).strip()] + + +# --------------------------------------------------------------------------- +# CLI: `cyclaudes verify [-- ]` +# --------------------------------------------------------------------------- + + +class _ResultCollector: + """Pytest plugin that captures failure detail and abstention reasons. + + Reads the markers :mod:`cyclaudes.pytest_plugin` stamps on each report so an + abstention (which that plugin represents as ``report.outcome == "failed"`` + with ``cyclaudes_abstained = True``) is never miscounted as a failure. The + *outcome* itself is taken from the process exit code (:func:`map_exit_code`) + -- authoritative and already correct -- and this collector supplies only the + human-readable ``detail`` for whichever outcome that turns out to be. + """ + + def __init__(self) -> None: + self.failures: list[tuple[str, str]] = [] + self.abstentions: list[tuple[str, str]] = [] + + def pytest_runtest_logreport(self, report) -> None: + if getattr(report, "cyclaudes_abstained", False): + reason = getattr(report, "cyclaudes_abstain_reason", None) + if not reason: + reason = _reason_from_properties(report) or "(no reason recorded)" + self.abstentions.append((report.nodeid, reason)) + elif report.failed: + where = "" if report.when == "call" else f" [{report.when}]" + self.failures.append((report.nodeid + where, report.longreprtext)) + + +def _reason_from_properties(report) -> str | None: + for key, value in getattr(report, "user_properties", []): + if key.startswith("cyclaudes_reason"): + return value + return None + + +def _resolve_session(project_dir: str, explicit: str | None) -> str: + """The session id: explicit flag, else env, else the sole pending session.""" + if explicit: + return explicit + env = os.environ.get("CLAUDE_SESSION_ID") + if env: + return env + pending_dir = os.path.join(_state_dir(project_dir), PENDING_SUBDIR) + try: + candidates = [f for f in os.listdir(pending_dir) if f.endswith(".json")] + except OSError: + candidates = [] + if len(candidates) == 1: + return candidates[0][: -len(".json")] + raise SystemExit( + "cyclaudes verify: could not determine the session id. Pass --session, " + "set CLAUDE_SESSION_ID, or ensure exactly one pending-ui record exists " + f"(found {len(candidates)})." + ) + + +def _cmd_verify(args) -> int: + import pytest + + project_dir = args.project_dir or os.getcwd() + session_id = _resolve_session(project_dir, args.session) + + collector = _ResultCollector() + ret = pytest.main(list(args.pytest_args), plugins=[collector]) + ret = int(ret) + + outcome = map_exit_code(ret) + # The exit code is authoritative for the outcome; the collector supplies the + # detail. classify() is the fallback when the collector saw the reports + # (they always agree), but map_exit_code wins so the CLI stays consistent + # with pytest's own abstain-vs-fail decision even on odd exit codes. + _, detail = classify(collector.failures, collector.abstentions) + if outcome == "fail" and not detail: + detail = f"pytest exited {ret} with no per-test detail collected." + + covered = args.covered or read_covered_from_pending(project_dir, session_id) + write_result(project_dir, session_id, outcome, covered, detail) + return ret + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="cyclaudes") + sub = parser.add_subparsers(dest="command", required=True) + + verify = sub.add_parser( + "verify", + help="run the cyclaudes UI checks and record verify-result/.json", + ) + verify.add_argument( + "--session", help="session id (default: $CLAUDE_SESSION_ID or the sole pending)" + ) + verify.add_argument( + "--project-dir", help="project root owning .cyclaudes/ (default: cwd)" + ) + verify.add_argument( + "--covered", + nargs="*", + help="UI files this run covers (default: the session's pending ui_touched)", + ) + verify.add_argument( + "pytest_args", + nargs=argparse.REMAINDER, + help="args passed through to pytest (put them after `--`)", + ) + verify.set_defaults(func=_cmd_verify) + return parser + + +def main(argv=None) -> int: + args = build_parser().parse_args(argv) + # argparse REMAINDER keeps a leading `--`; drop it so pytest sees clean args. + if getattr(args, "pytest_args", None) and args.pytest_args[:1] == ["--"]: + args.pytest_args = args.pytest_args[1:] + return args.func(args) + + +if __name__ == "__main__": # pragma: no cover - exercised via the console script + sys.exit(main()) diff --git a/tests/test_stop_gate.py b/tests/test_stop_gate.py new file mode 100644 index 0000000..0e50471 --- /dev/null +++ b/tests/test_stop_gate.py @@ -0,0 +1,342 @@ +"""The Stop gate must route the three outcomes deterministically. + +The load-bearing correctness of Phase 3 (planning/PHASE_3.md). Every branch of +:func:`stop_gate.decide` is exercised against real ``tmp_path`` state files -- +the same ``.cyclaudes/`` layout the PostToolUse flagger and the verify-result +writer produce -- so the assertions are against the actual routing an agent +would get, not the hook's own bookkeeping. + +The single most important test here is +:func:`test_abstain_allows_and_escalates_does_not_block`: if abstain ever +blocked, an unverifiable change would thrash into Claude Code's 8-block cap and +false-pass there, which is the exact failure the whole phase exists to prevent. + +The hook ships as a standalone ``python ${CLAUDE_PLUGIN_ROOT}/hooks/stop_gate.py`` +script (stdlib-only, no dependency on cyclaudes being importable), so it is +loaded here by file path rather than imported as a package module. +""" + +from __future__ import annotations + +import importlib.util +import json +import pathlib + +import pytest + +_HOOK_PATH = pathlib.Path(__file__).resolve().parents[1] / "hooks" / "stop_gate.py" +_spec = importlib.util.spec_from_file_location("cyclaudes_stop_gate", _HOOK_PATH) +stop_gate = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(stop_gate) + +SID = "sess-123" + + +# --------------------------------------------------------------------------- +# helpers: write the frozen-schema state files under /.cyclaudes/ +# --------------------------------------------------------------------------- + + +def _write_pending(project_dir: pathlib.Path, session_id: str, ui_touched: list[str]): + path = project_dir / ".cyclaudes" / "pending-ui" / f"{session_id}.json" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps({"session_id": session_id, "ui_touched": ui_touched}), + encoding="utf-8", + ) + + +def _write_result( + project_dir: pathlib.Path, + session_id: str, + outcome: str, + covered: list[str], + detail: str, + at: str = "2026-07-22T12:00:00+00:00", +): + path = project_dir / ".cyclaudes" / "verify-result" / f"{session_id}.json" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps( + { + "session_id": session_id, + "outcome": outcome, + "covered": covered, + "detail": detail, + "at": at, + } + ), + encoding="utf-8", + ) + + +def _payload(session_id: str = SID, *, stop_hook_active: bool = False) -> dict: + return {"session_id": session_id, "stop_hook_active": stop_hook_active} + + +def _counters(project_dir: pathlib.Path) -> dict: + path = project_dir / ".cyclaudes" / "counters.json" + if not path.exists(): + return {} + return json.loads(path.read_text(encoding="utf-8")) + + +# --------------------------------------------------------------------------- +# nothing to verify -> never nag +# --------------------------------------------------------------------------- + + +def test_no_pending_state_allows_silently(tmp_path): + """A turn with no pending-ui record must not block (non-UI turns stay fast).""" + decision = stop_gate.decide(_payload(), str(tmp_path)) + assert decision["decision"] == "allow" + assert decision["system_message"] is None + # No verification happened -> no audit counters touched. + assert _counters(tmp_path) == {} + + +def test_empty_ui_touched_allows(tmp_path): + _write_pending(tmp_path, SID, []) + assert stop_gate.decide(_payload(), str(tmp_path))["decision"] == "allow" + + +# --------------------------------------------------------------------------- +# block-when-unverified: reason names the files +# --------------------------------------------------------------------------- + + +def test_ui_touched_without_result_blocks_naming_files(tmp_path): + _write_pending(tmp_path, SID, ["src/App.tsx", "src/Panel.xaml"]) + decision = stop_gate.decide(_payload(), str(tmp_path)) + assert decision["decision"] == "block" + assert "src/App.tsx" in decision["reason"] + assert "src/Panel.xaml" in decision["reason"] + assert _counters(tmp_path).get("block") == 1 + + +# --------------------------------------------------------------------------- +# block-on-fail: reason carries the expected-vs-actual detail +# --------------------------------------------------------------------------- + + +def test_fail_blocks_and_reason_carries_the_diff(tmp_path): + diff = "expected Save button ENABLED, actual DISABLED" + _write_pending(tmp_path, SID, ["src/App.tsx"]) + _write_result(tmp_path, SID, "fail", ["src/App.tsx"], diff) + decision = stop_gate.decide(_payload(), str(tmp_path)) + assert decision["decision"] == "block" + assert diff in decision["reason"] + assert _counters(tmp_path).get("fail") == 1 + + +# --------------------------------------------------------------------------- +# allow-on-pass (covered) +# --------------------------------------------------------------------------- + + +def test_pass_covering_touched_set_allows(tmp_path): + _write_pending(tmp_path, SID, ["src/App.tsx"]) + _write_result(tmp_path, SID, "pass", ["src/App.tsx"], "") + decision = stop_gate.decide(_payload(), str(tmp_path)) + assert decision["decision"] == "allow" + assert decision["system_message"] is None + assert _counters(tmp_path).get("pass") == 1 + + +def test_pass_not_covering_touched_set_still_blocks(tmp_path): + """A pass that covers only some touched files does not satisfy the gate.""" + _write_pending(tmp_path, SID, ["src/App.tsx", "src/New.tsx"]) + _write_result(tmp_path, SID, "pass", ["src/App.tsx"], "") + decision = stop_gate.decide(_payload(), str(tmp_path)) + assert decision["decision"] == "block" + assert "src/New.tsx" in decision["reason"] + assert "src/App.tsx" not in decision["reason"] # already covered + + +# --------------------------------------------------------------------------- +# THE most important rule: abstain allows AND escalates, never blocks +# --------------------------------------------------------------------------- + + +def test_abstain_allows_and_escalates_does_not_block(tmp_path): + reason = "Save button absent from the tree; window may not have loaded" + _write_pending(tmp_path, SID, ["src/App.tsx"]) + _write_result(tmp_path, SID, "abstain", ["src/App.tsx"], reason) + + decision = stop_gate.decide(_payload(), str(tmp_path)) + + # Must NOT block -- this is the whole point. + assert decision["decision"] == "allow" + # Must escalate: the abstain reason is surfaced to the user. + assert decision["escalated"] is True + assert decision["system_message"] is not None + assert reason in decision["system_message"] + # And it is tallied as an abstention for audit (never swallowed into pass). + assert _counters(tmp_path).get("abstain") == 1 + assert _counters(tmp_path).get("pass") is None + + +def test_abstain_stays_allow_across_reentry(tmp_path): + """Re-firing on the same abstain must keep allowing, never start blocking.""" + _write_pending(tmp_path, SID, ["src/App.tsx"]) + _write_result(tmp_path, SID, "abstain", ["src/App.tsx"], "cannot see it") + for _ in range(5): + decision = stop_gate.decide( + _payload(stop_hook_active=True), str(tmp_path) + ) + assert decision["decision"] == "allow" + # Counted once, not once per re-entry (de-duped by the result's `at`). + assert _counters(tmp_path).get("abstain") == 1 + + +# --------------------------------------------------------------------------- +# coverage: a new UI file after a pass re-opens the gate +# --------------------------------------------------------------------------- + + +def test_new_ui_file_after_pass_reopens_the_gate(tmp_path): + _write_pending(tmp_path, SID, ["src/App.tsx"]) + _write_result(tmp_path, SID, "pass", ["src/App.tsx"], "") + assert stop_gate.decide(_payload(), str(tmp_path))["decision"] == "allow" + + # A later edit touches a new UI file the pass did not cover. + _write_pending(tmp_path, SID, ["src/App.tsx", "src/Extra.tsx"]) + decision = stop_gate.decide(_payload(), str(tmp_path)) + assert decision["decision"] == "block" + assert "src/Extra.tsx" in decision["reason"] + + +# --------------------------------------------------------------------------- +# stop_hook_active re-entry: does not thrash / does not re-block spuriously +# --------------------------------------------------------------------------- + + +def test_reentry_on_pass_does_not_block(tmp_path): + _write_pending(tmp_path, SID, ["src/App.tsx"]) + _write_result(tmp_path, SID, "pass", ["src/App.tsx"], "") + for _ in range(4): + decision = stop_gate.decide( + _payload(stop_hook_active=True), str(tmp_path) + ) + assert decision["decision"] == "allow" + + +# --------------------------------------------------------------------------- +# bounded retry: exhaustion escalates (allow + message), never infinite block +# --------------------------------------------------------------------------- + + +def test_retry_exhaustion_escalates_instead_of_blocking_forever(tmp_path, monkeypatch): + monkeypatch.setenv("CYCLAUDES_RETRY_CAP", "3") + _write_pending(tmp_path, SID, ["src/App.tsx"]) + _write_result(tmp_path, SID, "fail", ["src/App.tsx"], "still broken") + + # First `cap` re-verifies still fail -> block each time. + for _ in range(3): + decision = stop_gate.decide( + _payload(stop_hook_active=True), str(tmp_path) + ) + assert decision["decision"] == "block" + + # The next attempt exhausts the budget: escalate (allow) rather than + # blocking a 4th time toward the 8-block cap. + final = stop_gate.decide(_payload(stop_hook_active=True), str(tmp_path)) + assert final["decision"] == "allow" + assert final["escalated"] is True + assert "after 3 attempts" in final["system_message"] + assert _counters(tmp_path).get("escalate") == 1 + + +def test_unverified_retry_also_bounded(tmp_path, monkeypatch): + """Even if verification never runs, blocks are bounded, not infinite.""" + monkeypatch.setenv("CYCLAUDES_RETRY_CAP", "2") + _write_pending(tmp_path, SID, ["src/App.tsx"]) + for _ in range(2): + assert stop_gate.decide(_payload(), str(tmp_path))["decision"] == "block" + final = stop_gate.decide(_payload(), str(tmp_path)) + assert final["decision"] == "allow" + assert final["escalated"] is True + + +def test_block_budget_resets_after_a_pass(tmp_path, monkeypatch): + """A satisfied gate clears the retry budget so a later change starts fresh.""" + monkeypatch.setenv("CYCLAUDES_RETRY_CAP", "3") + _write_pending(tmp_path, SID, ["src/App.tsx"]) + stop_gate.decide(_payload(), str(tmp_path)) # block 1 + stop_gate.decide(_payload(), str(tmp_path)) # block 2 + + _write_result(tmp_path, SID, "pass", ["src/App.tsx"], "") + assert stop_gate.decide(_payload(), str(tmp_path))["decision"] == "allow" + + # New unverified file: the budget was reset, so we get full blocks again. + _write_pending(tmp_path, SID, ["src/App.tsx", "src/Two.tsx"]) + for _ in range(3): + assert stop_gate.decide(_payload(), str(tmp_path))["decision"] == "block" + + +# --------------------------------------------------------------------------- +# foreign / malformed state is treated as "no result", never as a pass +# --------------------------------------------------------------------------- + + +def test_result_for_a_different_session_is_ignored(tmp_path): + _write_pending(tmp_path, SID, ["src/App.tsx"]) + # A verify-result whose inner session_id does not match the keyed one. + _write_result(tmp_path, SID, "pass", ["src/App.tsx"], "") + path = tmp_path / ".cyclaudes" / "verify-result" / f"{SID}.json" + data = json.loads(path.read_text(encoding="utf-8")) + data["session_id"] = "someone-else" + path.write_text(json.dumps(data), encoding="utf-8") + + decision = stop_gate.decide(_payload(), str(tmp_path)) + assert decision["decision"] == "block" # not satisfied by a foreign record + + +# --------------------------------------------------------------------------- +# the __main__ shim emits the right JSON / exit code for each decision +# --------------------------------------------------------------------------- + + +def test_main_emits_block_json(tmp_path, monkeypatch, capsys): + _write_pending(tmp_path, SID, ["src/App.tsx"]) + monkeypatch.setattr( + "sys.stdin", + _FakeStdin(json.dumps({"session_id": SID, "cwd": str(tmp_path)})), + ) + ret = stop_gate.main() + out = json.loads(capsys.readouterr().out) + assert ret == 0 + assert out["decision"] == "block" + assert "src/App.tsx" in out["reason"] + + +def test_main_emits_system_message_on_abstain(tmp_path, monkeypatch, capsys): + _write_pending(tmp_path, SID, ["src/App.tsx"]) + _write_result(tmp_path, SID, "abstain", ["src/App.tsx"], "cannot see it") + monkeypatch.setattr( + "sys.stdin", + _FakeStdin(json.dumps({"session_id": SID, "cwd": str(tmp_path)})), + ) + ret = stop_gate.main() + out = json.loads(capsys.readouterr().out) + assert ret == 0 + assert "decision" not in out # an allow, not a block + assert "cannot see it" in out["systemMessage"] + + +def test_main_allows_silently_with_nothing_to_verify(tmp_path, monkeypatch, capsys): + monkeypatch.setattr( + "sys.stdin", + _FakeStdin(json.dumps({"session_id": SID, "cwd": str(tmp_path)})), + ) + ret = stop_gate.main() + assert ret == 0 + assert capsys.readouterr().out.strip() == "" # no nag, no output + + +class _FakeStdin: + def __init__(self, text: str) -> None: + self._text = text + + def read(self) -> str: + return self._text diff --git a/tests/test_verify_result.py b/tests/test_verify_result.py new file mode 100644 index 0000000..2c9d8f3 --- /dev/null +++ b/tests/test_verify_result.py @@ -0,0 +1,185 @@ +"""The verify-result writer must emit the frozen schema for each outcome. + +Phase 3, deliverable B (planning/PHASE_3.md, FROZEN INTERFACE). The Stop gate +routes on ``verify-result/.json``; this is the module that writes +it. Two things must hold and are proven here: + +* the record it writes matches the frozen schema exactly, for pass/fail/abstain; +* the outcome mapping is the *same* three-outcome semantics the rest of the + package enforces -- a real failure is never reclassified as an abstention, nor + the reverse (:mod:`cyclaudes.abstain`, :mod:`cyclaudes.pytest_plugin`). + +The pure functions are driven directly (no subprocess, no editable install); +one ``pytester`` run drives the full collector path end-to-end for each outcome. +""" + +from __future__ import annotations + +import json + +import pytest + +from cyclaudes import EXIT_ABSTAINED +from cyclaudes.verify_result import ( + OUTCOMES, + classify, + map_exit_code, + read_covered_from_pending, + write_result, +) + + +# --------------------------------------------------------------------------- +# outcome mapping mirrors the existing exit-code semantics exactly +# --------------------------------------------------------------------------- + + +def test_map_exit_code_matches_three_outcome_semantics(): + assert map_exit_code(0) == "pass" + assert map_exit_code(EXIT_ABSTAINED) == "abstain" # 12, not 1 + assert map_exit_code(1) == "fail" + + +def test_map_exit_code_errs_toward_fail_for_odd_codes(): + # A collection error (5) or usage error (4) has verified nothing -> fail, + # never a silent pass and never miscounted as an abstention. + for code in (2, 3, 4, 5): + assert map_exit_code(code) == "fail" + assert map_exit_code(EXIT_ABSTAINED) == "abstain" + + +def test_classify_failure_outranks_abstention(): + outcome, detail = classify( + failures=[("t::a", "expected X, got Y")], + abstentions=[("t::b", "could not see it")], + ) + assert outcome == "fail" + assert "expected X, got Y" in detail + + +def test_classify_abstention_when_no_failure(): + outcome, detail = classify(failures=[], abstentions=[("t::b", "no a11y grant")]) + assert outcome == "abstain" + assert "no a11y grant" in detail + + +def test_classify_pass_when_clean(): + assert classify(failures=[], abstentions=[]) == ("pass", "") + + +# --------------------------------------------------------------------------- +# the writer produces the frozen schema for each of pass / fail / abstain +# --------------------------------------------------------------------------- + +FROZEN_KEYS = {"session_id", "outcome", "covered", "detail", "at"} + + +@pytest.mark.parametrize( + "outcome,detail", + [ + ("pass", ""), + ("fail", "expected Save ENABLED, actual DISABLED"), + ("abstain", "Save button absent from the tree"), + ], +) +def test_write_result_matches_frozen_schema(tmp_path, outcome, detail): + path = write_result( + str(tmp_path), "sess-9", outcome, ["src/App.tsx"], detail + ) + record = json.loads(open(path, encoding="utf-8").read()) + + assert set(record) == FROZEN_KEYS + assert record["session_id"] == "sess-9" + assert record["outcome"] == outcome + assert record["covered"] == ["src/App.tsx"] + assert record["detail"] == detail + assert record["at"] # an iso8601 stamp is present + # Written where the Stop gate reads it. + assert path.endswith(f"verify-result{__import__('os').sep}sess-9.json") + + +def test_write_result_rejects_unknown_outcome(tmp_path): + with pytest.raises(ValueError): + write_result(str(tmp_path), "sess-9", "maybe", [], "") + assert set(OUTCOMES) == {"pass", "fail", "abstain"} + + +def test_read_covered_defaults_to_pending_touched_set(tmp_path): + pending = tmp_path / ".cyclaudes" / "pending-ui" / "sess-9.json" + pending.parent.mkdir(parents=True, exist_ok=True) + pending.write_text( + json.dumps({"session_id": "sess-9", "ui_touched": ["a.tsx", "b.xaml"]}), + encoding="utf-8", + ) + assert read_covered_from_pending(str(tmp_path), "sess-9") == ["a.tsx", "b.xaml"] + assert read_covered_from_pending(str(tmp_path), "absent") == [] + + +# --------------------------------------------------------------------------- +# end-to-end: a real (subprocess) pytest run maps each outcome via its exit code +# --------------------------------------------------------------------------- + +_PASS = "def test_ok(): assert True\n" +_FAIL = "def test_bad(): assert False, 'the button was enabled'\n" +_ABSTAIN = ( + "from cyclaudes import CannotVerify\n" + "def test_cannot(): raise CannotVerify('Save button absent from the tree')\n" +) + + +@pytest.mark.parametrize( + "body,expected_outcome", + [(_PASS, "pass"), (_FAIL, "fail"), (_ABSTAIN, "abstain")], +) +def test_exit_code_maps_through_a_real_run(pytester, body, expected_outcome): + """A real pytest run's exit code maps to the right outcome. + + The abstain body must exit :data:`EXIT_ABSTAINED` (12) and map to + ``abstain``, never ``fail`` -- proving the mapping rides the same + exit-code contract the abstention plugin already guarantees. Run in a + subprocess (this repo's isolation pattern) so nested in-process pytest + state cannot bleed. + """ + pytester.makepyfile(body) + result = pytester.runpytest_subprocess() + assert map_exit_code(result.ret) == expected_outcome + + +def test_collector_reads_abstain_and_fail_markers(): + """The collector must classify an abstention as such, not as a failure. + + The abstention plugin represents an abstain as ``report.outcome == + "failed"`` with ``cyclaudes_abstained = True``; the collector keys off that + marker, so a fake pair of reports is enough to prove it never miscounts an + abstention as a failure. + """ + import types + + import cyclaudes.verify_result as vr + + abstain_report = types.SimpleNamespace( + nodeid="t::abstains", + when="call", + failed=True, + longreprtext="irrelevant", + cyclaudes_abstained=True, + cyclaudes_abstain_reason="Save button absent from the tree", + user_properties=[], + ) + fail_report = types.SimpleNamespace( + nodeid="t::fails", + when="call", + failed=True, + longreprtext="expected ENABLED, got DISABLED", + user_properties=[], + ) + + collector = vr._ResultCollector() + collector.pytest_runtest_logreport(abstain_report) + collector.pytest_runtest_logreport(fail_report) + + assert collector.abstentions == [("t::abstains", "Save button absent from the tree")] + assert collector.failures == [("t::fails", "expected ENABLED, got DISABLED")] + # And classify puts the failure first (broken outranks unchecked). + outcome, _ = vr.classify(collector.failures, collector.abstentions) + assert outcome == "fail"