[Hook architecture](15) Put the categorical-scope-guard hook onto the shared hook code

engine/hooks/_sdk/runtime.py

@@ -25,7 +25,7 @@ def run_hook(
         event = json.load(sys.stdin)
     except json.JSONDecodeError as exc:
         _write_findings_file([])
-        print(f"catstack-hook-error {hook}: JSONDecodeError: {exc}", file=sys.stderr)
+        print(f"catstack-hook-error {hook}: JSONDecodeError: hook payload is not JSON: {exc}", file=sys.stderr)
         stdout_text, _stderr_text, _exit_code = render(
             harness,
             hook_event_name or "",

engine/hooks/categorical-scope-guard/README.md

@@ -91,9 +91,12 @@ Do one of two things:
 
 ## Three outcomes, and the fail direction
 
-- **HIT** — exit 2 with the message above.
+- **HIT** — finding `categorical-scope-guard.partial-status-filter`. The
+  registry default is `stop`, so the shared runtime exits 2 with the
+  message above unless a local mode override lowers it to a warning.
 - **CLEAN** — exit 0.
-- **UNCHECKED** — exit 2. **This hook fails closed.** It blocks, and says
+- **UNCHECKED** — finding `categorical-scope-guard.unchecked`. **This hook
+  fails closed.** With the default `stop` mode it blocks, and says
   `UNCHECKED`, when:
   - the transcript path is absent, the file is missing, a line is malformed
     JSON (a torn final line is tolerated), the window runs past the 64 MB
@@ -132,8 +135,10 @@ reason and let the user answer.
 
 ## Files
 
-- `detect.py` — command parser, human-turn reader, `decide()`.
-- `claude_pretooluse.py` — the entrypoint.
+- `detect.py` — command parser, human-turn reader, `decide()`, and
+  `detect(event)`.
+- `claude_pretooluse.py`, `cursor_pretooluse.py`, `codex_pretooluse.py` —
+  thin runtime entrypoints.
 - `claude.hook.json`, `install_claude_hook.py` — the settings merge that
   `install.sh` runs.
 - `tests/test_hooks.py`, `tests/fixtures/` — the fixtures are real commands

engine/hooks/categorical-scope-guard/claude_pretooluse.py

@@ -1,55 +1,19 @@
 #!/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 Code PreToolUse entrypoint for categorical-scope-guard."""
 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/codex_pretooluse.py

@@ -0,0 +1,20 @@
+#!/usr/bin/env python3
+"""Codex PreToolUse entrypoint for categorical-scope-guard."""
+from __future__ import annotations
+
+import os
+import sys
+
+sys.path.insert(0, os.path.join(
+    os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_sdk"))
+
+from detect import detect  # noqa: E402
+from runtime import run_hook  # noqa: E402
+
+
+def main() -> None:
+    run_hook("categorical-scope-guard", "codex", detect, "PreToolUse")
+
+
+if __name__ == "__main__":
+    main()

engine/hooks/categorical-scope-guard/cursor_pretooluse.py

@@ -0,0 +1,20 @@
+#!/usr/bin/env python3
+"""Cursor PreToolUse entrypoint for categorical-scope-guard."""
+from __future__ import annotations
+
+import os
+import sys
+
+sys.path.insert(0, os.path.join(
+    os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_sdk"))
+
+from detect import detect  # noqa: E402
+from runtime import run_hook  # noqa: E402
+
+
+def main() -> None:
+    run_hook("categorical-scope-guard", "cursor", detect, "PreToolUse")
+
+
+if __name__ == "__main__":
+    main()

engine/hooks/categorical-scope-guard/detect.py

@@ -14,14 +14,24 @@
 from __future__ import annotations
 import sys
 
+import hashlib
 import json
 import os
 import re
 from dataclasses import dataclass, field
+from pathlib import Path
+
+SDK_DIR = Path(__file__).resolve().parents[1] / "_sdk"
+if str(SDK_DIR) not in sys.path:
+    sys.path.insert(0, str(SDK_DIR))
+
+from finding import Finding  # noqa: E402
 
 HIT = "hit"
 CLEAN = "clean"
 UNCHECKED = "unchecked"
+RULE_PARTIAL_STATUS_FILTER = "categorical-scope-guard.partial-status-filter"
+RULE_UNCHECKED = "categorical-scope-guard.unchecked"
 
 LIVE_TURNS = 4
 MAX_SCAN_BYTES = 64 * 1024 * 1024
@@ -872,3 +882,54 @@ 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 ""))
+
+
+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 ""
+    )
+
+
+def _command_subject(payload: dict) -> str:
+    tool_input = payload.get("tool_input") or payload.get("toolInput") or {}
+    command = tool_input.get("command") if isinstance(tool_input, dict) else ""
+    if not isinstance(command, str):
+        command = ""
+    return "command:" + hashlib.sha256(command.encode("utf-8")).hexdigest()[:16]
+
+
+def _finding(rule_id: str, payload: dict, message: str) -> Finding:
+    return Finding(
+        rule_id=rule_id,
+        subject=_command_subject(payload),
+        message=message,
+        evidence=message,
+    )
+
+
+def detect(event: dict) -> list[Finding]:
+    if _tool_name(event) not in SHELL_LIKE_TOOL_NAMES:
+        return []
+    try:
+        verdict = decide_payload(event)
+    except Exception as exc:
+        print(f"catstack-hook-error categorical-scope-guard: {type(exc).__name__}: {exc}", file=sys.stderr)
+        message = (
+            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."
+        )
+        return [_finding(RULE_UNCHECKED, event, message)]
+    if verdict.outcome == CLEAN:
+        return []
+    rule_id = RULE_PARTIAL_STATUS_FILTER if verdict.outcome == HIT else RULE_UNCHECKED
+    return [_finding(rule_id, event, verdict.message)]

engine/hooks/categorical-scope-guard/tests/test_hooks_sdk_mode.py

@@ -0,0 +1,105 @@
+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)
+FIXTURES = os.path.join(HERE, "fixtures")
+ENTRYPOINT = os.path.join(HOOK_DIR, "claude_pretooluse.py")
+
+
+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[str, object]:
+    return {"type": "user", "message": {"role": "user", "content": text}}
+
+
+def write_transcript(entries: list[dict[str, object]]) -> 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
+
+
+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:
+        path = write_transcript([human("can you make all tasks use claude and local executor")])
+        try:
+            result = run_entrypoint(
+                {
+                    "tool_name": "Bash",
+                    "hook_event_name": "PreToolUse",
+                    "transcript_path": path,
+                    "tool_input": {"command": fixture("update_tasks_status_in_pending_queued.txt")},
+                },
+                {"CATSTACK_HOOK_MODE_CATEGORICAL_SCOPE_GUARD": "warn"},
+            )
+        finally:
+            os.unlink(path)
+
+        self.assertEqual(0, result.returncode, result.stderr)
+        self.assertEqual("", result.stderr)
+        rendered = json.loads(result.stdout)
+        self.assertIn(
+            "categorical-scope-guard",
+            rendered["hookSpecificOutput"]["additionalContext"],
+        )
+
+    def test_each_finding_writes_one_event_row_with_rule_id(self) -> None:
+        path = write_transcript([human("can you make all tasks use claude and local executor")])
+        try:
+            with tempfile.TemporaryDirectory() as tmp:
+                result = run_entrypoint(
+                    {
+                        "tool_name": "Bash",
+                        "hook_event_name": "PreToolUse",
+                        "session_id": "categorical-scope-guard-sdk-mode",
+                        "transcript_path": path,
+                        "tool_input": {"command": fixture("update_tasks_status_in_pending_queued.txt")},
+                    },
+                    {
+                        "CATSTACK_HOOK_METRICS_DIR": tmp,
+                        "CATSTACK_HOOK_MODE_CATEGORICAL_SCOPE_GUARD": "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(path)
+
+        self.assertEqual(0, result.returncode, result.stderr)
+        self.assertEqual(1, len(rows))
+        self.assertEqual("categorical-scope-guard", rows[0]["hook"])
+        self.assertEqual("categorical-scope-guard.partial-status-filter", 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()