Hook fleet (2) replay harness

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/run.py

@@ -2,6 +2,7 @@
 
 import argparse
 import datetime
+import hashlib
 import json
 import os
 import subprocess
@@ -19,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"
@@ -152,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:
@@ -258,6 +298,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)
@@ -276,6 +317,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(
@@ -319,6 +361,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/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")