diff --git a/engine/hooks/demo-freeze/claude_pretooluse_check.py b/engine/hooks/demo-freeze/claude_pretooluse_check.py index 3ccc1f1d..4d94563d 100755 --- a/engine/hooks/demo-freeze/claude_pretooluse_check.py +++ b/engine/hooks/demo-freeze/claude_pretooluse_check.py @@ -16,13 +16,19 @@ 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(): @@ -48,28 +54,35 @@ 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 {} - target = ( +def _tool_target(event): + tool_input = event.get("tool_input") or event.get("toolInput") or {} + if not isinstance(tool_input, dict): + return None + return ( tool_input.get("file_path") or tool_input.get("path") or tool_input.get("notebook_path") ) + + +def detect(event): + target = _tool_target(event) 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=message)] + 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..11eabc11 --- /dev/null +++ b/engine/hooks/demo-freeze/tests/test_hooks_sdk_mode.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +import json +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +HERE = os.path.dirname(os.path.abspath(__file__)) +HOOK_DIR = os.path.dirname(HERE) +ENTRYPOINT = os.path.join(HOOK_DIR, "claude_pretooluse_check.py") + + +def write_marker(lines: list[str]) -> str: + handle = tempfile.NamedTemporaryFile("w", suffix=".demo-freeze", delete=False, encoding="utf-8") + with handle: + handle.write("\n".join(lines) + "\n") + return handle.name + + +def run_entrypoint(payload: dict[str, object], env: dict[str, str]) -> subprocess.CompletedProcess[str]: + merged_env = os.environ.copy() + merged_env.update(env) + return subprocess.run( + [sys.executable, ENTRYPOINT], + input=json.dumps(payload), + capture_output=True, + text=True, + env=merged_env, + ) + + +class SdkModeTest(unittest.TestCase): + def test_mode_override_warn_changes_block_to_warning(self) -> None: + marker = write_marker(["/tmp/demo/call.html"]) + try: + result = run_entrypoint( + {"tool_name": "Edit", "tool_input": {"file_path": "/tmp/demo/call.html"}}, + { + "DEMO_FREEZE_FILE": marker, + "CATSTACK_HOOK_MODE_DEMO_FREEZE": "warn", + }, + ) + finally: + os.unlink(marker) + + self.assertEqual(0, result.returncode, result.stderr) + self.assertEqual("", result.stderr) + rendered = json.loads(result.stdout) + self.assertIn( + "Demo surface frozen", + rendered["hookSpecificOutput"]["additionalContext"], + ) + + def test_mode_stop_still_blocks_by_default(self) -> None: + marker = write_marker(["/tmp/demo/call.html"]) + try: + result = run_entrypoint( + {"tool_name": "Edit", "tool_input": {"file_path": "/tmp/demo/call.html"}}, + {"DEMO_FREEZE_FILE": marker}, + ) + finally: + os.unlink(marker) + + self.assertEqual(2, result.returncode) + self.assertIn("Demo surface frozen", result.stderr) + + def test_each_finding_writes_one_event_row_with_rule_id(self) -> None: + marker = write_marker(["/tmp/demo/call.html"]) + try: + with tempfile.TemporaryDirectory() as tmp: + result = run_entrypoint( + { + "tool_name": "Edit", + "session_id": "demo-freeze-sdk-mode", + "tool_input": {"file_path": "/tmp/demo/call.html"}, + }, + { + "DEMO_FREEZE_FILE": marker, + "CATSTACK_HOOK_METRICS_DIR": tmp, + "CATSTACK_HOOK_MODE_DEMO_FREEZE": "warn", + }, + ) + rows = [ + json.loads(line) + for file in Path(tmp).glob("events-*.jsonl") + for line in file.read_text(encoding="utf-8").splitlines() + ] + finally: + os.unlink(marker) + + self.assertEqual(0, result.returncode, result.stderr) + self.assertEqual(1, len(rows)) + self.assertEqual("demo-freeze", rows[0]["hook"]) + self.assertEqual("demo-freeze.frozen-path", rows[0]["rule_id"]) + self.assertEqual("warn", rows[0]["mode"]) + self.assertEqual("override", rows[0]["mode_source"]) + self.assertEqual("warned", rows[0]["action"]) + + +if __name__ == "__main__": + unittest.main()