diff --git a/engine/hooks/_runner/README.md b/engine/hooks/_runner/README.md
new file mode 100644
index 00000000..92f705de
--- /dev/null
+++ b/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.
diff --git a/engine/hooks/_runner/outcome.py b/engine/hooks/_runner/outcome.py
new file mode 100644
index 00000000..f4e7fed8
--- /dev/null
+++ b/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"
diff --git a/engine/hooks/_runner/run.py b/engine/hooks/_runner/run.py
new file mode 100644
index 00000000..bcc2fd7a
--- /dev/null
+++ b/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())
diff --git a/engine/hooks/_runner/tests/test_outcome.py b/engine/hooks/_runner/tests/test_outcome.py
new file mode 100644
index 00000000..2eb68d97
--- /dev/null
+++ b/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()
diff --git a/engine/hooks/_runner/tests/test_run.py b/engine/hooks/_runner/tests/test_run.py
new file mode 100644
index 00000000..cc5b2186
--- /dev/null
+++ b/engine/hooks/_runner/tests/test_run.py
@@ -0,0 +1,178 @@
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+
+RUNNER_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+
+class RunnerCLI(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.home = os.path.join(self.tmp.name, "home")
+        self.hooks_root = os.path.join(self.home, ".claude", "hooks")
+        self.runner_dir = os.path.join(self.hooks_root, "_runner")
+        self.fixture_dir = os.path.join(self.hooks_root, "fixture")
+        self.metrics_dir = os.path.join(self.tmp.name, "metrics")
+        os.makedirs(self.runner_dir)
+        os.makedirs(self.fixture_dir)
+        shutil.copy2(os.path.join(RUNNER_DIR, "run.py"), os.path.join(self.runner_dir, "run.py"))
+        shutil.copy2(os.path.join(RUNNER_DIR, "outcome.py"), os.path.join(self.runner_dir, "outcome.py"))
+        self._write_fixture("silent.py", "")
+        self._write_fixture("spoke.py", "import json\nprint(json.dumps({'hookSpecificOutput': {'additionalContext': 'hi'}}))\n")
+        self._write_fixture("block_exit2.py", "import sys\nsys.stderr.write('blocked\\n')\nsys.exit(2)\n")
+        self._write_fixture("block_json.py", "print('{\"decision\":\"block\",\"reason\":\"x\"}')\n")
+        self._write_fixture("crash.py", "raise RuntimeError('boom')\n")
+        self._write_fixture("slow.py", "import time\ntime.sleep(5)\n")
+        self._write_fixture("caught.py", "import sys\nsys.stderr.write('catstack-hook-error fixture: ValueError: x\\n')\n")
+
+    def _write_fixture(self, name: str, body: str) -> None:
+        with open(os.path.join(self.fixture_dir, name), "w", encoding="utf-8") as handle:
+            handle.write(body)
+
+    def _env(self, metrics_dir: str | None = None) -> dict[str, str]:
+        env = os.environ.copy()
+        env["CATSTACK_HOOK_METRICS_DIR"] = self.metrics_dir if metrics_dir is None else metrics_dir
+        return env
+
+    def _stdin(self) -> bytes:
+        return json.dumps({"hook_event_name": "PromptSubmit", "session_id": "s1"}).encode()
+
+    def _direct(self, script: str) -> subprocess.CompletedProcess[bytes]:
+        return subprocess.run(
+            [sys.executable, os.path.join(self.fixture_dir, script)],
+            input=self._stdin(),
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            env=self._env(),
+        )
+
+    def _runner(self, script: str, *args: str, metrics_dir: str | None = None) -> subprocess.CompletedProcess[bytes]:
+        return subprocess.run(
+            [sys.executable, os.path.join(self.runner_dir, "run.py"), *args, f"fixture/{script}"],
+            input=self._stdin(),
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            env=self._env(metrics_dir),
+        )
+
+    def _row(self) -> dict[str, object]:
+        with open(os.path.join(self.metrics_dir, "runs.jsonl"), encoding="utf-8") as handle:
+            rows = [json.loads(line) for line in handle]
+        self.assertEqual(len(rows), 1)
+        return rows[0]
+
+    def test_recording_failure_still_forwards_hook_result(self) -> None:
+        with open(os.path.join(self.runner_dir, "outcome.py"), "w", encoding="utf-8") as handle:
+            handle.write("def classify(*args, **kwargs):\n    raise RuntimeError('classify broke')\n")
+        direct = self._direct("spoke.py")
+        wrapped = self._runner("spoke.py")
+        self.assertEqual(wrapped.stdout, direct.stdout)
+        self.assertEqual(wrapped.returncode, direct.returncode)
+        self.assertIn(b"catstack-hook-metrics: could not record run: RuntimeError: classify broke", wrapped.stderr)
+
+    def _assert_run_matches_direct(self, script: str, outcome: str) -> None:
+        direct = self._direct(script)
+        wrapped = self._runner(script)
+        self.assertEqual(wrapped.stdout, direct.stdout)
+        self.assertEqual(wrapped.stderr, direct.stderr)
+        self.assertEqual(wrapped.returncode, direct.returncode)
+        row = self._row()
+        self.assertEqual(row["outcome"], outcome)
+        self.assertEqual(row["harness"], "claude")
+        self.assertEqual(row["hook"], "fixture")
+        self.assertEqual(row["script"], script)
+        self.assertEqual(row["event"], "PromptSubmit")
+        self.assertEqual(row["session_id"], "s1")
+        self.assertEqual(row["exit_code"], direct.returncode)
+        self.assertEqual(row["stdout_bytes"], len(direct.stdout))
+
+    def test_silent_hook_stays_silent(self):
+        self._assert_run_matches_direct("silent.py", "silent")
+
+    def test_spoke_hook_keeps_stdout_bytes(self):
+        self._assert_run_matches_direct("spoke.py", "spoke")
+
+    def test_exit_two_hook_blocks(self):
+        self._assert_run_matches_direct("block_exit2.py", "blocked")
+
+    def test_block_json_hook_blocks(self):
+        self._assert_run_matches_direct("block_json.py", "blocked")
+
+    def test_crash_hook_crashes(self):
+        self._assert_run_matches_direct("crash.py", "crashed")
+
+    def test_caught_error_hook_is_caught(self):
+        self._assert_run_matches_direct("caught.py", "caught_error")
+
+    def test_slow_hook_times_out(self):
+        wrapped = self._runner("slow.py", "--timeout", "1")
+        self.assertEqual(wrapped.stdout, b"")
+        self.assertIn(b"catstack-hook-runner: fixture/slow.py timed out after 1s\n", wrapped.stderr)
+        self.assertEqual(wrapped.returncode, 1)
+        row = self._row()
+        self.assertEqual(row["outcome"], "timed_out")
+        self.assertEqual(row["harness"], "claude")
+        self.assertEqual(row["hook"], "fixture")
+        self.assertEqual(row["script"], "slow.py")
+        self.assertEqual(row["event"], "PromptSubmit")
+        self.assertEqual(row["exit_code"], 1)
+
+    def test_missing_hook_script_records_crash(self):
+        wrapped = self._runner("missing.py")
+        self.assertEqual(wrapped.stdout, b"")
+        self.assertEqual(wrapped.returncode, 1)
+        self.assertIn(b"catstack-hook-runner: no such hook script:", wrapped.stderr)
+        row = self._row()
+        self.assertEqual(row["outcome"], "crashed")
+        self.assertEqual(row["hook"], "fixture")
+        self.assertEqual(row["script"], "missing.py")
+        self.assertEqual(row["exit_code"], 1)
+
+    def test_metrics_write_failure_adds_one_stderr_line(self):
+        direct = self._direct("spoke.py")
+        metrics_file = os.path.join(self.tmp.name, "metrics-file")
+        with open(metrics_file, "w", encoding="utf-8") as handle:
+            handle.write("")
+        wrapped = self._runner("spoke.py", metrics_dir=metrics_file)
+        self.assertEqual(wrapped.stdout, direct.stdout)
+        self.assertEqual(wrapped.returncode, direct.returncode)
+        self.assertEqual(direct.stderr, b"")
+        self.assertIn(b"catstack-hook-metrics: could not write row", wrapped.stderr)
+        self.assertEqual(len([line for line in wrapped.stderr.splitlines() if line]), 1)
+
+    def test_non_json_stdin_records_null_event_and_conversation_id_fallback(self):
+        wrapped = subprocess.run(
+            [sys.executable, os.path.join(self.runner_dir, "run.py"), "fixture/silent.py"],
+            input=b"not json",
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            env=self._env(),
+        )
+        self.assertEqual(wrapped.returncode, 0)
+        row = self._row()
+        self.assertIsNone(row["event"])
+        self.assertIsNone(row["session_id"])
+
+    def test_conversation_id_records_session_id_when_session_id_absent(self):
+        wrapped = subprocess.run(
+            [sys.executable, os.path.join(self.runner_dir, "run.py"), "fixture/silent.py"],
+            input=json.dumps({"hook_event_name": "Stop", "conversation_id": "c1"}).encode(),
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            env=self._env(),
+        )
+        self.assertEqual(wrapped.returncode, 0)
+        row = self._row()
+        self.assertEqual(row["event"], "Stop")
+        self.assertEqual(row["session_id"], "c1")
+
+
+if __name__ == "__main__":
+    unittest.main()