merge queue: checking #1080 on main (0d82f81), stacked on 4 pull requests ahead

corpus/skills/cat-mode/SKILL.md

@@ -50,10 +50,10 @@ Once direction is set, act — don't ask permission for each sub-step. One fully
 - **An open PR is not evidence it is still needed.** Run
   `scripts/ci/check_branch_not_superseded.py` and report it before a land.
 - **A hand-back ("open the app and do it") is an unverified claim.**
-  "Cannot" needs the same evidence as any claim; keep manual steps for what
-  only a human can do (OAuth consent, a store upload). Before handing back,
-  name every surface tried and grep the artifact already located (`--help`,
-  bundle/asar, DB, logs).
+  "Cannot" needs evidence; keep manual steps for what only a human can do
+  (OAuth consent). First try the step once in the main session: a block a
+  helper reports is the helper's, not yours (no known prior art). Then
+  name every surface tried and grep the artifact already located (logs, DB).
 - **A blocked hand-back relays the gate's exit word for word.** When a hook, guard, or check blocks and the user must act, paste its exit message and the exact command, path, or marker it names, unshortened; a summary can drop the one step that gets them out.
 - **Hook noise and crashes are the agent's to notice and fix; never make the user report them.** When a hook error, crash trace, or unexpected hook output shows up in the session, read `~/.cache/catstack-hook-metrics/runs.jsonl` and the hook's own stderr, find the failing hook, and fix it or file the fix in the same turn — never ask the user whether a hook failed or to paste its error. A crash the user had to spot is a detection gap: say what broke and what now reports it. No known prior art.
 - Destructive or hard-to-reverse actions (force-push, bypassing a merge

engine/hooks/_runner/dispatch.py

@@ -0,0 +1,257 @@
+from __future__ import annotations
+
+import argparse
+import concurrent.futures
+import glob
+import json
+import os
+import re
+import subprocess
+import sys
+import time
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+import outcome
+import run
+
+DEFAULT_BUDGET = 59.5
+OPT_OUT_KEY = "subagent_stop"
+MIRRORED_EVENTS = {"SubagentStop": "Stop"}
+DIRECT_RE = re.compile(r"^python3 \$HOME/\.(claude|cursor|codex)/hooks/([^/\s]+)/([^/\s]+\.py)((?:\s+.*)?)$")
+MESSAGE_KEYS = ("reason", "message", "additionalContext", "additional_context", "user_message")
+
+
+def load_event_hooks(hooks_root: str, harness: str, event: str) -> tuple[list[dict], list[str]]:
+    """Every hook registered for `event`, read live from each hook's own
+    `<harness>.hook.json` manifest under `hooks_root`.
+
+    For a mirrored event (SubagentStop mirrors Stop, the same way
+    scripts/install/mirror_stop_hooks_to_subagent_stop.py mirrors it into
+    settings.json at install time) a hook opts out with the same
+    `subagent_stop: {inherit: false, reason: ...}` manifest key, and the
+    matcher is dropped -- SubagentStop has no per-tool matcher.
+
+    Second return value: one warning line per manifest that exists but could
+    not be read -- a hook silently missing from an event because its
+    manifest was corrupt is a check that could not run, not a clean miss."""
+    source_event = MIRRORED_EVENTS.get(event, event)
+    mirrored = source_event != event
+    records: list[dict] = []
+    warnings: list[str] = []
+    for hook_dir in sorted(glob.glob(os.path.join(hooks_root, "*"))):
+        name = os.path.basename(hook_dir)
+        if name.startswith("_") or not os.path.isdir(hook_dir):
+            continue
+        fragment_path = os.path.join(hook_dir, f"{harness}.hook.json")
+        if not os.path.isfile(fragment_path):
+            continue
+        try:
+            with open(fragment_path, encoding="utf-8") as handle:
+                manifest = json.load(handle)
+        except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
+            warnings.append(f"catstack-hook-dispatcher: could not read {fragment_path}: {exc}\n")
+            continue
+        entries = manifest.get("hooks", {}).get(source_event) if isinstance(manifest, dict) else None
+        if not entries:
+            continue
+        if mirrored:
+            opt_out = manifest.get(OPT_OUT_KEY)
+            if isinstance(opt_out, dict) and opt_out.get("inherit") is False:
+                continue
+        for entry in entries:
+            if not isinstance(entry, dict) or not isinstance(entry.get("hooks"), list):
+                continue
+            matcher = None if mirrored else entry.get("matcher")
+            for hook in entry["hooks"]:
+                if not isinstance(hook, dict) or not isinstance(hook.get("command"), str):
+                    continue
+                identity = _parse_command(hook["command"], harness)
+                if identity is None or identity[0] != name:
+                    continue
+                _, script, trailing = identity
+                records.append(
+                    {
+                        "hook": name,
+                        "script": script,
+                        "args": trailing.split() if trailing else [],
+                        "timeout": hook.get("timeout"),
+                        "matcher": matcher,
+                    }
+                )
+    return records, warnings
+
+
+def _parse_command(command: str, harness: str) -> tuple[str, str, str] | None:
+    match = DIRECT_RE.fullmatch(command)
+    if not match or match.group(1) != harness:
+        return None
+    _, hook, script, trailing = match.groups()
+    return hook, script, trailing.strip()
+
+
+def _matcher_applies(matcher: object, payload: dict[str, object]) -> bool:
+    if matcher in (None, "", "*"):
+        return True
+    tool_name = payload.get("tool_name")
+    if not isinstance(tool_name, str):
+        return False
+    try:
+        return re.fullmatch(str(matcher), tool_name) is not None
+    except re.error:
+        return False
+
+
+def _payload(stdin: bytes) -> dict[str, object]:
+    try:
+        parsed = json.loads(stdin.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        return {}
+    return parsed if isinstance(parsed, dict) else {}
+
+
+def _run_one(hooks_root: str, python: str, record: dict, stdin: bytes, budget: float) -> dict:
+    hook, script = record["hook"], record["script"]
+    started = time.monotonic()
+    script_path = os.path.join(hooks_root, hook, script)
+    findings_path = run._make_findings_file()
+    stdout = b""
+    stderr = b""
+    exit_code = 1
+    timed_out = False
+    try:
+        if not os.path.isfile(script_path):
+            stderr = f"catstack-hook-runner: no such hook script: {script_path}\n".encode()
+        else:
+            env = os.environ.copy()
+            env["CATSTACK_HOOK_FINDINGS_FILE"] = findings_path
+            proc = subprocess.Popen(
+                [python, script_path, *record["args"]],
+                stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                cwd=os.getcwd(),
+                env=env,
+            )
+            try:
+                stdout, stderr = proc.communicate(stdin, timeout=budget)
+                exit_code = proc.returncode
+            except subprocess.TimeoutExpired:
+                proc.kill()
+                proc.communicate()
+                timed_out = True
+                stdout = b""
+                stderr = f"catstack-hook-runner: {hook}/{script} timed out after {budget}s\n".encode()
+                exit_code = 1
+        rule_ids, findings_error = run._read_rule_ids(findings_path)
+    finally:
+        run._delete_findings_file(findings_path)
+    result_outcome = outcome.classify(exit_code, stdout, stderr, timed_out)
+    row = run._row(hooks_root, hook, script, stdin, result_outcome, exit_code, started, stdout, stderr, rule_ids)
+    return {
+        "hook": hook,
+        "script": script,
+        "outcome": result_outcome,
+        "stdout": stdout,
+        "stderr": stderr + findings_error,
+        "row": row,
+    }
+
+
+def _merge_stdout(results: list[dict]) -> tuple[bytes, int]:
+    blocked = [r for r in results if r["outcome"] == "blocked"]
+    spoke = [r for r in results if r["stdout"].strip()]
+    if blocked:
+        reasons = [note for r in blocked for note in (_note(r),) if note]
+        payload: dict[str, object] = {"decision": "block", "continue": False}
+        if reasons:
+            payload["reason"] = "\n".join(reasons)
+        return json.dumps(payload).encode(), 2
+    if spoke:
+        contexts = [note for r in spoke for note in (_note(r),) if note]
+        payload = {"continue": True}
+        if contexts:
+            payload["additionalContext"] = "\n".join(contexts)
+        return json.dumps(payload).encode(), 0
+    return b"", 0
+
+
+def _note(result: dict) -> str:
+    parsed = outcome._stdout_json_object(result["stdout"])
+    text = None
+    if parsed is not None:
+        for key in MESSAGE_KEYS:
+            value = parsed.get(key)
+            if isinstance(value, str) and value:
+                text = value
+                break
+        if text is None:
+            hook_output = parsed.get("hookSpecificOutput")
+            if isinstance(hook_output, dict):
+                for key in ("permissionDecisionReason", "additionalContext"):
+                    value = hook_output.get(key)
+                    if isinstance(value, str) and value:
+                        text = value
+                        break
+    if text is None:
+        text = result["stdout"].decode("utf-8", errors="replace").strip()
+    if not text:
+        return ""
+    return f"[{result['hook']}] {text}"
+
+
+def run_dispatch(
+    hooks_root: str, harness: str, event: str, stdin: bytes, budget: float
+) -> tuple[int, bytes, bytes, list[dict]]:
+    payload = _payload(stdin)
+    all_records, warnings = load_event_hooks(hooks_root, harness, event)
+    records = [record for record in all_records if _matcher_applies(record["matcher"], payload)]
+    warning_bytes = "".join(warnings).encode()
+    if not records:
+        return 0, b"", warning_bytes, []
+
+    python = run._pick_python(sys.version_info, sys.executable, run._python_dirs(dict(os.environ)), dict(os.environ))
+    if python is None:
+        stderr = warning_bytes + (
+            f"catstack-hook-dispatcher: no Python {run.MIN_PYTHON[0]}.{run.MIN_PYTHON[1]}+ interpreter found "
+            f"for event {event}\n"
+        ).encode()
+        return 1, b"", stderr, []
+
+    results: list[dict] = []
+    with concurrent.futures.ThreadPoolExecutor(max_workers=len(records)) as pool:
+        futures = [pool.submit(_run_one, hooks_root, python, record, stdin, budget) for record in records]
+        for future in futures:
+            results.append(future.result())
+
+    stdout, exit_code = _merge_stdout(results)
+    stderr_lines = [r["stderr"] for r in results if r["stderr"]]
+    stderr = warning_bytes + b"".join(stderr_lines)
+    return exit_code, stdout, stderr, [r["row"] for r in results]
+
+
+def _parse_args(argv: list[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--event", required=True)
+    parser.add_argument("--timeout", type=float, default=DEFAULT_BUDGET)
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = _parse_args(sys.argv[1:] if argv is None else argv)
+    stdin = sys.stdin.buffer.read()
+    hooks_root = run._hooks_root()
+    harness = run._harness(hooks_root)
+    exit_code, stdout, stderr, rows = run_dispatch(hooks_root, harness, args.event, stdin, args.timeout)
+    metrics_path = run._metrics_path()
+    metrics_errors = b"".join(run._write_metrics(row, metrics_path) for row in rows)
+    sys.stdout.buffer.write(stdout)
+    sys.stdout.buffer.flush()
+    sys.stderr.buffer.write(stderr)
+    sys.stderr.buffer.write(metrics_errors)
+    sys.stderr.buffer.flush()
+    return exit_code
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

engine/hooks/_runner/doctor.py

@@ -54,6 +54,10 @@
 import sys
 import tempfile
 
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from run import MIN_PYTHON, PYTHON_OVERRIDE_ENV, _pick_python, _python_dirs
+
 HARNESS_DIRS = (".claude", ".cursor", ".codex")
 SKIP_PREFIXES = ("install_", "test_")
 SKIP_NAMES = ("detect.py", "state.py")
@@ -151,17 +155,19 @@ def _last_stderr_line(result) -> str:
     return lines[-1] if lines else f"exit={result.returncode}, no stderr"
 
 
-def classify(path: str, timeout: float, run=subprocess.run) -> tuple[str, str]:
+def classify(path: str, timeout: float, python: str, run=subprocess.run) -> tuple[str, str]:
     """(outcome, detail) for one script: ok, unreadable, import-fail, or slow.
 
     The child opens the file before importing anything, so `unreadable` beats
     `import-fail` whenever both would apply: the more specific cause is the one
     reported. A timeout is not a failure -- imports resolve before a module
-    reaches whatever it is still doing.
+    reaches whatever it is still doing. Probing under `python` rather than
+    `sys.executable` means this reports the interpreter the runner would
+    actually use for the hook, not whichever one started the doctor.
     """
     try:
         result = run(
-            [sys.executable, "-c", LOADER, path],
+            [python, "-c", LOADER, path],
             capture_output=True,
             text=True,
             timeout=timeout,
@@ -214,12 +220,24 @@ def check_hooks(home: str, timeout: float = DEFAULT_TIMEOUT, run=subprocess.run)
                 "unreadable, so nothing was verified"
             ],
         )
+    python = _pick_python(sys.version_info, sys.executable, _python_dirs(dict(os.environ)), dict(os.environ))
+    if python is None:
+        version = ".".join(str(part) for part in sys.version_info[:2])
+        return Result(
+            "hooks",
+            "fail",
+            [
+                f"the hooks need Python {MIN_PYTHON[0]}.{MIN_PYTHON[1]} or newer; this check started "
+                f"on {sys.executable} ({version}) and found no python3.N (N >= {MIN_PYTHON[1]}) on "
+                f"PATH or in the well-known python dirs; set {PYTHON_OVERRIDE_ENV} to a newer interpreter"
+            ],
+        )
     unreadable: list[tuple[str, str, str]] = []
     failures: list[str] = []
     slow: list[str] = []
     workers = min(HOOK_CHECK_WORKERS, len(scripts))
     with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:
-        results = list(executor.map(lambda path: (path, classify(path, timeout, run=run)), scripts))
+        results = list(executor.map(lambda path: (path, classify(path, timeout, python, run=run)), scripts))
     for path, (outcome, detail) in results:
         shown = os.path.relpath(path, home)
         if outcome == "unreadable":
@@ -229,8 +247,9 @@ def check_hooks(home: str, timeout: float = DEFAULT_TIMEOUT, run=subprocess.run)
         elif outcome == "slow":
             slow.append(f"{shown}: {detail} (imports resolved; still working)")
     lines = [
+        f"probing imports with {python}",
         f"checked={len(scripts)} unreadable={len(unreadable)} "
-        f"import-fail={len(failures)} slow={len(slow)}"
+        f"import-fail={len(failures)} slow={len(slow)}",
     ]
     lines.extend(f"{shown} -> {target}: {detail}" for shown, target, detail in unreadable)
     lines.extend(failures + slow)

engine/hooks/llm-judge/phrases/named-verb-guard-target-proof-request.json

@@ -0,0 +1,21 @@
+{
+  "checker": "named-verb-guard-target-proof-request",
+  "reads": "user",
+  "meaning": "The user's message requires the assistant to identify the exact live target before changing it, to reproduce the bad case on that target, or to prove the outcome on the real surface a person sees, rather than accepting a guess, a saved list, or a file read-back.",
+  "match": [
+    "find the exact device that is connected right now, then change it; don't guess",
+    "make sure you're touching the right instance before you edit anything",
+    "reproduce the bug on the real account first, then fix it",
+    "prove it shows up on the live page, not just in the file",
+    "show me it works on the actual screen",
+    "check which server is really serving traffic before you restart it"
+  ],
+  "not_match": [
+    "update the config file",
+    "which device was it last time?",
+    "the staging page was slow yesterday",
+    "\"identify the target first\" is a good rule",
+    "I don't need proof on the live site for this one"
+  ],
+  "on_hit": "named-verb-guard (target proof): the user asked for the exact live target or for proof on the real surface, and this turn changed something, but the reply's proof is a stand-in: a saved list, a read-back of the written file, or output from before the change. Query the live target before changing it, check the real surface after, paste that output, or tag the claim with a CAT-UNVERIFIED that names the blocker. Per CLAUDE.md evidence rules: proof must match the layer it claims."
+}

engine/hooks/named-verb-guard/README.md

@@ -13,6 +13,15 @@ really asked. A hit arrives on a later turn through the shared
 | delete, revert | [`named-verb-guard-delete-request.json`](../llm-judge/phrases/named-verb-guard-delete-request.json) | an `rm` / `git rm` / `git revert` / `git reset` / `git restore` command this turn, or that command in backticks |
 | stop | [`named-verb-guard-stop-request.json`](../llm-judge/phrases/named-verb-guard-stop-request.json) | no Bash / Edit / Write tool calls after the message |
 | proof, a second time this session | [`named-verb-guard-proof-demand.json`](../llm-judge/phrases/named-verb-guard-proof-demand.json) | a fenced block, `path:line`, or URL |
+| the exact live target, or proof on the real surface | [`named-verb-guard-target-proof-request.json`](../llm-judge/phrases/named-verb-guard-target-proof-request.json) | the turn changed nothing; or a live check (not a file read) ran before the first change, and a pasted output line is found in a check after the last change that is not a read-back of a file the turn wrote |
+
+Target proof reads tool results, so it can tell a stand-in from the real thing:
+a saved list read before the change, output from before the change, or a
+read-back of the written file. It sorts Bash commands by shape: a write
+(redirect, `mv`, `cp`, `tee`, `sed -i`, `git commit`, ...) is a change, and a
+command that starts with `cat`, `head`, `grep`, `jq`, `git show`, ... is a file
+read. Any other command counts as a live check, so a domain command that
+changes things without one of those shapes is missed as a change.
 
 A well-formed `{{CAT-UNVERIFIED}}` tag anywhere in the reply, or a reply that
 ends in a question, sends nothing: the guard wants proof or an honest "not
@@ -41,8 +50,11 @@ Mechanical half of the `Named constraints` and `Evidence rules` in
 
 - `detect.py` -- evidence shapes, transcript reading, and which request types to judge.
 - `claude_stop_check.py` -- Claude Stop entrypoint.
+- `repro_target_proof.py` -- runs every `tests/fixtures/target_proof/` turn through the real transcript reader; `fire_` fixtures must send target proof, `clean_` ones must not.
 - `eval_dictionary.py` -- asks the real judge two cases per dictionary; not run in CI, which has no model access.
 - `claude.hook.json` / `install_claude_hook.py` -- settings.json merge (idempotent).
 - `tests/test_hooks.py` -- which request types are sent, and inbox delivery with a fake judge.
 
 Tests: `python3 -m unittest discover -s engine/hooks/named-verb-guard/tests -v`
+
+Repro: `python3 engine/hooks/named-verb-guard/repro_target_proof.py`