Hook metrics (1) a runner records every hook run

engine/hooks/_runner/README.md

@@ -0,0 +1,48 @@
+# Hook runner
+
+Usage:
+
+```sh
+python3 engine/hooks/_runner/run.py [--timeout SECONDS] <hook>/<script.py> [args...]
+```
+
+The runner reads stdin, runs the hook script in a subprocess with that stdin,
+passes through the hook's stdout, stderr, and exit code, then appends one JSONL
+metrics row.
+
+Rows are written to `~/.cache/catstack-hook-metrics/runs.jsonl` by default. Set
+`CATSTACK_HOOK_METRICS_DIR` to write `runs.jsonl` under a different directory.
+
+Each row contains:
+
+- `ts`: UTC timestamp for the recorded run.
+- `harness`: `claude`, `cursor`, `codex`, or `unknown`, inferred from the hooks path.
+- `hook`: the first path segment from `<hook>/<script.py>`.
+- `script`: the rest of the hook script path after `hook`.
+- `event`: `hook_event_name` from JSON stdin, or `null`.
+- `session_id`: `session_id` from JSON stdin, falling back to `conversation_id`, or `null`.
+- `outcome`: classified result for the run.
+- `exit_code`: hook process exit code recorded by the runner.
+- `duration_ms`: elapsed runner time in milliseconds.
+- `stdout_bytes`: number of stdout bytes emitted by the hook.
+- `stderr_tail`: final 500 decoded stderr characters, with invalid UTF-8 replaced.
+
+Outcome precedence is:
+
+1. `timed_out` when the runner timeout kills the hook.
+2. `blocked` when `exit_code` is `2`.
+3. `crashed` when `exit_code` is any other nonzero value.
+4. `caught_error` when any stderr line starts with `catstack-hook-error `.
+5. `blocked` when stdout is a JSON object with `decision: "block"`, `continue: false`,
+   `hookSpecificOutput.permissionDecision: "deny"`, or `permission: "deny"`.
+6. `spoke` when stdout has non-whitespace bytes.
+7. `silent` otherwise.
+
+If a metrics row cannot be written, the runner appends one stderr line after the
+hook stderr:
+
+```text
+catstack-hook-metrics: could not write row to <path>: <error>
+```
+
+Nothing calls this runner until install wiring lands.

engine/hooks/_runner/outcome.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+import json
+
+OUTCOMES = {
+    "timed_out",
+    "crashed",
+    "blocked",
+    "caught_error",
+    "spoke",
+    "silent",
+}
+
+
+def _stderr_has_hook_error(stderr: bytes) -> bool:
+    return any(line.startswith(b"catstack-hook-error ") for line in stderr.splitlines())
+
+
+def _stdout_blocks(stdout: bytes) -> bool:
+    try:
+        payload = json.loads(stdout.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        return False
+    if not isinstance(payload, dict):
+        return False
+    if payload.get("decision") == "block":
+        return True
+    if payload.get("continue") is False:
+        return True
+    hook_output = payload.get("hookSpecificOutput")
+    if isinstance(hook_output, dict) and hook_output.get("permissionDecision") == "deny":
+        return True
+    return payload.get("permission") == "deny"
+
+
+def classify(exit_code: int | None, stdout: bytes, stderr: bytes, timed_out: bool) -> str:
+    if timed_out:
+        return "timed_out"
+    if exit_code == 2:
+        return "blocked"
+    if exit_code not in (0, None):
+        return "crashed"
+    if _stderr_has_hook_error(stderr):
+        return "caught_error"
+    if _stdout_blocks(stdout):
+        return "blocked"
+    if stdout.strip():
+        return "spoke"
+    return "silent"

engine/hooks/_runner/run.py

@@ -0,0 +1,150 @@
+from __future__ import annotations
+
+import argparse
+import datetime
+import json
+import os
+import subprocess
+import sys
+import time
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from outcome import classify
+
+
+def _hooks_root() -> str:
+    return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+
+def _harness(hooks_root: str) -> str:
+    for name in ("claude", "cursor", "codex"):
+        if f"/.{name}/" in hooks_root:
+            return name
+    return "unknown"
+
+
+def _stdin_fields(stdin: bytes) -> tuple[str | None, str | None]:
+    try:
+        payload = json.loads(stdin.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        return None, None
+    if not isinstance(payload, dict):
+        return None, None
+    session_id = payload.get("session_id")
+    if session_id is None:
+        session_id = payload.get("conversation_id")
+    return payload.get("hook_event_name"), session_id
+
+
+def _metrics_path() -> str:
+    root = os.environ.get("CATSTACK_HOOK_METRICS_DIR")
+    if not root:
+        root = os.path.expanduser(os.path.join("~", ".cache", "catstack-hook-metrics"))
+    return os.path.join(root, "runs.jsonl")
+
+
+def _write_metrics(row: dict[str, object], path: str) -> bytes:
+    try:
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        with open(path, "a", encoding="utf-8") as handle:
+            handle.write(json.dumps(row, sort_keys=True, separators=(",", ":")) + "\n")
+    except OSError as exc:
+        return f"catstack-hook-metrics: could not write row to {path}: {exc}\n".encode()
+    return b""
+
+
+def _format_timeout(seconds: float) -> str:
+    if seconds == int(seconds):
+        return str(int(seconds))
+    return str(seconds)
+
+
+def _parse_args(argv: list[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--timeout", type=float)
+    parser.add_argument("hook_script")
+    parser.add_argument("args", nargs=argparse.REMAINDER)
+    return parser.parse_args(argv)
+
+
+def _row(
+    hooks_root: str,
+    hook: str,
+    script: str,
+    stdin: bytes,
+    outcome: str,
+    exit_code: int | None,
+    started: float,
+    stdout: bytes,
+    stderr: bytes,
+) -> dict[str, object]:
+    event, session_id = _stdin_fields(stdin)
+    return {
+        "ts": datetime.datetime.now(datetime.timezone.utc).isoformat(),
+        "harness": _harness(hooks_root),
+        "hook": hook,
+        "script": script,
+        "event": event,
+        "session_id": session_id,
+        "outcome": outcome,
+        "exit_code": exit_code,
+        "duration_ms": int((time.monotonic() - started) * 1000),
+        "stdout_bytes": len(stdout),
+        "stderr_tail": stderr.decode("utf-8", errors="replace")[-500:],
+    }
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = _parse_args(sys.argv[1:] if argv is None else argv)
+    started = time.monotonic()
+    stdin = sys.stdin.buffer.read()
+    hooks_root = _hooks_root()
+    hook, script = args.hook_script.split("/", 1) if "/" in args.hook_script else (args.hook_script, "")
+    script_path = os.path.join(hooks_root, hook, script)
+    stdout = b""
+    stderr = b""
+    exit_code = 1
+    timed_out = False
+
+    if not script or not os.path.isfile(script_path):
+        stderr = f"catstack-hook-runner: no such hook script: {script_path}\n".encode()
+    else:
+        proc = subprocess.Popen(
+            [sys.executable, script_path, *args.args],
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            cwd=os.getcwd(),
+            env=os.environ.copy(),
+        )
+        try:
+            stdout, stderr = proc.communicate(stdin, timeout=args.timeout)
+            exit_code = proc.returncode
+        except subprocess.TimeoutExpired:
+            proc.kill()
+            proc.communicate()
+            timed_out = True
+            stdout = b""
+            stderr = (
+                f"catstack-hook-runner: {args.hook_script} timed out after "
+                f"{_format_timeout(args.timeout)}s\n"
+            ).encode()
+            exit_code = 1
+
+    try:
+        outcome = classify(exit_code, stdout, stderr, timed_out)
+        row = _row(hooks_root, hook, script, stdin, outcome, exit_code, started, stdout, stderr)
+        metrics_error = _write_metrics(row, _metrics_path())
+    except Exception as exc:
+        metrics_error = f"catstack-hook-metrics: could not record run: {type(exc).__name__}: {exc}\n".encode()
+    sys.stdout.buffer.write(stdout)
+    sys.stdout.buffer.flush()
+    sys.stderr.buffer.write(stderr)
+    sys.stderr.buffer.write(metrics_error)
+    sys.stderr.buffer.flush()
+    return exit_code
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

engine/hooks/_runner/tests/test_outcome.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+
+RUNNER_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, RUNNER_DIR)
+
+import outcome
+
+
+class ClassifyOutcomes(unittest.TestCase):
+    def test_timed_out_wins(self):
+        self.assertEqual(outcome.classify(2, b'{"decision":"block"}', b"", True), "timed_out")
+
+    def test_exit_two_blocks_before_crash(self):
+        self.assertEqual(outcome.classify(2, b"", b"", False), "blocked")
+
+    def test_other_nonzero_exit_crashes(self):
+        self.assertEqual(outcome.classify(1, b'{"decision":"block"}', b"", False), "crashed")
+
+    def test_stderr_hook_error_is_caught_error(self):
+        self.assertEqual(outcome.classify(0, b"", b"catstack-hook-error x\n", False), "caught_error")
+
+    def test_stderr_hook_error_requires_line_start(self):
+        self.assertEqual(outcome.classify(0, b"", b"x catstack-hook-error y\n", False), "silent")
+
+    def test_json_decision_block_blocks(self):
+        self.assertEqual(outcome.classify(0, b'{"decision":"block"}', b"", False), "blocked")
+
+    def test_json_continue_false_blocks(self):
+        self.assertEqual(outcome.classify(0, b'{"continue":false}', b"", False), "blocked")
+
+    def test_json_permission_decision_deny_blocks(self):
+        data = b'{"hookSpecificOutput":{"permissionDecision":"deny"}}'
+        self.assertEqual(outcome.classify(0, data, b"", False), "blocked")
+
+    def test_json_permission_deny_blocks(self):
+        self.assertEqual(outcome.classify(0, b'{"permission":"deny"}', b"", False), "blocked")
+
+    def test_non_json_stdout_speaks(self):
+        self.assertEqual(outcome.classify(0, b"{not json", b"", False), "spoke")
+
+    def test_json_array_stdout_speaks(self):
+        self.assertEqual(outcome.classify(0, b"[1]", b"", False), "spoke")
+
+    def test_whitespace_stdout_is_silent(self):
+        self.assertEqual(outcome.classify(0, b" \n\t", b"", False), "silent")
+
+    def test_empty_stdout_is_silent(self):
+        self.assertEqual(outcome.classify(0, b"", b"", False), "silent")
+
+
+if __name__ == "__main__":
+    unittest.main()