From 8001939b61c9a2b6bc60da6b657b1b09ad337c46 Mon Sep 17 00:00:00 2001 From: Edbert Chan Date: Sun, 11 Oct 2026 14:53:12 +0000 Subject: [PATCH 01/10] Add an opt-in hook-payload recorder and a replay-equivalence harness Review claim: - An opt-in payload recorder in the runner plus a replay script can feed recorded hook-event payloads through installed hook commands and produce a verdict-equivalence report (stdout/context/block/exit) per detector. Review lane: behavior Safety invariant: - Recording is off unless CATSTACK_HOOK_PAYLOAD_DIR is set; when on it never alters hook stdin/stdout/exit and caps each payload (default 256KB) and total dir (default 2000 files, oldest evicted). - Replay runs against a scratch metrics/state dir, never live settings or the live runs.jsonl. Goal: - Runner: when env var set, append the raw stdin payload to $CATSTACK_HOOK_PAYLOAD_DIR/-.json before spawning the hook child, where is sha256(stdin)[:12]. - scripts/test/replay_hook_payloads.py: for each recorded payload, run the hook command under a scratch env, capture stdout/stderr/exit/ findings; emit an equivalence report JSON + table comparing two command sets (fleet vs dispatcher args taken as parameters). Motivation: - The catastrophic regression risk of the dispatcher is a detector that silently stops speaking; the only credible gate is byte-identical verdicts on real payloads. Cherry-picked from origin/plan/hook-fleet-dispatcher-measure-the-herd-then-dispatch-it (e356e8c8), scoped to only this claim's files. Dropped this slice's dependency on the separate metrics-event-uid claim (17ef1517, not on this branch): the payload recorder computes its own local hash instead of reusing a published `event_uid` metrics-row field, and the README wording and test imports were adjusted to match. Co-Authored-By: Claude Sonnet 5 --- engine/hooks/_runner/README.md | 35 +++ engine/hooks/_runner/run.py | 43 ++++ engine/hooks/_runner/tests/test_run.py | 56 ++++- scripts/test/replay_hook_payloads.py | 246 ++++++++++++++++++++++ scripts/test/test_replay_hook_payloads.py | 100 +++++++++ 5 files changed, 477 insertions(+), 3 deletions(-) create mode 100644 scripts/test/replay_hook_payloads.py create mode 100644 scripts/test/test_replay_hook_payloads.py diff --git a/engine/hooks/_runner/README.md b/engine/hooks/_runner/README.md index 1b5768d45..f410ae929 100644 --- a/engine/hooks/_runner/README.md +++ b/engine/hooks/_runner/README.md @@ -125,6 +125,18 @@ hook bypasses the metrics runner: 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 `-.json` under that +directory, where `` 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: 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. diff --git a/engine/hooks/_runner/run.py b/engine/hooks/_runner/run.py index 081747e3c..414f75d80 100644 --- a/engine/hooks/_runner/run.py +++ b/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 diff --git a/engine/hooks/_runner/tests/test_run.py b/engine/hooks/_runner/tests/test_run.py index 563b95519..07fc1e565 100644 --- a/engine/hooks/_runner/tests/test_run.py +++ b/engine/hooks/_runner/tests/test_run.py @@ -1,5 +1,6 @@ from __future__ import annotations +import hashlib import json import os import shutil @@ -7,8 +8,11 @@ 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") diff --git a/scripts/test/replay_hook_payloads.py b/scripts/test/replay_hook_payloads.py new file mode 100644 index 000000000..e52686ba8 --- /dev/null +++ b/scripts/test/replay_hook_payloads.py @@ -0,0 +1,246 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import base64 +import hashlib +import json +import os +from pathlib import Path +import shutil +import subprocess +import sys +import tempfile +from dataclasses import dataclass + +RUNNER_DIR = Path(__file__).resolve().parents[2] / "engine" / "hooks" / "_runner" +sys.path.insert(0, str(RUNNER_DIR)) + +import outcome # noqa: E402 + +STATE_ENVS = ( + "CATSTACK_HOOK_REMINDER_STATE_DIR", + "CATSTACK_NARROW_THE_SCOPE_STATE_DIR", + "CATSTACK_LLM_JUDGE_STATE_DIR", + "CATSTACK_UNVERIFIED_TAG_CHECK_STATE_DIR", + "CATSTACK_BUILD_THE_LEVER_STATE_DIR", + "CATSTACK_TAG_LEDGER_DIR", + "CATSTACK_SPLIT_SCOPE_STATE_DIR", +) + + +@dataclass(frozen=True) +class CommandSpec: + label: str + command: str + + +def _parse_command_spec(raw: str) -> CommandSpec: + label, sep, command = raw.partition("=") + if not sep or not label or not command: + raise argparse.ArgumentTypeError("expected LABEL=COMMAND") + return CommandSpec(label, command) + + +def _read_json_object(raw: bytes) -> dict[str, object] | None: + try: + payload = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + return None + return payload if isinstance(payload, dict) else None + + +def _context(stdout: bytes) -> str: + payload = _read_json_object(stdout) + if payload is None: + return stdout.decode("utf-8", errors="replace").strip() + for key in ("additionalContext", "additional_context", "message", "reason", "user_message"): + value = payload.get(key) + if isinstance(value, str) and value: + return value + hook_output = payload.get("hookSpecificOutput") + if isinstance(hook_output, dict): + for key in ("additionalContext", "permissionDecisionReason", "reason"): + value = hook_output.get(key) + if isinstance(value, str) and value: + return value + return "" + + +def _rule_ids(metrics_dir: Path, findings_path: Path) -> list[str]: + seen: set[str] = set() + ordered: list[str] = [] + + def add(values: object) -> None: + if isinstance(values, list): + for value in values: + if isinstance(value, str) and value not in seen: + seen.add(value) + ordered.append(value) + + try: + add(json.loads(findings_path.read_text(encoding="utf-8"))) + except (OSError, UnicodeDecodeError, json.JSONDecodeError): + pass + runs = metrics_dir / "runs.jsonl" + try: + with runs.open(encoding="utf-8") as handle: + for line in handle: + row = json.loads(line) + if isinstance(row, dict): + add(row.get("rule_ids")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError): + pass + return ordered + + +def _scratch_env(base: Path, side: str, label: str, payload_name: str) -> tuple[dict[str, str], Path, Path]: + root = base / side / label / payload_name + metrics_dir = root / "metrics" + state_dir = root / "state" + metrics_dir.mkdir(parents=True, exist_ok=True) + state_dir.mkdir(parents=True, exist_ok=True) + findings_path = root / "findings.json" + findings_path.write_text("[]", encoding="utf-8") + env = os.environ.copy() + env["CATSTACK_HOOK_METRICS_DIR"] = str(metrics_dir) + env["CATSTACK_HOOK_FINDINGS_FILE"] = str(findings_path) + env.pop("CATSTACK_HOOK_PAYLOAD_DIR", None) + for name in STATE_ENVS: + env[name] = str(state_dir / name.lower()) + return env, metrics_dir, findings_path + + +def _run_command(spec: CommandSpec, payload: bytes, payload_name: str, side: str, scratch: Path, timeout: float) -> dict: + env, metrics_dir, findings_path = _scratch_env(scratch, side, spec.label, payload_name) + timed_out = False + try: + proc = subprocess.run( + spec.command, + shell=True, + input=payload, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=env, + timeout=timeout, + ) + stdout = proc.stdout + stderr = proc.stderr + exit_code = proc.returncode + except subprocess.TimeoutExpired as exc: + timed_out = True + stdout = exc.stdout or b"" + stderr = (exc.stderr or b"") + f"replay_hook_payloads: timed out after {timeout}s\n".encode() + exit_code = 1 + result = outcome.classify(exit_code, stdout, stderr, timed_out) + return { + "command": spec.command, + "exit_code": exit_code, + "outcome": result, + "blocked": result == "blocked", + "context": _context(stdout), + "findings": _rule_ids(metrics_dir, findings_path), + "stdout_b64": base64.b64encode(stdout).decode("ascii"), + "stdout_sha256": hashlib.sha256(stdout).hexdigest(), + "stdout_text": stdout.decode("utf-8", errors="replace"), + "stderr_b64": base64.b64encode(stderr).decode("ascii"), + "stderr_text": stderr.decode("utf-8", errors="replace"), + } + + +def _by_label(specs: list[CommandSpec], name: str) -> dict[str, CommandSpec]: + result: dict[str, CommandSpec] = {} + for spec in specs: + if spec.label in result: + raise SystemExit(f"duplicate {name} label: {spec.label}") + result[spec.label] = spec + return result + + +def _comparison_keys(verdict: dict) -> dict: + return { + "exit_code": verdict["exit_code"], + "blocked": verdict["blocked"], + "context": verdict["context"], + "findings": verdict["findings"], + "stdout_b64": verdict["stdout_b64"], + } + + +def _payload_files(payload_dir: Path) -> list[Path]: + return sorted(path for path in payload_dir.glob("*.json") if path.is_file()) + + +def _print_table(rows: list[dict]) -> None: + print("payload detector verdict fleet_exit dispatcher_exit fleet_outcome dispatcher_outcome") + for row in rows: + verdict = "match" if row["match"] else "mismatch" + print( + f"{row['payload']} {row['detector']} {verdict} " + f"{row['fleet']['exit_code']} {row['dispatcher']['exit_code']} " + f"{row['fleet']['outcome']} {row['dispatcher']['outcome']}" + ) + + +def _parse_args(argv: list[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--payload-dir", required=True) + parser.add_argument("--fleet", action="append", type=_parse_command_spec, required=True) + parser.add_argument("--dispatcher", action="append", type=_parse_command_spec, required=True) + parser.add_argument("--timeout", type=float, default=30.0) + parser.add_argument("--report-json") + parser.add_argument("--keep-scratch", action="store_true") + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = _parse_args(sys.argv[1:] if argv is None else argv) + payload_dir = Path(args.payload_dir) + payloads = _payload_files(payload_dir) + if not payloads: + raise SystemExit(f"no recorded payloads in {payload_dir}") + fleet = _by_label(args.fleet, "fleet") + dispatcher = _by_label(args.dispatcher, "dispatcher") + if set(fleet) != set(dispatcher): + missing_dispatcher = sorted(set(fleet) - set(dispatcher)) + missing_fleet = sorted(set(dispatcher) - set(fleet)) + raise SystemExit( + f"fleet/dispatcher labels differ: missing dispatcher={missing_dispatcher} missing fleet={missing_fleet}" + ) + + scratch = Path(tempfile.mkdtemp(prefix="catstack-hook-replay-")) + rows: list[dict] = [] + try: + for payload_path in payloads: + payload = payload_path.read_bytes() + for label in sorted(fleet): + fleet_verdict = _run_command(fleet[label], payload, payload_path.name, "fleet", scratch, args.timeout) + dispatcher_verdict = _run_command( + dispatcher[label], payload, payload_path.name, "dispatcher", scratch, args.timeout + ) + rows.append( + { + "payload": payload_path.name, + "detector": label, + "match": _comparison_keys(fleet_verdict) == _comparison_keys(dispatcher_verdict), + "fleet": fleet_verdict, + "dispatcher": dispatcher_verdict, + } + ) + finally: + if args.keep_scratch: + print(f"scratch={scratch}", file=sys.stderr) + else: + shutil.rmtree(scratch) + + report = {"payload_dir": str(payload_dir), "match": all(row["match"] for row in rows), "rows": rows} + if args.report_json: + with open(args.report_json, "w", encoding="utf-8") as handle: + json.dump(report, handle, indent=2, sort_keys=True) + handle.write("\n") + _print_table(rows) + return 0 if report["match"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/test/test_replay_hook_payloads.py b/scripts/test/test_replay_hook_payloads.py new file mode 100644 index 000000000..263fce353 --- /dev/null +++ b/scripts/test/test_replay_hook_payloads.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +import hashlib +import json +import os +from pathlib import Path +import shlex +import subprocess +import sys +import tempfile +import unittest + +SCRIPT = Path(__file__).resolve().parent / "replay_hook_payloads.py" + + +class ReplayHookPayloadsCLI(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) + self.payload_dir = self.root / "payloads" + self.payload_dir.mkdir() + self.hook = self.root / "hook.py" + self.hook.write_text( + "import json\n" + "import os\n" + "import sys\n" + "payload = sys.stdin.buffer.read().decode('utf-8')\n" + "mode = sys.argv[1]\n" + "with open(os.environ['CATSTACK_HOOK_FINDINGS_FILE'], 'w', encoding='utf-8') as handle:\n" + " json.dump(['fixture.rule'], handle)\n" + "if mode == 'block':\n" + " print(json.dumps({'decision': 'block', 'reason': payload}))\n" + "elif mode == 'exit2':\n" + " sys.stderr.write(payload + '\\n')\n" + " sys.exit(2)\n" + "else:\n" + " print(json.dumps({'hookSpecificOutput': {'additionalContext': mode + ':' + payload}}))\n", + encoding="utf-8", + ) + + def _payload(self, body: bytes = b'{"hello":"world"}') -> str: + name = f"{hashlib.sha256(body).hexdigest()[:12]}-fixture.json" + (self.payload_dir / name).write_bytes(body) + return name + + def _command(self, mode: str) -> str: + return f"{shlex.quote(sys.executable)} {shlex.quote(str(self.hook))} {shlex.quote(mode)}" + + def _run(self, fleet_mode: str, dispatcher_mode: str) -> subprocess.CompletedProcess[str]: + report = self.root / "report.json" + return subprocess.run( + [ + sys.executable, + str(SCRIPT), + "--payload-dir", + str(self.payload_dir), + "--fleet", + f"fixture={self._command(fleet_mode)}", + "--dispatcher", + f"fixture={self._command(dispatcher_mode)}", + "--report-json", + str(report), + ], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + + def test_matching_commands_emit_match_table_and_json_report(self) -> None: + payload_name = self._payload() + + result = self._run("same", "same") + + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn(f"{payload_name} fixture match 0 0 spoke spoke", result.stdout) + with open(self.root / "report.json", encoding="utf-8") as handle: + report = json.load(handle) + self.assertTrue(report["match"]) + self.assertEqual(report["rows"][0]["detector"], "fixture") + self.assertEqual(report["rows"][0]["fleet"]["findings"], ["fixture.rule"]) + + def test_mismatching_context_returns_nonzero_and_marks_the_row(self) -> None: + self._payload() + + result = self._run("fleet", "dispatcher") + + self.assertEqual(result.returncode, 1, result.stdout + result.stderr) + self.assertIn("fixture mismatch 0 0 spoke spoke", result.stdout) + with open(self.root / "report.json", encoding="utf-8") as handle: + report = json.load(handle) + self.assertFalse(report["match"]) + self.assertNotEqual( + report["rows"][0]["fleet"]["context"], + report["rows"][0]["dispatcher"]["context"], + ) + + +if __name__ == "__main__": + unittest.main() From aa14b4f0ef863052e94dc2e3e65774cd0a31bc41 Mon Sep 17 00:00:00 2001 From: Invoker Bot Date: Sun, 11 Oct 2026 14:57:30 +0000 Subject: [PATCH 02/10] =?UTF-8?q?invoker:=20wf-1791728212248-12/publish-re?= =?UTF-8?q?play=20=E2=80=94=20Review=20claim:=20An=20opt-in=20payload=20re?= =?UTF-8?q?corder=20in=20the=20runner=20plus=20a=20replay=20script=20can?= =?UTF-8?q?=20feed=20recorded=20hook-event=20payloads=20through=20installe?= =?UTF-8?q?d=20hook=20commands=20and=20produce=20a=20verdict-equivalence?= =?UTF-8?q?=20report=20per=20detector.=20Review=20lane:=20behavior=20Safet?= =?UTF-8?q?y=20invariant:=20Hook=20stdout=20and=20exit=20codes=20stay=20th?= =?UTF-8?q?e=20same.=20This=20slice=20does=20not=20change=20the=20other=20?= =?UTF-8?q?four=20claims.=20Files:=20engine/hooks/=5Frunner/run.py,=20scri?= =?UTF-8?q?pts/test/replay=5Fhook=5Fpayloads.py=20Change=20types:=20bugfix?= =?UTF-8?q?=20Acceptance=20criteria:=20The=20branch=20contains=20only=20th?= =?UTF-8?q?is=20claim,=20and=20the=20review=20step=20publishes=20one=20pul?= =?UTF-8?q?l=20request.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Exit code: 0 Invoker-Finalize-Id: 61e213a0-a6cd-4950-ba23-0acc17742997 From 005c4768fa6e90477bcea3edfb7d1b21a16adc18 Mon Sep 17 00:00:00 2001 From: Invoker Bot Date: Thu, 1 Oct 2026 11:56:12 +0000 Subject: [PATCH 03/10] Run the SubagentStop hook fleet through one dispatcher process. The fleet self-saturates: one SubagentStop burst spawned 15-30 processes that each re-parsed the same transcript, measured at 15.1s p50 wall idle against 1.55s for a single process, with per-hook durations escalating in launch order. dispatch.py now runs opted-in detectors in-process, sequentially, parsing the event once and passing the same parsed payload to every detector. A hook opts in with subagent_stop.dispatch in its manifest, naming a MODULE:FUNCTION entry point (default detect:detect). A detector with no resolvable entry point still runs as a subprocess and its metrics row is marked subprocess_fallback. Each detector keeps its own verdict, its own runs.jsonl row under its own hook name, and its own rule_ids; a detector that raises, exits, or overruns its 4s soft budget is caught and recorded without stopping its siblings. The event budget defaults to 9s for SubagentStop. mirror_stop_hooks_to_subagent_stop.py emits one dispatcher entry for the opted-in hooks and leaves every other hook's direct entry alone. (cherry picked from commit 531f5c86 on plan/hook-fleet-dispatcher-measure-the-herd-then-dispatch-it) --- engine/hooks/_runner/dispatch.py | 207 +++++++++++++++++- engine/hooks/_runner/tests/test_dispatch.py | 106 +++++++-- .../mirror_stop_hooks_to_subagent_stop.py | 19 +- tests/test_install.py | 15 +- tests/test_mirror_subagent_stop.py | 22 +- 5 files changed, 339 insertions(+), 30 deletions(-) diff --git a/engine/hooks/_runner/dispatch.py b/engine/hooks/_runner/dispatch.py index 7f1dd37f2..35743450d 100644 --- a/engine/hooks/_runner/dispatch.py +++ b/engine/hooks/_runner/dispatch.py @@ -1,11 +1,12 @@ from __future__ import annotations import argparse -import concurrent.futures import glob +import importlib.util import json import os import re +import signal import subprocess import sys import time @@ -16,7 +17,11 @@ import run DEFAULT_BUDGET = 59.5 +DEFAULT_SUBAGENT_STOP_EVENT_BUDGET = 9.0 +DEFAULT_SUBAGENT_STOP_DETECTOR_BUDGET = 4.0 OPT_OUT_KEY = "subagent_stop" +DISPATCH_KEY = "dispatch" +DEFAULT_DISPATCH_ENTRY = "detect:detect" MIRRORED_EVENTS = {"SubagentStop": "Stop"} DIRECT_RE = re.compile(r"^python3 \$HOME/\.(claude|cursor|codex)/hooks/([^/\s]+)/([^/\s]+\.py)((?:\s+.*)?)$") MESSAGE_KEYS = ("reason", "message", "additionalContext", "additional_context", "user_message") @@ -157,6 +162,11 @@ def _load_manifest_hooks( opt_out = manifest.get(OPT_OUT_KEY) if isinstance(opt_out, dict) and opt_out.get("inherit") is False: continue + dispatch_entry = _dispatch_entry(manifest) + if dispatch_entry is None: + continue + else: + dispatch_entry = None for entry in entries: matcher = None if not mirrored and isinstance(entry, dict): @@ -178,11 +188,26 @@ def _load_manifest_hooks( "args": list(args), "timeout": hook.get("timeout"), "matcher": matcher, + "dispatch_entry": dispatch_entry, } ) return records, warnings +def _dispatch_entry(manifest: dict) -> str | None: + config = manifest.get(OPT_OUT_KEY) + if not isinstance(config, dict): + return None + raw = config.get(DISPATCH_KEY) + if raw is True: + return DEFAULT_DISPATCH_ENTRY + if isinstance(raw, dict): + entry = raw.get("entry") + if isinstance(entry, str) and entry.strip(): + return entry.strip() + return None + + def _parse_command(command: str, harness: str) -> tuple[str, str, str] | None: match = DIRECT_RE.fullmatch(command) if not match or match.group(1) != harness: @@ -211,7 +236,22 @@ def _payload(stdin: bytes) -> dict[str, object]: return parsed if isinstance(parsed, dict) else {} -def _run_one(hooks_root: str, python: str, record: dict, stdin: bytes, budget: float) -> dict: +def _run_one( + hooks_root: str, + python: str, + record: dict, + stdin: bytes, + budget: float, + parsed_event: dict[str, object] | None = None, +) -> dict: + if record.get("dispatch_entry") and _entrypoint_exists(hooks_root, record["hook"], str(record["dispatch_entry"])): + return _run_sdk_one(hooks_root, record, stdin, budget, parsed_event) + return _run_subprocess_one(hooks_root, python, record, stdin, budget, fallback=True) + + +def _run_subprocess_one( + hooks_root: str, python: str, record: dict, stdin: bytes, budget: float, fallback: bool = False +) -> dict: hook, script = record["hook"], record["script"] started = time.monotonic() script_path = os.path.join(hooks_root, hook, script) @@ -249,6 +289,8 @@ def _run_one(hooks_root: str, python: str, record: dict, stdin: bytes, budget: f run._delete_findings_file(findings_path) result_outcome = outcome.classify(exit_code, stdout, stderr, timed_out) row = run._row(hooks_root, hook, script, stdin, result_outcome, exit_code, started, stdout, stderr, rule_ids) + if fallback: + row["dispatch_path"] = "subprocess_fallback" return { "hook": hook, "script": script, @@ -260,12 +302,158 @@ def _run_one(hooks_root: str, python: str, record: dict, stdin: bytes, budget: f } +def _sdk_dirs(hooks_root: str) -> tuple[str, str]: + engine_hooks = os.path.abspath(hooks_root) + sdk_dir = os.path.join(engine_hooks, "_sdk") + return engine_hooks, sdk_dir + + +def _prefer_import_path(path: str, index: int) -> None: + try: + sys.path.remove(path) + except ValueError: + pass + sys.path.insert(index, path) + + +def _load_entrypoint(hooks_root: str, hook: str, entry: str): + module_name, sep, function_name = entry.partition(":") + if not sep or not module_name or not function_name: + raise ValueError(f"dispatch entry must be MODULE:FUNCTION, got {entry!r}") + module_path = _entrypoint_path(hooks_root, hook, module_name) + if not os.path.isfile(module_path): + raise FileNotFoundError(module_path) + _engine_hooks, sdk_dir = _sdk_dirs(hooks_root) + _prefer_import_path(sdk_dir, 0) + _prefer_import_path(os.path.dirname(module_path), 1) + unique_name = f"_catstack_dispatch_{hook.replace('-', '_')}_{abs(hash(module_path))}" + spec = importlib.util.spec_from_file_location(unique_name, module_path) + if spec is None or spec.loader is None: + raise ImportError(f"could not load {module_path}") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + target = getattr(module, function_name) + if not callable(target): + raise TypeError(f"{entry!r} did not resolve to a callable") + return target + + +def _entrypoint_path(hooks_root: str, hook: str, module_name: str) -> str: + return os.path.join(hooks_root, hook, *module_name.split(".")) + ".py" + + +def _entrypoint_exists(hooks_root: str, hook: str, entry: str) -> bool: + module_name, sep, function_name = entry.partition(":") + return bool(sep and module_name and function_name and os.path.isfile(_entrypoint_path(hooks_root, hook, module_name))) + + +def _event(stdin: bytes, harness: str, event_name: str) -> dict[str, object]: + raw = stdin.decode("utf-8", errors="replace") + try: + parsed = json.loads(raw) + except json.JSONDecodeError as exc: + parsed = { + "_raw_payload": raw, + "_payload_error": f"the hook payload is not JSON ({exc})", + "_payload_error_type": type(exc).__name__, + "_payload_error_detail": str(exc), + } + if not isinstance(parsed, dict): + parsed = {"_raw_payload": raw, "_payload_error": "the hook payload is not a JSON object"} + parsed.setdefault("_raw_payload", raw) + parsed.setdefault("_catstack_harness", harness) + parsed.setdefault("hook_event_name", event_name) + return parsed + + +def _render_findings(hooks_root: str, hook: str, harness: str, event_name: str, event: dict[str, object], findings: list): + _engine_hooks, sdk_dir = _sdk_dirs(hooks_root) + _prefer_import_path(sdk_dir, 0) + from modes import effective_finding_modes + from render import render + from runtime import _renderable_findings + + mode, _mode_source, finding_modes = effective_finding_modes(hook, event, findings) + rendered_mode, rendered_findings = _renderable_findings(finding_modes, findings, mode) + stdout_text, stderr_text, exit_code = render(harness, event_name, rendered_mode, rendered_findings) + return stdout_text.encode(), stderr_text.encode(), exit_code, [finding.rule_id for finding in findings] + + +def _run_sdk_one( + hooks_root: str, + record: dict, + stdin: bytes, + budget: float, + parsed_event: dict[str, object] | None = None, +) -> dict: + hook, script = record["hook"], record["script"] + started = time.monotonic() + stdout = b"" + stderr = b"" + exit_code = 0 + timed_out = False + rule_ids: list[str] = [] + harness = run._harness(hooks_root) + payload = dict(parsed_event) if parsed_event is not None else _event(stdin, harness, "SubagentStop") + previous_handler = None + try: + detect = _load_entrypoint(hooks_root, hook, str(record["dispatch_entry"])) + if hasattr(signal, "SIGALRM"): + previous_handler = signal.getsignal(signal.SIGALRM) + + def _timeout(_signum, _frame): + raise TimeoutError(f"{hook}/{record['dispatch_entry']} timed out after {budget}s") + + signal.signal(signal.SIGALRM, _timeout) + signal.setitimer(signal.ITIMER_REAL, budget) + findings = detect(payload) + if not isinstance(findings, list): + raise TypeError("detect(event) must return a list") + stdout, stderr, exit_code, rule_ids = _render_findings(hooks_root, hook, harness, "SubagentStop", payload, findings) + lines = payload.get("_catstack_stderr_lines") + if isinstance(lines, list): + stderr += "".join(f"{line}\n" for line in lines if isinstance(line, str) and line).encode() + except TimeoutError: + timed_out = True + exit_code = 1 + stdout = b"" + stderr = f"catstack-hook-runner: {hook}/{script} timed out after {budget}s\n".encode() + except BaseException as exc: + if isinstance(exc, KeyboardInterrupt): + raise + if isinstance(exc, SystemExit): + exit_code = exc.code if isinstance(exc.code, int) else 1 + else: + exit_code = 0 + stdout = b"" + stderr = f"catstack-hook-error {hook}: {type(exc).__name__}: {exc}\n".encode() + finally: + if hasattr(signal, "SIGALRM"): + signal.setitimer(signal.ITIMER_REAL, 0) + if previous_handler is not None: + signal.signal(signal.SIGALRM, previous_handler) + result_outcome = outcome.classify(exit_code, stdout, stderr, timed_out) + row = run._row(hooks_root, hook, script, stdin, result_outcome, exit_code, started, stdout, stderr, rule_ids) + row["dispatch_path"] = "sdk" if not timed_out and not stderr.startswith(b"catstack-hook-error ") else "sdk_error" + return { + "hook": hook, + "script": script, + "outcome": result_outcome, + "exit_code": exit_code, + "stdout": stdout, + "stderr": stderr, + "row": row, + } + + def _merge_stdout(results: list[dict]) -> tuple[bytes, bytes, int]: blocked = [r for r in results if r["outcome"] == "blocked"] spoke = [r for r in results if r["stdout"].strip()] if blocked: return _merge_blocks(blocked) if spoke: + if len(spoke) == 1: + return spoke[0]["stdout"], b"", spoke[0]["exit_code"] contexts = [note for r in spoke for note in (_note(r),) if note] payload: dict[str, object] = {"continue": True} if contexts: @@ -340,11 +528,9 @@ def run_dispatch( ).encode() return 1, b"", stderr, [] - results: list[dict] = [] - with concurrent.futures.ThreadPoolExecutor(max_workers=len(records)) as pool: - futures = [pool.submit(_run_one, hooks_root, python, record, stdin, budget) for record in records] - for future in futures: - results.append(future.result()) + detector_budget = min(budget, DEFAULT_SUBAGENT_STOP_DETECTOR_BUDGET) if event == "SubagentStop" else budget + parsed_event = _event(stdin, harness, event) if event == "SubagentStop" else None + results = [_run_one(hooks_root, python, record, stdin, detector_budget, parsed_event) for record in records] stdout, block_notes, exit_code = _merge_stdout(results) stderr_lines = [r["stderr"] for r in results if r["stderr"]] @@ -355,7 +541,7 @@ def run_dispatch( def _parse_args(argv: list[str]) -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--event", required=True) - parser.add_argument("--timeout", type=float, default=DEFAULT_BUDGET) + parser.add_argument("--timeout", type=float) return parser.parse_args(argv) @@ -364,7 +550,10 @@ def main(argv: list[str] | None = None) -> int: stdin = sys.stdin.buffer.read() hooks_root = run._hooks_root() harness = run._harness(hooks_root) - exit_code, stdout, stderr, rows = run_dispatch(hooks_root, harness, args.event, stdin, args.timeout) + budget = args.timeout + if budget is None: + budget = DEFAULT_SUBAGENT_STOP_EVENT_BUDGET if args.event == "SubagentStop" else DEFAULT_BUDGET + exit_code, stdout, stderr, rows = run_dispatch(hooks_root, harness, args.event, stdin, budget) metrics_path = run._metrics_path() metrics_errors = b"".join(run._write_metrics(row, metrics_path) for row in rows) sys.stdout.buffer.write(stdout) diff --git a/engine/hooks/_runner/tests/test_dispatch.py b/engine/hooks/_runner/tests/test_dispatch.py index 05bd2a7c5..20dc6e02c 100644 --- a/engine/hooks/_runner/tests/test_dispatch.py +++ b/engine/hooks/_runner/tests/test_dispatch.py @@ -7,6 +7,7 @@ import sys import tempfile import unittest +from unittest import mock RUNNER_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) @@ -23,15 +24,34 @@ def setUp(self): for name in ("run.py", "outcome.py", "dispatch.py"): shutil.copy2(os.path.join(RUNNER_DIR, name), os.path.join(self.runner_dir, name)) - def _write_hook(self, name: str, script: str, body: str, event: str = "PreToolUse", matcher=None) -> None: + def _install_sdk(self) -> None: + hooks_dir = os.path.dirname(RUNNER_DIR) + shutil.copytree(os.path.join(hooks_dir, "_sdk"), os.path.join(self.hooks_root, "_sdk")) + shutil.copytree(os.path.join(hooks_dir, "_flags"), os.path.join(self.hooks_root, "_flags")) + + def _write_hook( + self, + name: str, + script: str, + body: str, + event: str = "PreToolUse", + matcher=None, + subagent_stop: dict[str, object] | None = None, + extra_files: dict[str, str] | None = None, + ) -> None: hook_dir = os.path.join(self.hooks_root, name) os.makedirs(hook_dir) with open(os.path.join(hook_dir, script), "w", encoding="utf-8") as handle: handle.write(body) + for filename, contents in (extra_files or {}).items(): + with open(os.path.join(hook_dir, filename), "w", encoding="utf-8") as handle: + handle.write(contents) entry = {"hooks": [{"type": "command", "command": f"python3 $HOME/.claude/hooks/{name}/{script}"}]} if matcher is not None: entry["matcher"] = matcher manifest = {"hooks": {event: [entry]}} + if subagent_stop is not None: + manifest["subagent_stop"] = subagent_stop with open(os.path.join(hook_dir, "claude.hook.json"), "w", encoding="utf-8") as handle: json.dump(manifest, handle) @@ -173,6 +193,74 @@ def test_matcher_on_the_manifest_filters_which_hooks_run(self): self.assertEqual(rows_outcome(self._rows(), "fixture-bash"), "spoke") self.assertEqual(self._rows(), [row for row in self._rows() if row["hook"] == "fixture-bash"]) + def test_subagent_stop_dispatch_only_loads_manifest_opted_in_hooks(self): + self._write_hook( + "fixture-dispatched", + "ok.py", + "import json\nprint(json.dumps({'hookSpecificOutput': {'additionalContext': 'dispatched'}}))\n", + event="Stop", + subagent_stop={"dispatch": True}, + ) + self._write_hook( + "fixture-direct", + "direct.py", + "import json\nprint(json.dumps({'hookSpecificOutput': {'additionalContext': 'direct'}}))\n", + event="Stop", + ) + + result = self._run(event="SubagentStop") + + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn(b"dispatched", result.stdout) + self.assertNotIn(b"direct", result.stdout) + rows = self._rows() + self.assertEqual([row["hook"] for row in rows], ["fixture-dispatched"]) + + def test_missing_sdk_entrypoint_falls_back_to_the_existing_script_and_marks_the_row(self): + self._write_hook( + "fixture-fallback", + "ok.py", + "import json\nprint(json.dumps({'hookSpecificOutput': {'additionalContext': 'fallback ran'}}))\n", + event="Stop", + subagent_stop={"dispatch": True}, + ) + + result = self._run(event="SubagentStop") + + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn(b"fallback ran", result.stdout) + row = self._row_for("fixture-fallback") + self.assertEqual(row["dispatch_path"], "subprocess_fallback") + + def test_sdk_entrypoint_runs_in_process_and_emits_rule_ids(self): + self._install_sdk() + self._write_hook( + "fixture-sdk", + "ok.py", + "print('subprocess should not run')\n", + event="Stop", + subagent_stop={"dispatch": {"entry": "detect:detect"}}, + extra_files={ + "detect.py": ( + "from finding import Finding\n" + "def detect(event):\n" + " return [Finding(rule_id='fixture.sdk', subject='s', message='sdk spoke', evidence='e')]\n" + ) + }, + ) + env_patch = mock.patch.dict(os.environ, {"CATSTACK_HOOK_MODE_FIXTURE_SDK": "warn"}) + env_patch.start() + self.addCleanup(env_patch.stop) + + result = self._run(event="SubagentStop") + + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn(b"sdk spoke", result.stdout) + self.assertNotIn(b"subprocess should not run", result.stdout) + row = self._row_for("fixture-sdk") + self.assertEqual(row["dispatch_path"], "sdk") + self.assertEqual(row["rule_ids"], ["fixture.sdk"]) + FIXTURES_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "fixtures") FIXTURE_HOOK = "fixture-extra-manifest" @@ -283,12 +371,7 @@ def test_a_speaking_hook_in_an_extra_manifest_reaches_dispatch_with_the_run_alon self.assertEqual(alone.stderr, expected_stderr) self.assertEqual(dispatched.returncode, 0, dispatched.stderr) - self.assertEqual( - dispatched.stdout, - json.dumps( - {"continue": True, "additionalContext": f"[{FIXTURE_HOOK}] {FIXTURE_MESSAGE}"} - ).encode(), - ) + self.assertEqual(dispatched.stdout, alone.stdout) self.assertEqual(dispatched.stderr, expected_stderr) alone_row = self._only_row(alone_rows) @@ -421,12 +504,7 @@ def test_a_speaking_installed_hook_reaches_dispatch_with_the_run_alone_contract( self.assertEqual(alone.stderr, expected_stderr) self.assertEqual(dispatched.returncode, 0, dispatched.stderr) - self.assertEqual( - dispatched.stdout, - json.dumps( - {"continue": True, "additionalContext": f"[{INSTALLED_HOOK}] {FIXTURE_MESSAGE}"} - ).encode(), - ) + self.assertEqual(dispatched.stdout, alone.stdout) self.assertEqual(dispatched.stderr, expected_stderr) alone_row = self._only_row(alone_rows) @@ -480,7 +558,7 @@ def test_rewrapping_an_already_collapsed_cursor_install_keeps_its_hooks(self): env=env, ) self.assertEqual(result.returncode, 0, result.stderr) - self.assertIn(f"[{INSTALLED_HOOK}] {FIXTURE_MESSAGE}".encode(), result.stdout) + self.assertIn(FIXTURE_MESSAGE.encode(), result.stdout) def test_an_unreadable_cursor_registry_is_reported_not_read_as_no_hooks(self): home = self._install("cursor", "speak", "1") diff --git a/scripts/install/mirror_stop_hooks_to_subagent_stop.py b/scripts/install/mirror_stop_hooks_to_subagent_stop.py index 132a53927..4ffc0c142 100644 --- a/scripts/install/mirror_stop_hooks_to_subagent_stop.py +++ b/scripts/install/mirror_stop_hooks_to_subagent_stop.py @@ -37,7 +37,9 @@ SOURCE_EVENT = "Stop" TARGET_EVENT = "SubagentStop" OPT_OUT_KEY = "subagent_stop" +DISPATCH_KEY = "dispatch" HOOKS_PREFIX_LITERAL = "$HOME/.claude/hooks/" +DISPATCHER_TIMEOUT = 9.5 sys.path.insert(0, SDK_DIR) from registry import load_registry @@ -49,6 +51,7 @@ class StopManifest: path: str entries: list = field(default_factory=list) inherit: bool = True + dispatch: bool = False reason: str = "" @property @@ -90,6 +93,8 @@ def load_manifests( raise ValueError(f"{path}: {OPT_OUT_KEY}.inherit is false but no reason is given") record.inherit = False record.reason = reason + elif isinstance(opt_out, dict) and bool(opt_out.get(DISPATCH_KEY)): + record.dispatch = True found.append(record) return found @@ -102,6 +107,11 @@ def _mirrored_entry(entry: dict) -> dict: return copy +def _dispatcher_entry(timeout: float = DISPATCHER_TIMEOUT) -> dict: + command = f"python3 $HOME/.claude/hooks/_runner/dispatch.py --event {TARGET_EVENT} --timeout {timeout:g}" + return {"hooks": [{"type": "command", "command": command, "timeout": timeout + 0.5}]} + + def _entry_is_for(entry: dict, prefixes: list[str]) -> bool: for hook in entry.get("hooks", []) if isinstance(entry, dict) else []: command = str(hook.get("command", "")) if isinstance(hook, dict) else "" @@ -123,9 +133,11 @@ def mirror(settings: dict, manifests: list[StopManifest]) -> tuple[dict, bool]: settings = json.loads(json.dumps(settings)) entry_list = settings.setdefault("hooks", {}).setdefault(TARGET_EVENT, []) before = json.dumps(entry_list, sort_keys=True) - managed = [m.command_prefix for m in manifests] + managed = [m.command_prefix for m in manifests] + [f"{HOOKS_PREFIX_LITERAL}_runner/dispatch.py"] kept = [e for e in entry_list if not _entry_is_for(e, managed)] - added = [_mirrored_entry(e) for m in manifests if m.inherit for e in m.entries] + direct = [_mirrored_entry(e) for m in manifests if m.inherit and not m.dispatch for e in m.entries] + dispatcher = [_dispatcher_entry()] if any(m.inherit and m.dispatch for m in manifests) else [] + added = direct + dispatcher entry_list[:] = kept + added if not entry_list: del settings["hooks"][TARGET_EVENT] @@ -145,7 +157,8 @@ def main() -> int: new_settings, changed = mirror(settings, manifests) for manifest in manifests: if manifest.inherit: - print(f"ok claude {TARGET_EVENT} {manifest.name} (mirrored from {SOURCE_EVENT})") + mode = "dispatcher" if manifest.dispatch else "mirrored" + print(f"ok claude {TARGET_EVENT} {manifest.name} ({mode} from {SOURCE_EVENT})") else: print(f"skip claude {TARGET_EVENT} {manifest.name} (opt-out: {manifest.reason})") if not changed: diff --git a/tests/test_install.py b/tests/test_install.py index e065f2d94..154a73c73 100644 --- a/tests/test_install.py +++ b/tests/test_install.py @@ -1799,6 +1799,13 @@ def wrapped_claude_command_re(hook): ) +def subagent_dispatcher_command_re(): + return ( + r"(?:python3|/\S+) " + + re.escape("$HOME/.claude/hooks/_runner/dispatch.py --event SubagentStop --timeout 9.5") + ) + + class TestSubagentStopInheritance(unittest.TestCase): """Every hook that wires Stop is also wired as SubagentStop after install.sh, unless its manifest opts out with a reason. Parametrized @@ -1832,7 +1839,13 @@ def test_every_stop_hook_is_mirrored_or_opted_out_with_reason(self): with self.subTest(hook=manifest.name, command=hook["command"]): expected = wrapped_claude_command_re(hook) self.assertTrue(any(re.fullmatch(expected, command) for command in stop), stop) - if manifest.inherit: + if manifest.inherit and manifest.dispatch: + self.assertTrue( + any(re.fullmatch(subagent_dispatcher_command_re(), command) for command in subagent_stop), + subagent_stop, + ) + self.assertFalse(any(re.fullmatch(expected, command) for command in subagent_stop), subagent_stop) + elif manifest.inherit: self.assertTrue(any(re.fullmatch(expected, command) for command in subagent_stop), subagent_stop) else: self.assertTrue(manifest.reason) diff --git a/tests/test_mirror_subagent_stop.py b/tests/test_mirror_subagent_stop.py index 0ccd73241..93f7f5e4c 100644 --- a/tests/test_mirror_subagent_stop.py +++ b/tests/test_mirror_subagent_stop.py @@ -62,6 +62,15 @@ def test_opt_out_with_reason_is_recorded(self): self.assertFalse(manifest.inherit) self.assertEqual(manifest.reason, "talks to the human only") + def test_dispatch_opt_in_is_recorded(self): + write_manifest(self.hooks_dir, "fast", { + "hooks": {"Stop": [stop_entry("fast")]}, + "subagent_stop": {"dispatch": True}, + }) + [manifest] = mod.load_manifests(self.hooks_dir, active_hooks={"fast"}) + self.assertTrue(manifest.inherit) + self.assertTrue(manifest.dispatch) + def test_opt_out_without_reason_is_rejected(self): write_manifest(self.hooks_dir, "silent", { "hooks": {"Stop": [stop_entry("silent")]}, @@ -85,17 +94,17 @@ def test_real_repo_opted_out_set_is_the_documented_one(self): class TestMirror(unittest.TestCase): def manifests(self): return [ - mod.StopManifest(name="a", path="a", entries=[stop_entry("a")]), + mod.StopManifest(name="a", path="a", entries=[stop_entry("a")], dispatch=True), mod.StopManifest(name="b", path="b", entries=[stop_entry("b", "claude_stop_b.py")]), mod.StopManifest(name="quiet", path="quiet", entries=[stop_entry("quiet")], inherit=False, reason="human only"), ] - def test_every_inheriting_stop_hook_gets_a_subagent_stop_entry_without_matcher(self): + def test_dispatch_opted_hooks_share_one_dispatcher_entry_and_others_stay_direct(self): out, changed = mod.mirror({"hooks": {}}, self.manifests()) self.assertTrue(changed) self.assertEqual(commands(out, "SubagentStop"), [ - "python3 $HOME/.claude/hooks/a/claude_stop_check.py", "python3 $HOME/.claude/hooks/b/claude_stop_b.py", + "python3 $HOME/.claude/hooks/_runner/dispatch.py --event SubagentStop --timeout 9.5", ]) for entry in out["hooks"]["SubagentStop"]: self.assertNotIn("matcher", entry) @@ -120,6 +129,13 @@ def test_rerun_is_idempotent(self): self.assertFalse(changed) self.assertEqual(first, second) + def test_stale_dispatcher_entry_is_removed_when_no_manifest_opts_in(self): + stale = {"hooks": {"SubagentStop": [mod._dispatcher_entry()]}} + direct = [mod.StopManifest(name="b", path="b", entries=[stop_entry("b")])] + out, changed = mod.mirror(stale, direct) + self.assertTrue(changed) + self.assertEqual(commands(out, "SubagentStop"), ["python3 $HOME/.claude/hooks/b/claude_stop_check.py"]) + def test_empty_result_does_not_leave_an_empty_event_key(self): out, changed = mod.mirror({"hooks": {}}, []) self.assertFalse(changed) From 6a3c5a1c2628248a908f6e0a62e85f3befab1808 Mon Sep 17 00:00:00 2001 From: Invoker Bot Date: Sun, 11 Oct 2026 15:37:29 +0000 Subject: [PATCH 04/10] =?UTF-8?q?invoker:=20wf-1791728212829-14/publish-pi?= =?UTF-8?q?lot=20=E2=80=94=20Review=20claim:=20SubagentStop=20hooks=20disp?= =?UTF-8?q?atch=20through=20one=20process=20that=20parses=20the=20event=20?= =?UTF-8?q?once=20and=20shares=20one=20transcript=20parse=20across=20all?= =?UTF-8?q?=20opted-in=20detectors,=20emitting=20the=20same=20per-detector?= =?UTF-8?q?=20metrics=20rows=20and=20verdicts=20as=20the=20subprocess=20fl?= =?UTF-8?q?eet.=20Review=20lane:=20behavior=20Safety=20invariant:=20Hook?= =?UTF-8?q?=20stdout=20and=20exit=20codes=20stay=20the=20same.=20This=20sl?= =?UTF-8?q?ice=20does=20not=20change=20the=20other=20four=20claims.=20File?= =?UTF-8?q?s:=20engine/hooks/=5Frunner/dispatch.py=20Change=20types:=20bug?= =?UTF-8?q?fix=20Acceptance=20criteria:=20The=20branch=20contains=20only?= =?UTF-8?q?=20this=20claim,=20and=20the=20review=20step=20publishes=20one?= =?UTF-8?q?=20pull=20request.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Exit code: 0 Invoker-Finalize-Id: aaa80940-f2d5-4d79-bb40-79c68fe333ad From 6d1920236ed00d57ab7f724acf9940aa71725e50 Mon Sep 17 00:00:00 2001 From: Invoker Bot Date: Thu, 1 Oct 2026 12:00:21 +0000 Subject: [PATCH 05/10] Record a dispatched detector that raises as crashed, not caught_error. Subprocess isolation was the hook fleet's free safety property, so the dispatcher has to prove it reproduced that property rather than assume it. _run_sdk_one caught a raising detector but left exit_code at 0, so outcome.classify saw a zero exit with a catstack-hook-error stderr line and recorded caught_error instead of crashed; the except branch now sets exit code 1. Four chaos fixtures under engine/hooks/_runner/tests/chaos (raise, sleep past the soft budget, exit 3, and a sibling that speaks) drive a dispatcher test asserting each failure gets its own row with the right outcome and exit code while both sibling detectors still emit their verdicts and rows. The fixtures are test-only: they live under tests/ and are registered in a scratch manifest, so the mirror script's name allowlist can never install them into a live settings.json. (cherry picked from commit b57f27c3 on plan/hook-fleet-dispatcher-measure-the-herd-then-dispatch-it) --- engine/hooks/_runner/dispatch.py | 2 +- engine/hooks/_runner/tests/chaos/exit_hook.py | 7 +++ engine/hooks/_runner/tests/chaos/hang_hook.py | 8 +++ engine/hooks/_runner/tests/chaos/ok_hook.py | 15 ++++++ .../hooks/_runner/tests/chaos/raise_hook.py | 5 ++ engine/hooks/_runner/tests/test_dispatch.py | 50 +++++++++++++++++++ 6 files changed, 86 insertions(+), 1 deletion(-) create mode 100644 engine/hooks/_runner/tests/chaos/exit_hook.py create mode 100644 engine/hooks/_runner/tests/chaos/hang_hook.py create mode 100644 engine/hooks/_runner/tests/chaos/ok_hook.py create mode 100644 engine/hooks/_runner/tests/chaos/raise_hook.py diff --git a/engine/hooks/_runner/dispatch.py b/engine/hooks/_runner/dispatch.py index 35743450d..090acbb09 100644 --- a/engine/hooks/_runner/dispatch.py +++ b/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: diff --git a/engine/hooks/_runner/tests/chaos/exit_hook.py b/engine/hooks/_runner/tests/chaos/exit_hook.py new file mode 100644 index 000000000..709aa8065 --- /dev/null +++ b/engine/hooks/_runner/tests/chaos/exit_hook.py @@ -0,0 +1,7 @@ +from __future__ import annotations + +import sys + + +def detect(event): + raise SystemExit(3) diff --git a/engine/hooks/_runner/tests/chaos/hang_hook.py b/engine/hooks/_runner/tests/chaos/hang_hook.py new file mode 100644 index 000000000..8643e7979 --- /dev/null +++ b/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 [] diff --git a/engine/hooks/_runner/tests/chaos/ok_hook.py b/engine/hooks/_runner/tests/chaos/ok_hook.py new file mode 100644 index 000000000..c56eb8665 --- /dev/null +++ b/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", + ) + ] diff --git a/engine/hooks/_runner/tests/chaos/raise_hook.py b/engine/hooks/_runner/tests/chaos/raise_hook.py new file mode 100644 index 000000000..39a4c6d0c --- /dev/null +++ b/engine/hooks/_runner/tests/chaos/raise_hook.py @@ -0,0 +1,5 @@ +from __future__ import annotations + + +def detect(event): + raise RuntimeError("chaos fixture boom") diff --git a/engine/hooks/_runner/tests/test_dispatch.py b/engine/hooks/_runner/tests/test_dispatch.py index 20dc6e02c..67afad755 100644 --- a/engine/hooks/_runner/tests/test_dispatch.py +++ b/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,42 @@ 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" From d4a7a00253d93b0b03170fb49703bb85416748c0 Mon Sep 17 00:00:00 2001 From: Invoker Bot Date: Sun, 11 Oct 2026 16:22:36 +0000 Subject: [PATCH 06/10] =?UTF-8?q?invoker:=20wf-1791728213123-15/publish-ch?= =?UTF-8?q?aos=20=E2=80=94=20Review=20claim:=20Fixture=20detectors=20that?= =?UTF-8?q?=20raise,=20hang=20past=20the=20soft=20budget,=20and=20exit=20n?= =?UTF-8?q?on-zero=20each=20produce=20their=20own=20failure=20row=20while?= =?UTF-8?q?=20every=20sibling=20detector's=20verdict=20and=20row=20is=20st?= =?UTF-8?q?ill=20emitted.=20Review=20lane:=20behavior=20Safety=20invariant?= =?UTF-8?q?:=20Hook=20stdout=20and=20exit=20codes=20stay=20the=20same.=20T?= =?UTF-8?q?his=20slice=20does=20not=20change=20the=20other=20four=20claims?= =?UTF-8?q?.=20Files:=20engine/hooks/=5Frunner/tests/chaos=20Change=20type?= =?UTF-8?q?s:=20bugfix=20Acceptance=20criteria:=20The=20branch=20contains?= =?UTF-8?q?=20only=20this=20claim,=20and=20the=20review=20step=20publishes?= =?UTF-8?q?=20one=20pull=20request.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Exit code: 0 Invoker-Finalize-Id: 7ca2543e-d062-46fa-bbce-58264f6ae48f From d63b1fc922d925dc6f89685398b42e2b02139371 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Sun, 11 Oct 2026 19:22:22 +0000 Subject: [PATCH 07/10] Fix replay hook payload lint --- scripts/test/test_replay_hook_payloads.py | 1 - 1 file changed, 1 deletion(-) diff --git a/scripts/test/test_replay_hook_payloads.py b/scripts/test/test_replay_hook_payloads.py index 263fce353..a8925a5ec 100644 --- a/scripts/test/test_replay_hook_payloads.py +++ b/scripts/test/test_replay_hook_payloads.py @@ -2,7 +2,6 @@ import hashlib import json -import os from pathlib import Path import shlex import subprocess From 9787af8f07826c77ba0072030a241cf253e6a08c Mon Sep 17 00:00:00 2001 From: CI Bot Date: Sun, 11 Oct 2026 19:22:27 +0000 Subject: [PATCH 08/10] =?UTF-8?q?invoker:=20wf-1791746491297-4/repair=20?= =?UTF-8?q?=E2=80=94=20Repair=20PR=20#1333=20(failed=20check=20lint)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Exit code: 0 From 6bbfc6e7d6e1d42389124c13ac72657ccf6763a2 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Sun, 11 Oct 2026 19:22:54 +0000 Subject: [PATCH 09/10] Fix lint in chaos exit fixture --- engine/hooks/_runner/tests/chaos/exit_hook.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/engine/hooks/_runner/tests/chaos/exit_hook.py b/engine/hooks/_runner/tests/chaos/exit_hook.py index 709aa8065..d1b944041 100644 --- a/engine/hooks/_runner/tests/chaos/exit_hook.py +++ b/engine/hooks/_runner/tests/chaos/exit_hook.py @@ -1,7 +1,4 @@ from __future__ import annotations -import sys - - def detect(event): raise SystemExit(3) From 1c103820c305ab276a10c015d741a2926c8569cd Mon Sep 17 00:00:00 2001 From: CI Bot Date: Sun, 11 Oct 2026 19:23:06 +0000 Subject: [PATCH 10/10] =?UTF-8?q?invoker:=20wf-1791746497228-5/repair=20?= =?UTF-8?q?=E2=80=94=20Repair=20PR=20#1336=20(failed=20check=20lint)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Exit code: 0