merge queue: checking #1080 on main (0d82f81), stacked on #1042, #1055 and #1076

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)