diff --git a/engine/hooks/cat-mode-default/README.md b/engine/hooks/cat-mode-default/README.md index dc571635..d293dfc9 100644 --- a/engine/hooks/cat-mode-default/README.md +++ b/engine/hooks/cat-mode-default/README.md @@ -60,13 +60,22 @@ It stays silent when the flag is off or when the prompt already mentions cat-mode anywhere (a parent that told the subagent to read it gets no second copy). Same flag resolution as the prompt hook. +The shared registry keeps this hook in `warn` mode. Set +`CATSTACK_HOOK_MODE_CAT_MODE_DEFAULT=off|warn|stop` for a machine-local +override; `stop` turns the `PreToolUse` (Agent) companion into a real block +(exit 2) instead of rewriting the subagent's prompt. Detection or metrics +failures allow the harness action. + ## Files -- `detect.py`: flag resolution, typed `/cat-mode` detection, context text. -- `claude_prompt_submit.py`: the Claude entrypoint; fail-open, never denies. -- `claude.prompt.hook.json`: settings fragment `install_claude_hook.py` merges. -- `claude_pretooluse_agent.py` + `claude.agent.hook.json`: the `PreToolUse` - (`Agent`) companion that carries the default into subagent prompts. +- `detect.py`: flag resolution, typed `/cat-mode` detection, context text, + and `detect(event)`, the SDK entrypoint returning `Finding` objects. +- `claude_prompt_submit.py` / `claude_pretooluse_agent.py`: thin calls into + `engine/hooks/_sdk/runtime.py`. The agent finding carries + `output={"updatedInput": ...}`, which the shared renderer emits as + `updatedInput` rather than its generic `additionalContext`. +- `claude.prompt.hook.json` / `claude.agent.hook.json`: settings fragments + `install_claude_hook.py` merges. - `tests/fixtures/*.json`: one scenario each (fires / silent) with the environment, optional `.env` content, and payload. - `tests/fixtures/agent_*.json`: the same for the Agent-tool companion. diff --git a/engine/hooks/cat-mode-default/tests/test_agent_hook.py b/engine/hooks/cat-mode-default/tests/test_agent_hook.py index c87cfafd..c4427043 100644 --- a/engine/hooks/cat-mode-default/tests/test_agent_hook.py +++ b/engine/hooks/cat-mode-default/tests/test_agent_hook.py @@ -13,6 +13,7 @@ import json import os import sys +import tempfile import unittest from contextlib import redirect_stderr, redirect_stdout from unittest.mock import patch @@ -150,6 +151,19 @@ def test_unrelated_prompt_does_not_count(self) -> None: class FailOpenCase(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.metrics_env = patch.dict( + os.environ, + {"CATSTACK_HOOK_METRICS_DIR": self.tmp.name}, + clear=False, + ) + self.metrics_env.start() + + def tearDown(self) -> None: + self.metrics_env.stop() + self.tmp.cleanup() + def test_malformed_stdin_prints_nothing(self) -> None: out = io.StringIO() with patch.object(sys, "stdin", io.StringIO("not json")): diff --git a/engine/hooks/cat-mode-default/tests/test_hooks.py b/engine/hooks/cat-mode-default/tests/test_hooks.py index 9a080562..61b0a782 100644 --- a/engine/hooks/cat-mode-default/tests/test_hooks.py +++ b/engine/hooks/cat-mode-default/tests/test_hooks.py @@ -260,6 +260,19 @@ def test_flags_missing_install_instead_of_dead_path(self) -> None: class FailOpenCase(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.metrics_env = patch.dict( + os.environ, + {"CATSTACK_HOOK_METRICS_DIR": self.tmp.name}, + clear=False, + ) + self.metrics_env.start() + + def tearDown(self) -> None: + self.metrics_env.stop() + self.tmp.cleanup() + def test_malformed_stdin_prints_nothing(self) -> None: out = io.StringIO() with patch.object(sys, "stdin", io.StringIO("not json")): diff --git a/engine/hooks/categorical-scope-guard/claude_pretooluse.py b/engine/hooks/categorical-scope-guard/claude_pretooluse.py index 673c734c..5b4228cb 100644 --- a/engine/hooks/categorical-scope-guard/claude_pretooluse.py +++ b/engine/hooks/categorical-scope-guard/claude_pretooluse.py @@ -1,55 +1,20 @@ #!/usr/bin/env python3 -"""Claude PreToolUse on Bash: block a status-narrowed mutation of a target the -live human turn quantified with all / every / each. Exits 2 with the reason on -stderr for HIT and for UNCHECKED; exits 0 for CLEAN. -""" +"""Claude PreToolUse entrypoint for categorical-scope-guard: a thin call into +the shared hook runtime, which applies the registry mode and writes events.""" from __future__ import annotations -import json import os import sys -sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_sdk")) -from detect import CLEAN, decide_payload - -SHELL_LIKE_TOOL_NAMES = ( - "Bash", "bash", "shell", "Shell", "exec", "exec_command", - "run_terminal_cmd", "local_shell", "run_command", "shell_call", -) - - -def _tool_name(payload: dict) -> str: - return str( - payload.get("tool_name") - or payload.get("toolName") - or payload.get("tool") - or payload.get("name") - or "" - ) +from detect import detect # noqa: E402 +from runtime import run_hook # noqa: E402 def main() -> None: - raw = sys.stdin.read() - try: - payload = json.loads(raw) - except (json.JSONDecodeError, OSError) as exc: - sys.stderr.write(f"categorical-scope-guard: hook payload is not JSON, nothing to classify: {exc}\n") - return - if not isinstance(payload, dict) or _tool_name(payload) not in SHELL_LIKE_TOOL_NAMES: - return - try: - verdict = decide_payload(payload) - except Exception as exc: - sys.stderr.write( - f"categorical-scope-guard: UNCHECKED -- the detector failed ({exc!r}) while classifying a " - "status-filtered mutation. Blocked rather than passed; drop the status filter or rephrase the command.\n" - ) - sys.exit(2) - if verdict.outcome == CLEAN: - return - sys.stderr.write(verdict.message + "\n") - sys.exit(2) + run_hook("categorical-scope-guard", "claude", detect, "PreToolUse") if __name__ == "__main__": diff --git a/engine/hooks/categorical-scope-guard/detect.py b/engine/hooks/categorical-scope-guard/detect.py index bc0f1912..8f5cf4a2 100644 --- a/engine/hooks/categorical-scope-guard/detect.py +++ b/engine/hooks/categorical-scope-guard/detect.py @@ -14,15 +14,29 @@ from __future__ import annotations import sys +import hashlib import json import os import re from dataclasses import dataclass, field +sys.path.insert(0, os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_sdk")) + +from finding import Finding # noqa: E402 + HIT = "hit" CLEAN = "clean" UNCHECKED = "unchecked" +SHELL_LIKE_TOOL_NAMES = ( + "Bash", "bash", "shell", "Shell", "exec", "exec_command", + "run_terminal_cmd", "local_shell", "run_command", "shell_call", +) + +RULE_NARROWED_MUTATION = "categorical-scope-guard.narrowed-mutation" +RULE_UNREADABLE = "categorical-scope-guard.unreadable" + LIVE_TURNS = 4 MAX_SCAN_BYTES = 64 * 1024 * 1024 CHUNK_BYTES = 1024 * 1024 @@ -872,3 +886,51 @@ def decide_payload(payload: dict) -> Verdict: return Verdict(CLEAN) path = payload.get("transcript_path") or payload.get("transcriptPath") or "" return decide(command, lambda: read_live_window(path if isinstance(path, str) else "")) + + +def _event_tool_name(event: dict) -> str: + return str( + event.get("tool_name") + or event.get("toolName") + or event.get("tool") + or event.get("name") + or "" + ) + + +def _event_command(event: dict) -> str: + tool_input = event.get("tool_input") or event.get("toolInput") or {} + command = tool_input.get("command") if isinstance(tool_input, dict) else None + return command if isinstance(command, str) else "" + + +def _finding_subject(event: dict, command: str) -> str: + for key in ("tool_call_id", "toolCallId", "tool_use_id", "toolUseId", "id"): + value = event.get(key) + if isinstance(value, str) and value: + return f"tool-call:{value}" + return f"command:{hashlib.sha256(command.encode('utf-8')).hexdigest()[:16]}" + + +def detect(event: dict) -> list[Finding]: + """Return findings for the shared hook runtime. + + Only a shell-like tool call is classified, matching the guard the old + entrypoint applied before ever calling `decide_payload`. HIT and + UNCHECKED both become a finding -- this hook fails closed, so an + unreadable case still blocks in `stop` mode.""" + if not isinstance(event, dict) or _event_tool_name(event) not in SHELL_LIKE_TOOL_NAMES: + return [] + verdict = decide_payload(event) + if verdict.outcome == CLEAN: + return [] + rule_id = RULE_NARROWED_MUTATION if verdict.outcome == HIT else RULE_UNREADABLE + command = _event_command(event) + return [ + Finding( + rule_id=rule_id, + subject=_finding_subject(event, command), + message=verdict.message, + evidence=command, + ) + ] diff --git a/engine/hooks/categorical-scope-guard/tests/test_hooks.py b/engine/hooks/categorical-scope-guard/tests/test_hooks.py index bd53c461..cb2ebff3 100644 --- a/engine/hooks/categorical-scope-guard/tests/test_hooks.py +++ b/engine/hooks/categorical-scope-guard/tests/test_hooks.py @@ -381,7 +381,7 @@ def test_non_shell_tool_is_ignored(self): def test_garbage_payload_fails_open_with_log(self): result = subprocess.run([sys.executable, ENTRYPOINT], input="not json", capture_output=True, text=True) self.assertEqual(result.returncode, 0) - self.assertIn("not JSON", result.stderr) + self.assertIn("catstack-hook-error categorical-scope-guard: JSONDecodeError", result.stderr) if __name__ == "__main__": diff --git a/engine/hooks/categorical-scope-guard/tests/test_hooks_sdk_mode.py b/engine/hooks/categorical-scope-guard/tests/test_hooks_sdk_mode.py new file mode 100644 index 00000000..9f028f68 --- /dev/null +++ b/engine/hooks/categorical-scope-guard/tests/test_hooks_sdk_mode.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +from contextlib import contextmanager, redirect_stderr, redirect_stdout +from datetime import datetime, timezone +from io import StringIO +import json +import os +from pathlib import Path +import sys +import tempfile +import unittest +from unittest.mock import patch + +HERE = os.path.dirname(os.path.abspath(__file__)) +HOOK_DIR = os.path.dirname(HERE) +FIXTURES = os.path.join(HERE, "fixtures") +sys.path.insert(0, HOOK_DIR) + +import claude_pretooluse # noqa: E402 + +ALL_TASKS_FIRST = "can you make all tasks use claude and local executor" + + +def fixture(name: str) -> str: + with open(os.path.join(FIXTURES, name), encoding="utf-8") as handle: + return handle.read() + + +def human(text: str) -> dict: + return {"type": "user", "message": {"role": "user", "content": text}} + + +def write_transcript(entries: list[dict]) -> str: + handle = tempfile.NamedTemporaryFile("w", suffix=".jsonl", delete=False, encoding="utf-8") + with handle: + for entry in entries: + handle.write(json.dumps(entry) + "\n") + return handle.name + + +@contextmanager +def isolated_hook_env(**updates: str): + old = dict(os.environ) + try: + os.environ.update(updates) + yield + finally: + os.environ.clear() + os.environ.update(old) + + +def run_claude_pretooluse(payload: dict) -> tuple[int, str, str]: + stdin = StringIO(json.dumps(payload)) + stdout = StringIO() + stderr = StringIO() + with patch.object(sys, "stdin", stdin), redirect_stdout(stdout), redirect_stderr(stderr): + try: + claude_pretooluse.main() + except SystemExit as exc: + return int(exc.code or 0), stdout.getvalue(), stderr.getvalue() + return 0, stdout.getvalue(), stderr.getvalue() + + +def event_rows(metrics_dir: Path) -> list[dict[str, object]]: + today = datetime.now(timezone.utc).date().isoformat() + path = metrics_dir / f"events-{today}.jsonl" + return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()] + + +def blocking_payload(transcript_path: str) -> dict[str, object]: + return { + "hook_event_name": "PreToolUse", + "session_id": "cat-scope-sdk-mode", + "tool_name": "Bash", + "transcript_path": transcript_path, + "tool_input": {"command": fixture("update_tasks_status_in_pending_queued.txt")}, + } + + +class SdkModeTest(unittest.TestCase): + def test_mode_override_warn_changes_block_to_warning(self) -> None: + path = write_transcript([human(ALL_TASKS_FIRST)]) + try: + with tempfile.TemporaryDirectory() as tmp: + with isolated_hook_env( + CATSTACK_HOOK_METRICS_DIR=tmp, + CATSTACK_HOOK_MODE_CATEGORICAL_SCOPE_GUARD="warn", + ): + code, stdout, stderr = run_claude_pretooluse(blocking_payload(path)) + finally: + os.unlink(path) + + self.assertEqual(0, code) + self.assertEqual("", stderr) + body = json.loads(stdout) + output = body["hookSpecificOutput"] + self.assertEqual("PreToolUse", output["hookEventName"]) + self.assertIn(ALL_TASKS_FIRST, output["additionalContext"]) + + def test_each_finding_writes_one_event_row_with_rule_id(self) -> None: + path = write_transcript([human(ALL_TASKS_FIRST)]) + try: + with tempfile.TemporaryDirectory() as tmp: + with isolated_hook_env(CATSTACK_HOOK_METRICS_DIR=tmp): + code, _stdout, stderr = run_claude_pretooluse(blocking_payload(path)) + rows = event_rows(Path(tmp)) + finally: + os.unlink(path) + + self.assertEqual(2, code) + self.assertIn(ALL_TASKS_FIRST, stderr) + finding_rows = [row for row in rows if row["action"] == "stopped"] + self.assertEqual(1, len(finding_rows)) + self.assertEqual("categorical-scope-guard.narrowed-mutation", finding_rows[0]["rule_id"]) + self.assertTrue(all(row["hook"] == "categorical-scope-guard" for row in finding_rows)) + + +if __name__ == "__main__": + unittest.main() diff --git a/engine/hooks/demo-freeze/claude_pretooluse_check.py b/engine/hooks/demo-freeze/claude_pretooluse_check.py index 3ccc1f1d..399c7985 100755 --- a/engine/hooks/demo-freeze/claude_pretooluse_check.py +++ b/engine/hooks/demo-freeze/claude_pretooluse_check.py @@ -16,14 +16,21 @@ not haunt tomorrow's session), and any parse/read error fails open. """ import fnmatch -import json import os import sys import time +sys.path.insert(0, os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_sdk")) + +from finding import Finding # noqa: E402 +from runtime import run_hook # noqa: E402 + MARKER = os.environ.get("DEMO_FREEZE_FILE", "/tmp/.demo-freeze") MAX_AGE_SECS = 2 * 3600 +RULE_FROZEN_PATH = "demo-freeze.frozen-path" + def frozen_patterns(): try: @@ -48,28 +55,34 @@ def matches(target, pattern): return target_abs == os.path.abspath(pattern) -def main(): - try: - data = json.load(sys.stdin) - except json.JSONDecodeError: - return - tool_input = data.get("tool_input") or {} +def detect(event): + tool_input = event.get("tool_input") or {} target = ( tool_input.get("file_path") or tool_input.get("path") or tool_input.get("notebook_path") ) if not target: - return + return [] for pattern in frozen_patterns(): if matches(target, pattern): - sys.stderr.write( + message = ( f"Demo surface frozen: {target} matches {pattern!r} in {MARKER}. " "The user is mid-test — don't change what they're looking at unless " "they asked or the test is failing. Remove the marker file to " - "unfreeze once the live window ends.\n" + "unfreeze once the live window ends." ) - sys.exit(2) + return [Finding( + rule_id=RULE_FROZEN_PATH, + subject=target, + message=message, + evidence=pattern, + )] + return [] + + +def main(): + run_hook("demo-freeze", "claude", detect, "PreToolUse") if __name__ == "__main__": diff --git a/engine/hooks/demo-freeze/tests/test_hooks.py b/engine/hooks/demo-freeze/tests/test_hooks.py index ded538ab..56b60110 100755 --- a/engine/hooks/demo-freeze/tests/test_hooks.py +++ b/engine/hooks/demo-freeze/tests/test_hooks.py @@ -74,7 +74,10 @@ def test_no_marker_fails_open(self): with patch.object(claude_pretooluse_check, "MARKER", "/nonexistent/.demo-freeze"): with patch.object(sys, "stdin", io.StringIO(json.dumps(payload))): with redirect_stderr(err): - claude_pretooluse_check.main() + try: + claude_pretooluse_check.main() + except SystemExit: + pass self.assertEqual(err.getvalue(), "") def test_non_file_tool_input_passes(self): diff --git a/engine/hooks/demo-freeze/tests/test_hooks_sdk_mode.py b/engine/hooks/demo-freeze/tests/test_hooks_sdk_mode.py new file mode 100644 index 00000000..7f29969c --- /dev/null +++ b/engine/hooks/demo-freeze/tests/test_hooks_sdk_mode.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from io import StringIO +import json +import os +from pathlib import Path +import sys +import tempfile +import unittest +from contextlib import contextmanager, redirect_stderr, redirect_stdout +from unittest.mock import patch + +HERE = os.path.dirname(os.path.abspath(__file__)) +HOOK_DIR = os.path.dirname(HERE) +sys.path.insert(0, HOOK_DIR) + +import claude_pretooluse_check # noqa: E402 + +TARGET = "/tmp/demo/call.html" + + +@contextmanager +def isolated_hook_env(**updates: str): + old = dict(os.environ) + try: + os.environ.update(updates) + yield + finally: + os.environ.clear() + os.environ.update(old) + + +@contextmanager +def frozen_marker(*lines: str): + marker = tempfile.NamedTemporaryFile(mode="w", suffix=".freeze", delete=False) + marker.write("\n".join(lines) + "\n") + marker.close() + try: + with patch.object(claude_pretooluse_check, "MARKER", marker.name): + yield + finally: + os.unlink(marker.name) + + +def run_claude_pretooluse(payload: dict) -> tuple[int, str, str]: + stdin = StringIO(json.dumps(payload)) + stdout = StringIO() + stderr = StringIO() + with patch.object(sys, "stdin", stdin), redirect_stdout(stdout), redirect_stderr(stderr): + try: + claude_pretooluse_check.main() + except SystemExit as exc: + return int(exc.code or 0), stdout.getvalue(), stderr.getvalue() + return 0, stdout.getvalue(), stderr.getvalue() + + +def event_rows(metrics_dir: Path) -> list[dict[str, object]]: + today = datetime.now(timezone.utc).date().isoformat() + path = metrics_dir / f"events-{today}.jsonl" + return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()] + + +def blocking_payload() -> dict[str, object]: + return { + "hook_event_name": "PreToolUse", + "session_id": "demo-freeze-sdk-mode", + "tool_name": "Edit", + "tool_input": {"file_path": TARGET}, + } + + +class SdkModeTest(unittest.TestCase): + def test_mode_override_warn_changes_block_to_warning(self) -> None: + with frozen_marker("/tmp/demo/"): + with tempfile.TemporaryDirectory() as tmp: + with isolated_hook_env( + CATSTACK_HOOK_METRICS_DIR=tmp, + CATSTACK_HOOK_MODE_DEMO_FREEZE="warn", + ): + code, stdout, stderr = run_claude_pretooluse(blocking_payload()) + + self.assertEqual(0, code) + self.assertEqual("", stderr) + body = json.loads(stdout) + output = body["hookSpecificOutput"] + self.assertEqual("PreToolUse", output["hookEventName"]) + self.assertIn("Demo surface frozen", output["additionalContext"]) + + def test_each_finding_writes_one_event_row_with_rule_id(self) -> None: + with frozen_marker("/tmp/demo/"): + with tempfile.TemporaryDirectory() as tmp: + with isolated_hook_env(CATSTACK_HOOK_METRICS_DIR=tmp): + code, _stdout, stderr = run_claude_pretooluse(blocking_payload()) + rows = event_rows(Path(tmp)) + + self.assertEqual(2, code) + self.assertIn("Demo surface frozen", stderr) + finding_rows = [row for row in rows if row["action"] == "stopped"] + self.assertEqual(1, len(finding_rows)) + self.assertEqual("demo-freeze.frozen-path", finding_rows[0]["rule_id"]) + self.assertTrue(all(row["hook"] == "demo-freeze" for row in finding_rows)) + + +if __name__ == "__main__": + unittest.main()