merge queue: checking #731 on main (4c4fcd5)

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__":

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):

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()