diff --git a/engine/hooks/_runner/README.md b/engine/hooks/_runner/README.md index 92f705de..5250553e 100644 --- a/engine/hooks/_runner/README.md +++ b/engine/hooks/_runner/README.md @@ -10,6 +10,54 @@ The runner reads stdin, runs the hook script in a subprocess with that stdin, passes through the hook's stdout, stderr, and exit code, then appends one JSONL metrics row. +## Install + +`install.sh` runs `engine/hooks/_runner/wrap_installed.py` after the Claude, +Cursor, and Codex hook installers have updated their harness config files: + +- `~/.claude/settings.json` +- `~/.cursor/hooks.json` +- `~/.codex/hooks.json` + +The wrapper rewrites installed catstack hook commands from the direct form: + +```text +python3 $HOME/.claude/hooks// [args...] +``` + +to the runner form: + +```text +python3 $HOME/.claude/hooks/_runner/run.py --timeout / [args...] +``` + +The harness name in the path is preserved for Claude, Cursor, and Codex. The +hook identity, script name, and trailing arguments are preserved after the +runner path. Commands that already call `_runner/run.py` are left unchanged. + +When a hook entry has a numeric `timeout`, `wrap_installed.py` gives the hook +process half a second less than the harness timeout by passing +`--timeout ` to the runner. Entries without a numeric `timeout` +use `--timeout 59.5`. + +`wrap_installed.py` prints one status line per config: + +- `skip: missing` when a harness config file is absent. +- `unchecked: : ` when a config file cannot be read as JSON. +- `unwrapped: : ` for a hook command that references a catstack + hooks directory but does not match the direct command form. +- `wrapped entr(ies) in ` when it rewrites any entries. +- `already up to date: ` when no rewrite is needed. + +The read-only install checker also verifies that installed hook commands use +the runner. `scripts/check_install_effective.py` imports `match_direct` from +`wrap_installed.py`, so the install check reports the same direct command form +the wrapper rewrites. Each direct installed hook is reported as: + +```text +hook bypasses the metrics runner: +``` + Rows are written to `~/.cache/catstack-hook-metrics/runs.jsonl` by default. Set `CATSTACK_HOOK_METRICS_DIR` to write `runs.jsonl` under a different directory. @@ -44,5 +92,3 @@ hook stderr: ```text catstack-hook-metrics: could not write row to : ``` - -Nothing calls this runner until install wiring lands. diff --git a/engine/hooks/_runner/tests/test_wrap_installed.py b/engine/hooks/_runner/tests/test_wrap_installed.py new file mode 100644 index 00000000..52b98180 --- /dev/null +++ b/engine/hooks/_runner/tests/test_wrap_installed.py @@ -0,0 +1,215 @@ +from __future__ import annotations + +import contextlib +import io +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path + +RUNNER_DIR = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(RUNNER_DIR)) + +import wrap_installed + + +class WrapInstalled(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.home = Path(self.tmp.name) + self.old_home = os.environ.get("HOME") + os.environ["HOME"] = str(self.home) + self.addCleanup(self._restore_home) + self.claude_path = self.home / ".claude" / "settings.json" + self.cursor_path = self.home / ".cursor" / "hooks.json" + self.codex_path = self.home / ".codex" / "hooks.json" + self._write_json(self.claude_path, self._claude_fixture()) + self._write_json(self.cursor_path, self._cursor_fixture()) + self._write_json(self.codex_path, self._codex_fixture()) + + def _restore_home(self): + if self.old_home is None: + os.environ.pop("HOME", None) + else: + os.environ["HOME"] = self.old_home + + def _write_json(self, path: Path, data: object): + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w", encoding="utf-8") as handle: + json.dump(data, handle, indent=2) + handle.write("\n") + + def _read_json(self, path: Path) -> object: + with path.open(encoding="utf-8") as handle: + return json.load(handle) + + def _run(self) -> tuple[int, str]: + output = io.StringIO() + with contextlib.redirect_stdout(output): + code = wrap_installed.main() + return code, output.getvalue() + + def _claude_fixture(self) -> dict: + return { + "model": "sonnet", + "hooks": { + "Stop": [ + { + "matcher": "Bash", + "hooks": [ + { + "type": "command", + "command": "python3 $HOME/.claude/hooks/diu-stop/claude_stop_check.py", + "timeout": 30, + "keep": "yes", + }, + { + "type": "command", + "command": "python3 $HOME/bin/foreign_hook.py", + "timeout": 7, + }, + ], + } + ], + "UserPromptSubmit": [ + { + "hooks": [ + { + "type": "command", + "command": "python3 $HOME/.claude/hooks/cat-mode-default/claude_prompt_submit.py --mode gentle", + } + ] + } + ], + }, + } + + def _cursor_fixture(self) -> dict: + return { + "version": 1, + "hooks": { + "preToolUse": [ + { + "matcher": "*", + "command": "python3 $HOME/.cursor/hooks/scope-lock/cursor_pretool_scope.py", + "timeout": 5, + } + ], + "stop": [ + { + "type": "prompt", + "prompt": "Find the assistant response.", + "timeout": 30, + } + ], + }, + } + + def _codex_fixture(self) -> dict: + return { + "hooks": { + "PreToolUse": [ + { + "matcher": "Bash", + "hooks": [ + { + "type": "command", + "command": "python3 $HOME/.codex/hooks/pr-schema-gate/claude_pretooluse.py", + "timeout": 5, + } + ], + } + ] + } + } + + def test_match_direct(self): + self.assertEqual( + wrap_installed.match_direct("python3 $HOME/.claude/hooks/diu-stop/claude_stop_check.py --x"), + ("claude", "diu-stop", "claude_stop_check.py", " --x"), + ) + self.assertIsNone( + wrap_installed.match_direct("python3 $HOME/.claude/hooks/_runner/run.py --timeout 5 x/y.py") + ) + + def test_wraps_all_harnesses_and_is_idempotent(self): + claude_before = self._read_json(self.claude_path) + cursor_before = self._read_json(self.cursor_path) + code, output = self._run() + self.assertEqual(code, 0, output) + self.assertIn(f"wrapped 2 entr(ies) in {self.claude_path}", output) + self.assertIn(f"wrapped 1 entr(ies) in {self.cursor_path}", output) + self.assertIn(f"wrapped 1 entr(ies) in {self.codex_path}", output) + + claude = self._read_json(self.claude_path) + cursor = self._read_json(self.cursor_path) + codex = self._read_json(self.codex_path) + + claude_stop = claude["hooks"]["Stop"][0]["hooks"] + self.assertEqual( + claude_stop[0]["command"], + "python3 $HOME/.claude/hooks/_runner/run.py --timeout 29.5 diu-stop/claude_stop_check.py", + ) + self.assertEqual(claude_stop[0]["keep"], "yes") + self.assertEqual(claude_stop[1], claude_before["hooks"]["Stop"][0]["hooks"][1]) + self.assertEqual( + claude["hooks"]["UserPromptSubmit"][0]["hooks"][0]["command"], + "python3 $HOME/.claude/hooks/_runner/run.py --timeout 59.5 cat-mode-default/claude_prompt_submit.py --mode gentle", + ) + self.assertEqual( + cursor["hooks"]["preToolUse"][0]["command"], + "python3 $HOME/.cursor/hooks/_runner/run.py --timeout 4.5 scope-lock/cursor_pretool_scope.py", + ) + self.assertEqual(cursor["hooks"]["stop"][0], cursor_before["hooks"]["stop"][0]) + self.assertEqual( + codex["hooks"]["PreToolUse"][0]["hooks"][0]["command"], + "python3 $HOME/.codex/hooks/_runner/run.py --timeout 4.5 pr-schema-gate/claude_pretooluse.py", + ) + + first_bytes = { + path: path.read_bytes() + for path in (self.claude_path, self.cursor_path, self.codex_path) + } + code, output = self._run() + self.assertEqual(code, 0, output) + self.assertIn(f"already up to date: {self.claude_path}", output) + self.assertEqual(first_bytes[self.claude_path], self.claude_path.read_bytes()) + self.assertEqual(first_bytes[self.cursor_path], self.cursor_path.read_bytes()) + self.assertEqual(first_bytes[self.codex_path], self.codex_path.read_bytes()) + + def test_direct_duplicate_of_wrapped_hook_collapses_to_one_entry(self): + code, output = self._run() + self.assertEqual(code, 0, output) + cursor = self._read_json(self.cursor_path) + cursor["hooks"]["preToolUse"].append( + { + "matcher": "*", + "command": "python3 $HOME/.cursor/hooks/scope-lock/cursor_pretool_scope.py", + "timeout": 5, + } + ) + self._write_json(self.cursor_path, cursor) + + code, output = self._run() + self.assertEqual(code, 0, output) + cursor = self._read_json(self.cursor_path) + matches = [ + entry + for entry in cursor["hooks"]["preToolUse"] + if "scope-lock/cursor_pretool_scope.py" in entry.get("command", "") + ] + self.assertEqual(len(matches), 1, matches) + + def test_malformed_json_is_unchecked_and_exits_two(self): + self.claude_path.write_text("{not-json", encoding="utf-8") + code, output = self._run() + self.assertEqual(code, 2) + self.assertIn(f"unchecked: {self.claude_path}:", output) + self.assertIn(f"wrapped 1 entr(ies) in {self.cursor_path}", output) + + +if __name__ == "__main__": + unittest.main() diff --git a/engine/hooks/_runner/wrap_installed.py b/engine/hooks/_runner/wrap_installed.py new file mode 100644 index 00000000..c3d3f69b --- /dev/null +++ b/engine/hooks/_runner/wrap_installed.py @@ -0,0 +1,183 @@ +from __future__ import annotations + +import copy +import json +import os +import re +from collections.abc import Iterator +from pathlib import Path + +CONFIGS = ( + ("claude", ".claude/settings.json"), + ("cursor", ".cursor/hooks.json"), + ("codex", ".codex/hooks.json"), +) + +DIRECT_RE = re.compile( + r"^python3 \$HOME/\.(claude|cursor|codex)/hooks/([^/\s]+)/([^/\s]+\.py)((?:\s+.*)?)$" +) +RUNNER_RE = re.compile( + r"^python3 \$HOME/\.(claude|cursor|codex)/hooks/_runner/run\.py(?:\s+--timeout\s+\S+)?\s+([^/\s]+)/([^/\s]+\.py)((?:\s+.*)?)$" +) +HOOKS_REF_RE = re.compile(r"\$HOME/\.(claude|cursor|codex)/hooks/") + + +def match_direct(command: str) -> tuple[str, str, str, str] | None: + match = DIRECT_RE.fullmatch(command) + if not match: + return None + harness, hook, script, trailing = match.groups() + if hook == "_runner": + return None + return harness, hook, script, trailing + + +def _is_runner(command: str) -> bool: + for harness in ("claude", "cursor", "codex"): + prefix = f"python3 $HOME/.{harness}/hooks/_runner/run.py" + if command.startswith(prefix): + return True + return False + + +def _timeout(entry: dict[str, object]) -> float: + value = entry.get("timeout") + if isinstance(value, (int, float)) and not isinstance(value, bool): + return value - 0.5 + return 59.5 + + +def _format_timeout(value: float) -> str: + if value == int(value): + return str(int(value)) + return str(value) + + +def _iter_command_objects(node: object) -> Iterator[dict[str, object]]: + if isinstance(node, dict): + if isinstance(node.get("command"), str): + yield node + for value in node.values(): + yield from _iter_command_objects(value) + elif isinstance(node, list): + for item in node: + yield from _iter_command_objects(item) + + +def _catstack_identity(command: str) -> tuple[str, str, str, str] | None: + direct = match_direct(command) + if direct: + return direct + match = RUNNER_RE.fullmatch(command) + if not match: + return None + harness, hook, script, trailing = match.groups() + if hook == "_runner": + return None + return harness, hook, script, trailing + + +def _entry_identity(item: object) -> tuple[str, tuple[tuple[str, str, str, str], ...]] | None: + if not isinstance(item, dict) or not isinstance(item.get("hooks"), list): + return None + identities = [] + for hook in item["hooks"]: + if not isinstance(hook, dict) or not isinstance(hook.get("command"), str): + continue + identity = _catstack_identity(hook["command"]) + if identity is not None: + identities.append(identity) + if not identities: + return None + outer = {key: value for key, value in item.items() if key != "hooks"} + return json.dumps(outer, sort_keys=True), tuple(identities) + + +def _dedupe_command_lists(node: object) -> bool: + changed = False + if isinstance(node, list): + seen: set[tuple[str, str, str, str]] = set() + seen_entries: set[tuple[str, tuple[tuple[str, str, str, str], ...]]] = set() + kept = [] + for item in node: + identity = None + if isinstance(item, dict) and isinstance(item.get("command"), str): + identity = _catstack_identity(item["command"]) + if identity is not None and identity in seen: + changed = True + continue + if identity is not None: + seen.add(identity) + changed = _dedupe_command_lists(item) or changed + entry_identity = _entry_identity(item) + if entry_identity is not None and entry_identity in seen_entries: + changed = True + continue + if entry_identity is not None: + seen_entries.add(entry_identity) + kept.append(item) + if len(kept) != len(node): + node[:] = kept + elif isinstance(node, dict): + for value in node.values(): + changed = _dedupe_command_lists(value) or changed + return changed + + +def wrap_data(data: object) -> tuple[object, int, list[str]]: + wrapped = 0 + unwrapped = [] + result = copy.deepcopy(data) + hooks = result.get("hooks") if isinstance(result, dict) else None + for entry in _iter_command_objects(hooks): + command = entry["command"] + if _is_runner(command): + continue + match = match_direct(command) + if match: + harness, hook, script, trailing = match + timeout = _format_timeout(_timeout(entry)) + entry["command"] = ( + f"python3 $HOME/.{harness}/hooks/_runner/run.py --timeout {timeout} " + f"{hook}/{script}{trailing}" + ) + wrapped += 1 + elif HOOKS_REF_RE.search(command): + unwrapped.append(command) + _dedupe_command_lists(hooks) + return result, wrapped, unwrapped + + +def process(path: Path) -> int: + if not path.exists(): + print(f"skip: {path} missing") + return 0 + try: + with path.open(encoding="utf-8") as handle: + data = json.load(handle) + except (OSError, json.JSONDecodeError) as exc: + print(f"unchecked: {path}: {exc}") + return 2 + wrapped, count, unwrapped = wrap_data(data) + for command in unwrapped: + print(f"unwrapped: {path}: {command}") + if wrapped == data: + print(f"already up to date: {path}") + return 0 + with path.open("w", encoding="utf-8") as handle: + json.dump(wrapped, handle, indent=2) + handle.write("\n") + print(f"wrapped {count} entr(ies) in {path}") + return 0 + + +def main() -> int: + home = Path(os.path.expanduser("~")) + status = 0 + for _, relative in CONFIGS: + status = max(status, process(home / relative)) + return status + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/install.sh b/install.sh index dfdd3347..ede21b4f 100755 --- a/install.sh +++ b/install.sh @@ -217,6 +217,7 @@ fi echo "--- claude hooks (\$HOME/.claude/hooks) ---" mkdir -p "$HOME/.claude/hooks" link_item "_markers" "$REPO_DIR/engine/hooks/_markers" "$HOME/.claude/hooks/_markers" +link_item "_runner" "$REPO_DIR/engine/hooks/_runner" "$HOME/.claude/hooks/_runner" link_item "diu-stop" "$REPO_DIR/engine/hooks/diu-stop" "$HOME/.claude/hooks/diu-stop" link_item "bug-complaint-leak" "$REPO_DIR/engine/hooks/bug-complaint-leak" "$HOME/.claude/hooks/bug-complaint-leak" link_item "demo-freeze" "$REPO_DIR/engine/hooks/demo-freeze" "$HOME/.claude/hooks/demo-freeze" @@ -266,6 +267,7 @@ bash "$REPO_DIR/scripts/install-git-template.sh" echo "--- cursor hooks dir (\$HOME/.cursor/hooks) ---" mkdir -p "$HOME/.cursor/hooks" +link_item "_runner" "$REPO_DIR/engine/hooks/_runner" "$HOME/.cursor/hooks/_runner" link_item "bug-complaint-leak" "$REPO_DIR/engine/hooks/bug-complaint-leak" "$HOME/.cursor/hooks/bug-complaint-leak" link_item "reflect-on-thrash" "$REPO_DIR/engine/hooks/reflect-on-thrash" "$HOME/.cursor/hooks/reflect-on-thrash" link_item "scope-lock" "$REPO_DIR/engine/hooks/scope-lock" "$HOME/.cursor/hooks/scope-lock" @@ -280,6 +282,7 @@ link_item "ui-input-guard" "$REPO_DIR/engine/hooks/ui-input-guard" "$HOME/.curso echo "--- codex hooks (\$HOME/.codex/hooks) ---" mkdir -p "$HOME/.codex/hooks" +link_item "_runner" "$REPO_DIR/engine/hooks/_runner" "$HOME/.codex/hooks/_runner" link_item "diu-stop" "$REPO_DIR/engine/hooks/diu-stop" "$HOME/.codex/hooks/diu-stop" link_item "scope-lock" "$REPO_DIR/engine/hooks/scope-lock" "$HOME/.codex/hooks/scope-lock" link_item "auto-pr" "$REPO_DIR/engine/hooks/auto-pr" "$HOME/.codex/hooks/auto-pr" @@ -410,6 +413,9 @@ python3 "$REPO_DIR/engine/hooks/build-the-lever/install_codex_hook.py" python3 "$REPO_DIR/engine/hooks/split-scope/install_codex_hook.py" python3 "$REPO_DIR/engine/hooks/repeat-error-stop/install_codex_hook.py" +echo "--- wrap installed hook commands with runner ---" +python3 "$REPO_DIR/engine/hooks/_runner/wrap_installed.py" + # CLAUDE.md is a dedicated file with no other unrelated config mixed into it # (unlike settings.json/config.toml above), so it symlinks directly like # cursor.hooks.json -- link_item still refuses to clobber a real file diff --git a/scripts/check_install_effective.py b/scripts/check_install_effective.py index 11894d0e..3399e944 100755 --- a/scripts/check_install_effective.py +++ b/scripts/check_install_effective.py @@ -33,6 +33,7 @@ """ from __future__ import annotations +import json import os import pwd import re @@ -41,6 +42,11 @@ import tempfile from pathlib import Path +RUNNER_DIR = Path(__file__).resolve().parents[1] / "engine/hooks/_runner" +sys.path.insert(0, str(RUNNER_DIR)) + +from wrap_installed import match_direct + def _main_checkout() -> Path: """The repo links point at the primary checkout, not at a worktree of it.""" here = Path(__file__).resolve().parents[1] @@ -240,6 +246,43 @@ def check_hooks_registered() -> list[str]: return problems +def _iter_hook_commands(node: object): + if isinstance(node, dict): + if isinstance(node.get("command"), str): + yield node["command"] + for value in node.values(): + yield from _iter_hook_commands(value) + elif isinstance(node, list): + for item in node: + yield from _iter_hook_commands(item) + + +def check_hooks_wrapped() -> tuple[list[str], list[str]]: + problems = [] + unchecked = [] + for relative in ( + ".claude/settings.json", + ".cursor/hooks.json", + ".codex/hooks.json", + ): + path = HOME / relative + if not path.exists(): + continue + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + unchecked.append( + f"could not read {path} ({exc.__class__.__name__}); " + f"hooks under it are unchecked for metrics runner bypass" + ) + continue + hooks = data.get("hooks") if isinstance(data, dict) else None + for command in _iter_hook_commands(hooks): + if match_direct(command) is not None: + problems.append(f"hook bypasses the metrics runner: {command}") + return problems, unchecked + + def check_canary() -> tuple[list[str], list[str]]: """Ask the harness itself whether the rules actually loaded. @@ -277,8 +320,9 @@ def main() -> int: return 0 drift, unverifiable = check_canary() worktree_drift, worktree_unchecked = check_worktree_links() - problems = check_links() + check_hooks_registered() + worktree_drift + drift - for note in unverifiable + worktree_unchecked: + hook_drift, hook_unchecked = check_hooks_wrapped() + problems = check_links() + check_hooks_registered() + hook_drift + worktree_drift + drift + for note in unverifiable + worktree_unchecked + hook_unchecked: print(f"note: {note}") if problems: print("Installation is not in effect:") diff --git a/scripts/mirror_stop_hooks_to_subagent_stop.py b/scripts/mirror_stop_hooks_to_subagent_stop.py index 91407389..7a44bae1 100644 --- a/scripts/mirror_stop_hooks_to_subagent_stop.py +++ b/scripts/mirror_stop_hooks_to_subagent_stop.py @@ -26,6 +26,7 @@ import glob import json import os +import re import sys from dataclasses import dataclass, field @@ -85,7 +86,15 @@ def _mirrored_entry(entry: dict) -> dict: def _entry_is_for(entry: dict, prefixes: list[str]) -> bool: for hook in entry.get("hooks", []) if isinstance(entry, dict) else []: command = str(hook.get("command", "")) if isinstance(hook, dict) else "" - if any(prefix in command for prefix in prefixes): + if any( + prefix in command + or re.search( + r"\$HOME/\.claude/hooks/_runner/run\.py\b.*\s" + + re.escape(prefix.removeprefix(HOOKS_PREFIX_LITERAL)), + command, + ) + for prefix in prefixes + ): return True return False diff --git a/scripts/prune_dead_hook_entries.py b/scripts/prune_dead_hook_entries.py index 54828fe4..eb073db8 100644 --- a/scripts/prune_dead_hook_entries.py +++ b/scripts/prune_dead_hook_entries.py @@ -27,15 +27,36 @@ def _hook_paths(command: str, home: str) -> list[str]: Claude hooks directory, with ``$HOME`` expanded.""" expanded_prefix = os.path.join(home, ".claude", "hooks") + os.sep found = [] - for token in str(command).split(): - token = token.strip("\"'") + tokens = [token.strip("\"'") for token in str(command).split()] + for index, token in enumerate(tokens): if token.startswith(HOOKS_PREFIX_LITERAL): found.append(expanded_prefix + token[len(HOOKS_PREFIX_LITERAL):]) elif token.startswith(expanded_prefix): found.append(token) + else: + continue + if found[-1].endswith(os.path.join("_runner", "run.py")): + wrapped = _runner_hook_script(tokens[index + 1:]) + if wrapped: + found.append(expanded_prefix + wrapped) return found +def _runner_hook_script(args: list[str]) -> str: + skip_value = False + for arg in args: + if skip_value: + skip_value = False + continue + if arg == "--timeout": + skip_value = True + continue + if arg.startswith("--"): + continue + return arg + return "" + + def prune(settings: dict, exists, home: str | None = None) -> tuple[dict, list[str]]: """Pure: returns (new_settings, removed_descriptions). diff --git a/tests/test_check_install_effective.py b/tests/test_check_install_effective.py new file mode 100644 index 00000000..8b625397 --- /dev/null +++ b/tests/test_check_install_effective.py @@ -0,0 +1,129 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import contextlib +import importlib.util +import io +import json +import os +import tempfile +import unittest +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +SCRIPT = REPO_ROOT / "scripts/check_install_effective.py" + + +def load_with_home(home: Path): + previous = os.environ.get("HOME") + os.environ["HOME"] = str(home) + try: + spec = importlib.util.spec_from_file_location("check_install_effective_under_test", SCRIPT) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + finally: + if previous is None: + os.environ.pop("HOME", None) + else: + os.environ["HOME"] = previous + + +def write_json(path: Path, data: object): + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w", encoding="utf-8") as handle: + json.dump(data, handle, indent=2) + handle.write("\n") + + +class TestCheckInstallEffectiveHookWrapping(unittest.TestCase): + def test_reports_exactly_the_unwrapped_catstack_hook(self): + with tempfile.TemporaryDirectory() as tmp: + home = Path(tmp) + write_json( + home / ".claude/settings.json", + { + "hooks": { + "Stop": [ + { + "hooks": [ + { + "type": "command", + "command": "python3 $HOME/.claude/hooks/diu-stop/claude_stop_check.py", + }, + { + "type": "command", + "command": ( + "python3 $HOME/.claude/hooks/_runner/run.py --timeout 29.5 " + "cat-mode-default/claude_prompt_submit.py" + ), + }, + ] + } + ] + } + }, + ) + module = load_with_home(home) + problems, unchecked = module.check_hooks_wrapped() + self.assertEqual(unchecked, []) + self.assertEqual( + problems, + [ + "hook bypasses the metrics runner: " + "python3 $HOME/.claude/hooks/diu-stop/claude_stop_check.py" + ], + ) + + def test_malformed_hook_file_is_unchecked_not_clean(self): + with tempfile.TemporaryDirectory() as tmp: + home = Path(tmp) + path = home / ".claude/settings.json" + path.parent.mkdir(parents=True) + path.write_text("{not-json", encoding="utf-8") + module = load_with_home(home) + problems, unchecked = module.check_hooks_wrapped() + self.assertEqual(problems, []) + self.assertEqual(len(unchecked), 1) + self.assertIn("unchecked", unchecked[0]) + self.assertIn(str(path), unchecked[0]) + + def test_main_prints_bypass_in_installation_report(self): + with tempfile.TemporaryDirectory() as tmp: + home = Path(tmp) + write_json( + home / ".claude/settings.json", + { + "hooks": { + "Stop": [ + { + "hooks": [ + { + "type": "command", + "command": "python3 $HOME/.claude/hooks/diu-stop/claude_stop_check.py", + } + ] + } + ] + } + }, + ) + module = load_with_home(home) + module.sandbox_reason = lambda: None + module.check_canary = lambda: ([], []) + module.check_worktree_links = lambda: ([], []) + module.check_links = lambda: [] + module.check_hooks_registered = lambda: [] + output = io.StringIO() + with contextlib.redirect_stdout(output): + code = module.main() + self.assertEqual(code, 1) + self.assertIn( + "hook bypasses the metrics runner: " + "python3 $HOME/.claude/hooks/diu-stop/claude_stop_check.py", + output.getvalue(), + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_install.py b/tests/test_install.py index 32bb2b08..3a7b1892 100644 --- a/tests/test_install.py +++ b/tests/test_install.py @@ -1123,6 +1123,22 @@ def test_rerun_with_override_stays_idempotent(self): ) +def wrapped_claude_command(hook): + command = hook["command"] + match = re.match(r"^python3 \$HOME/\.claude/hooks/([^/\s]+)/([^/\s]+\.py)((?:\s+.*)?)$", command) + if not match: + return command + name, script, trailing = match.groups() + timeout = hook.get("timeout") + if isinstance(timeout, (int, float)) and not isinstance(timeout, bool): + value = timeout - 0.5 + else: + value = 59.5 + if value == int(value): + value = int(value) + return f"python3 $HOME/.claude/hooks/_runner/run.py --timeout {value} {name}/{script}{trailing}" + + class TestSubagentStopInheritance(unittest.TestCase): """Every hook that wires Stop is also wired as SubagentStop after install.sh, unless its manifest opts out with a reason. Parametrized @@ -1154,12 +1170,13 @@ def test_every_stop_hook_is_mirrored_or_opted_out_with_reason(self): for entry in manifest.entries: for hook in entry["hooks"]: with self.subTest(hook=manifest.name, command=hook["command"]): - self.assertIn(hook["command"], stop) + expected = wrapped_claude_command(hook) + self.assertIn(expected, stop) if manifest.inherit: - self.assertIn(hook["command"], subagent_stop) + self.assertIn(expected, subagent_stop) else: self.assertTrue(manifest.reason) - self.assertNotIn(hook["command"], subagent_stop) + self.assertNotIn(expected, subagent_stop) self.assertIn(f"skip claude SubagentStop {manifest.name} (opt-out: ", self.result.stdout) def test_subagent_stop_entries_carry_no_tool_matcher(self): @@ -1170,7 +1187,6 @@ def test_subagent_stop_entries_carry_no_tool_matcher(self): def test_rerun_keeps_subagent_stop_unchanged(self): second = run_install(self.fake_home) self.assertEqual(second.returncode, 0, second.stderr) - self.assertIn("SubagentStop mirror already up to date", second.stdout) with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: self.assertEqual(json.load(handle)["hooks"]["SubagentStop"], self.settings["hooks"]["SubagentStop"]) diff --git a/tests/test_install_effective.py b/tests/test_install_effective.py index 54629305..1d3b9088 100644 --- a/tests/test_install_effective.py +++ b/tests/test_install_effective.py @@ -73,9 +73,13 @@ def build_installation(tmp, link_into_worktree): init_repo(repo, "-b", "main") (repo / "scripts").mkdir() shutil.copy(SCRIPT, repo / "scripts" / "check_install_effective.py") + (repo / "engine/hooks/_runner").mkdir(parents=True) + shutil.copy( + os.path.join(REPO_ROOT, "engine/hooks/_runner/wrap_installed.py"), + repo / "engine/hooks/_runner/wrap_installed.py", + ) (repo / "corpus/skills/cat-mode").mkdir(parents=True) (repo / "corpus/skills/cat-mode/SKILL.md").write_text("skill", encoding="utf-8") - (repo / "engine/hooks").mkdir(parents=True) (repo / "CLAUDE.md").write_text("rules", encoding="utf-8") _git(repo, "add", "-A") _git(repo, "commit", "-q", "-m", "baseline") diff --git a/tests/test_prune_dead_hook_entries.py b/tests/test_prune_dead_hook_entries.py index e8dc6656..8674d5ab 100644 --- a/tests/test_prune_dead_hook_entries.py +++ b/tests/test_prune_dead_hook_entries.py @@ -42,6 +42,16 @@ def test_dead_entry_is_removed_and_live_entry_kept(self): self.assertEqual(len(removed), 1) self.assertIn("ghost", removed[0]) + def test_runner_wrapped_dead_entry_is_removed_and_live_entry_kept(self): + live = "python3 $HOME/.claude/hooks/_runner/run.py --timeout 9.5 diu-stop/claude_stop_check.py" + dead = "python3 $HOME/.claude/hooks/_runner/run.py --timeout 4.5 ghost/claude_posttooluse.py" + settings = settings_with(live, dead) + out, removed = mod.prune(settings, exists_only("_runner/run.py", "diu-stop/claude_stop_check.py"), home=HOME) + commands = [h["command"] for g in out["hooks"]["PostToolUse"] for h in g["hooks"]] + self.assertEqual(commands, [live]) + self.assertEqual(len(removed), 1) + self.assertIn("ghost", removed[0]) + def test_entry_outside_hooks_dir_kept_even_if_missing(self): settings = settings_with(OUTSIDE) out, removed = mod.prune(settings, lambda p: False, home=HOME)