[Hook architecture](16) Put the demo-freeze hook onto the shared hook code

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.

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

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

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

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,
+        )
+    ]

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

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