merge queue: checking #1332 on main (facdeaf), stacked on #1333 and #1336

engine/hooks/_runner/README.md

@@ -125,6 +125,18 @@ hook bypasses the metrics runner: <command>
 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.
 
+Set `CATSTACK_HOOK_PAYLOAD_DIR` to record the raw hook-event payload bytes
+before the hook subprocess starts. Recording is off by default. When enabled,
+the runner writes one file named `<payload_hash>-<hook>.json` under that
+directory, where `<payload_hash>` is the first 12 hex characters of the
+stdin bytes' SHA-256 digest, leaves the hook's stdin/stdout and exit code
+unchanged, skips payloads larger than 256KB, and keeps at most 2000 recorded
+payload files by evicting the oldest files. Override those caps with
+`CATSTACK_HOOK_PAYLOAD_MAX_BYTES` and `CATSTACK_HOOK_PAYLOAD_KEEP`.
+
+Two identical consecutive events for the same hook produce the same filename,
+so the newer payload replaces the older identical payload.
+
 Each row contains:
 
 - `ts`: UTC timestamp for the recorded run.
@@ -236,3 +248,26 @@ unsupported --since value: <value>
 
 With `--json`, the same report is printed as JSON with `registered`,
 `unregistered`, `malformed_rows`, `config_warnings`, and `window_rows`.
+
+## Replay Recorded Payloads
+
+Usage:
+
+```sh
+python3 scripts/test/replay_hook_payloads.py \
+  --payload-dir /tmp/hook-payloads \
+  --fleet detector='python3 ~/.claude/hooks/example/check.py' \
+  --dispatcher detector='python3 ~/.claude/hooks/_runner/dispatch.py --event Stop --timeout 10' \
+  --report-json /tmp/hook-replay-report.json
+```
+
+Each `--fleet` and `--dispatcher` value is `label=command`; labels must match on
+both sides. The command runs once per recorded payload with that payload on
+stdin. Replay sets `CATSTACK_HOOK_METRICS_DIR` and known hook state env vars to a
+temporary scratch tree and unsets `CATSTACK_HOOK_PAYLOAD_DIR`, so it does not
+write the live `runs.jsonl` or recursively record replay payloads.
+
+The script prints a table with one `match` or `mismatch` row per
+payload/label pair and writes the JSON report when `--report-json` is provided.
+The equivalence check compares stdout bytes, extracted context, block verdict,
+exit code, and reported findings. It exits 0 only when every row matches.

engine/hooks/_runner/dispatch.py

@@ -424,7 +424,7 @@ def _timeout(_signum, _frame):
         if isinstance(exc, SystemExit):
             exit_code = exc.code if isinstance(exc.code, int) else 1
         else:
-            exit_code = 0
+            exit_code = 1
         stdout = b""
         stderr = f"catstack-hook-error {hook}: {type(exc).__name__}: {exc}\n".encode()
     finally:

engine/hooks/_runner/run.py

@@ -20,8 +20,13 @@
 WELL_KNOWN_PYTHON_DIRS = ("/opt/homebrew/bin", "/usr/local/bin", os.path.expanduser("~/.local/bin"))
 METRICS_MAX_BYTES_ENV = "CATSTACK_HOOK_METRICS_MAX_BYTES"
 METRICS_KEEP_ENV = "CATSTACK_HOOK_METRICS_KEEP"
+PAYLOAD_DIR_ENV = "CATSTACK_HOOK_PAYLOAD_DIR"
+PAYLOAD_MAX_BYTES_ENV = "CATSTACK_HOOK_PAYLOAD_MAX_BYTES"
+PAYLOAD_KEEP_ENV = "CATSTACK_HOOK_PAYLOAD_KEEP"
 DEFAULT_METRICS_MAX_BYTES = 20 * 1024 * 1024
 DEFAULT_METRICS_KEEP = 3
+DEFAULT_PAYLOAD_MAX_BYTES = 256 * 1024
+DEFAULT_PAYLOAD_KEEP = 2000
 REPLY_EVENTS = ("Stop", "SubagentStop")
 FENCE = "```"
 SKIP_MACHINE_DELIVERABLE = "machine-deliverable"
@@ -153,6 +158,40 @@ def _write_metrics(row: dict[str, object], path: str) -> bytes:
     return message
 
 
+def _payload_path(root: str, event_uid: str, hook: str) -> str:
+    safe_hook = "".join(char if char.isalnum() or char in "._-" else "_" for char in hook) or "unknown"
+    return os.path.join(root, f"{event_uid}-{safe_hook}.json")
+
+
+def _evict_payloads(root: str, keep: int) -> None:
+    entries = [
+        entry
+        for entry in os.scandir(root)
+        if entry.is_file() and entry.name.endswith(".json")
+    ]
+    for entry in sorted(entries, key=lambda item: (item.stat().st_mtime_ns, item.name))[: max(0, len(entries) - keep)]:
+        os.unlink(entry.path)
+
+
+def _record_payload(stdin: bytes, hook: str) -> bytes:
+    root = os.environ.get(PAYLOAD_DIR_ENV)
+    if not root:
+        return b""
+    max_bytes, _max_message = _positive_env(PAYLOAD_MAX_BYTES_ENV, DEFAULT_PAYLOAD_MAX_BYTES)
+    keep, _keep_message = _positive_env(PAYLOAD_KEEP_ENV, DEFAULT_PAYLOAD_KEEP)
+    if len(stdin) > max_bytes:
+        return b""
+    try:
+        os.makedirs(root, exist_ok=True)
+        event_uid = hashlib.sha256(stdin).hexdigest()[:12]
+        with open(_payload_path(root, event_uid, hook), "wb") as handle:
+            handle.write(stdin)
+        _evict_payloads(root, keep)
+    except OSError as exc:
+        return f"catstack-hook-payloads: could not record payload: {exc}\n".encode()
+    return b""
+
+
 def _make_findings_file() -> str:
     fd, path = tempfile.mkstemp(prefix="catstack-hook-findings-", suffix=".json")
     with os.fdopen(fd, "w", encoding="utf-8") as handle:
@@ -260,6 +299,7 @@ def main(argv: list[str] | None = None) -> int:
     timed_out = False
     findings_path = _make_findings_file()
     findings_error = b""
+    payload_error = b""
     rule_ids: list[str] = []
     hook_ran = False
     skipped = None if args.notify else _skip_reason(stdin)
@@ -278,6 +318,7 @@ def main(argv: list[str] | None = None) -> int:
                 f"Set {PYTHON_OVERRIDE_ENV} to a newer interpreter.\n"
             ).encode()
         else:
+            payload_error = _record_payload(meta, hook)
             env = os.environ.copy()
             env["CATSTACK_HOOK_FINDINGS_FILE"] = findings_path
             proc = subprocess.Popen(
@@ -321,6 +362,7 @@ def main(argv: list[str] | None = None) -> int:
     sys.stdout.buffer.flush()
     sys.stderr.buffer.write(stderr)
     sys.stderr.buffer.write(findings_error)
+    sys.stderr.buffer.write(payload_error)
     sys.stderr.buffer.write(metrics_error)
     sys.stderr.buffer.flush()
     return exit_code

engine/hooks/_runner/tests/chaos/exit_hook.py

@@ -0,0 +1,4 @@
+from __future__ import annotations
+
+def detect(event):
+    raise SystemExit(3)

engine/hooks/_runner/tests/chaos/hang_hook.py

@@ -0,0 +1,8 @@
+from __future__ import annotations
+
+import time
+
+
+def detect(event):
+    time.sleep(30)
+    return []

engine/hooks/_runner/tests/chaos/ok_hook.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+from finding import Finding
+
+
+def detect(event):
+    hook_name = str(event.get("_catstack_current_hook") or "chaos-ok")
+    return [
+        Finding(
+            rule_id=f"{hook_name}.spoke",
+            subject=hook_name,
+            message=f"{hook_name} sibling spoke",
+            evidence="chaos fixture",
+        )
+    ]

engine/hooks/_runner/tests/chaos/raise_hook.py

@@ -0,0 +1,5 @@
+from __future__ import annotations
+
+
+def detect(event):
+    raise RuntimeError("chaos fixture boom")

engine/hooks/_runner/tests/test_dispatch.py

@@ -10,6 +10,7 @@
 from unittest import mock
 
 RUNNER_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+CHAOS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "chaos")
 
 
 class DispatchCLI(unittest.TestCase):
@@ -55,6 +56,19 @@ def _write_hook(
         with open(os.path.join(hook_dir, "claude.hook.json"), "w", encoding="utf-8") as handle:
             json.dump(manifest, handle)
 
+    def _write_chaos_sdk_hook(self, name: str, module_file: str) -> None:
+        with open(os.path.join(CHAOS_DIR, module_file), encoding="utf-8") as handle:
+            module_body = handle.read()
+        module_name = os.path.splitext(module_file)[0]
+        self._write_hook(
+            name,
+            "fallback.py",
+            "raise SystemExit('chaos SDK fixture unexpectedly used subprocess fallback')\n",
+            event="Stop",
+            subagent_stop={"dispatch": {"entry": f"{module_name}:detect"}},
+            extra_files={module_file: module_body},
+        )
+
     def _env(self) -> dict[str, str]:
         env = os.environ.copy()
         env["HOME"] = self.home
@@ -261,6 +275,41 @@ def test_sdk_entrypoint_runs_in_process_and_emits_rule_ids(self):
         self.assertEqual(row["dispatch_path"], "sdk")
         self.assertEqual(row["rule_ids"], ["fixture.sdk"])
 
+    def test_sdk_chaos_failures_each_emit_rows_without_silencing_siblings(self):
+        self._install_sdk()
+        self._write_chaos_sdk_hook("chaos-a-ok", "ok_hook.py")
+        self._write_chaos_sdk_hook("chaos-b-raise", "raise_hook.py")
+        self._write_chaos_sdk_hook("chaos-c-hang", "hang_hook.py")
+        self._write_chaos_sdk_hook("chaos-d-exit", "exit_hook.py")
+        self._write_chaos_sdk_hook("chaos-e-ok", "ok_hook.py")
+        mode_env = {
+            "CATSTACK_HOOK_MODE_CHAOS_A_OK": "warn",
+            "CATSTACK_HOOK_MODE_CHAOS_E_OK": "warn",
+        }
+        env_patch = mock.patch.dict(os.environ, mode_env)
+        env_patch.start()
+        self.addCleanup(env_patch.stop)
+
+        result = self._run(event="SubagentStop", timeout="1")
+
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertIn(b"chaos-ok sibling spoke", result.stdout)
+        rows = {row["hook"]: row for row in self._rows()}
+        self.assertEqual(
+            set(rows),
+            {"chaos-a-ok", "chaos-b-raise", "chaos-c-hang", "chaos-d-exit", "chaos-e-ok"},
+        )
+        self.assertEqual(rows["chaos-a-ok"]["outcome"], "spoke")
+        self.assertEqual(rows["chaos-e-ok"]["outcome"], "spoke")
+        self.assertEqual(rows["chaos-b-raise"]["outcome"], "crashed")
+        self.assertEqual(rows["chaos-c-hang"]["outcome"], "timed_out")
+        self.assertEqual(rows["chaos-d-exit"]["outcome"], "crashed")
+        self.assertEqual(rows["chaos-b-raise"]["exit_code"], 1)
+        self.assertEqual(rows["chaos-c-hang"]["exit_code"], 1)
+        self.assertEqual(rows["chaos-d-exit"]["exit_code"], 3)
+        self.assertEqual(rows["chaos-b-raise"]["dispatch_path"], "sdk_error")
+        self.assertEqual(rows["chaos-c-hang"]["dispatch_path"], "sdk_error")
+        self.assertEqual(rows["chaos-d-exit"]["dispatch_path"], "sdk_error")
 
 FIXTURES_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "fixtures")
 FIXTURE_HOOK = "fixture-extra-manifest"

engine/hooks/_runner/tests/test_run.py

@@ -1,14 +1,18 @@
 from __future__ import annotations
 
+import hashlib
 import json
 import os
 import shutil
 import subprocess
 import sys
 import tempfile
 import unittest
+from unittest import mock
 
 RUNNER_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, RUNNER_DIR)
+import run  # noqa: E402
 
 
 class RunnerCLI(unittest.TestCase):
@@ -42,9 +46,13 @@ 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]:
+    def _env(self, metrics_dir: str | None = None, payload_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
+        if payload_dir is not None:
+            env["CATSTACK_HOOK_PAYLOAD_DIR"] = payload_dir
+        else:
+            env.pop("CATSTACK_HOOK_PAYLOAD_DIR", None)
         return env
 
     def _stdin(self) -> bytes:
@@ -59,13 +67,19 @@ def _direct(self, script: str) -> subprocess.CompletedProcess[bytes]:
             env=self._env(),
         )
 
-    def _runner(self, script: str, *args: str, metrics_dir: str | None = None) -> subprocess.CompletedProcess[bytes]:
+    def _runner(
+        self,
+        script: str,
+        *args: str,
+        metrics_dir: str | None = None,
+        payload_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),
+            env=self._env(metrics_dir, payload_dir),
         )
 
     def _row(self) -> dict[str, object]:
@@ -100,6 +114,42 @@ def _assert_run_matches_direct(self, script: str, outcome: str) -> None:
         self.assertEqual(row["rule_ids"], [])
         self.assertEqual(row["stdout_bytes"], len(direct.stdout))
 
+    def test_payload_recorder_is_inert_without_env(self):
+        with mock.patch("run.os.makedirs") as makedirs, mock.patch("run.open", mock.mock_open()) as opened:
+            self.assertEqual(run._record_payload(b'{"x":1}', "fixture"), b"")
+        makedirs.assert_not_called()
+        opened.assert_not_called()
+
+    def test_payload_recorder_writes_stdin_when_enabled(self):
+        payload_dir = os.path.join(self.tmp.name, "payloads")
+        stdin = self._stdin()
+
+        wrapped = self._runner("silent.py", payload_dir=payload_dir)
+
+        self.assertEqual((wrapped.returncode, wrapped.stdout, wrapped.stderr), (0, b"", b""))
+        event_uid = hashlib.sha256(stdin).hexdigest()[:12]
+        with open(os.path.join(payload_dir, f"{event_uid}-fixture.json"), "rb") as handle:
+            self.assertEqual(handle.read(), stdin)
+
+    def test_payload_recorder_evicts_old_payloads_past_keep_cap(self):
+        payload_dir = os.path.join(self.tmp.name, "payloads")
+        with mock.patch.dict(
+            os.environ,
+            {
+                "CATSTACK_HOOK_PAYLOAD_DIR": payload_dir,
+                "CATSTACK_HOOK_PAYLOAD_KEEP": "1",
+            },
+            clear=False,
+        ):
+            self.assertEqual(run._record_payload(b"first", "fixture"), b"")
+            first_name = os.listdir(payload_dir)[0]
+            os.utime(os.path.join(payload_dir, first_name), (1, 1))
+            self.assertEqual(run._record_payload(b"second", "fixture"), b"")
+
+        names = os.listdir(payload_dir)
+        self.assertEqual(len(names), 1)
+        self.assertEqual(names[0], f"{hashlib.sha256(b'second').hexdigest()[:12]}-fixture.json")
+
     def test_notify_mode_reads_the_payload_argument_not_stdin(self):
         argv_out = os.path.join(self.tmp.name, "argv.json")
         self._write_fixture("notify.py", f"import json, sys\njson.dump(sys.argv[1:], open({argv_out!r}, 'w'))\n")

engine/hooks/_runner/tests/test_wrap_installed.py

@@ -24,6 +24,9 @@ def setUp(self):
         self.old_home = os.environ.get("HOME")
         os.environ["HOME"] = str(self.home)
         self.addCleanup(self._restore_home)
+        dispatcher_patch = mock.patch.dict(os.environ, {"CATSTACK_HOOK_DISPATCHER": "0"})
+        dispatcher_patch.start()
+        self.addCleanup(dispatcher_patch.stop)
         self.claude_path = self.home / ".claude" / "settings.json"
         self.cursor_path = self.home / ".cursor" / "hooks.json"
         self.codex_path = self.home / ".codex" / "hooks.json"