diff --git a/corpus/skills/cat-mode/SKILL.md b/corpus/skills/cat-mode/SKILL.md index 4e0ffcc4..3464274d 100644 --- a/corpus/skills/cat-mode/SKILL.md +++ b/corpus/skills/cat-mode/SKILL.md @@ -178,14 +178,17 @@ What happens to a number once it exists: - **Retractions cover the conversation, not just the artifacts.** - **An admission lists every live instance of the mistake**, including work the agent launched itself. - **A claim about the repo's own history is a query, not a recollection.** -- **A blank or synthetic model on an analytics event is an emit bug.** Fix it at write or backfill — never hide it with a chart filter. [[principle-assert-invariants-not-last-bug]], [[principle-explicit-errors]]. -- **"Full backfill" means every emitter of the metric**, not the first harness that was easy. Same class as Harness-agnostic product defaults above. +- **A blank or synthetic model on an analytics event is an emit or resolve bug.** Fix it at write, resolve, or backfill — never hide it with a chart filter, and never treat a blank-include chart change as the fix when emit or resolve owns the gap. Prove the live resolve path for that harness first. [[principle-assert-invariants-not-last-bug]], [[principle-explicit-errors]]. +- **A multi-emitter backfill is done only with a harness×source matrix plus fill table** (for example when the ask is "full backfill" or "all harnesses"), not the first harness that was easy. Same class as Harness-agnostic product defaults above. +- **Never narrate by-model or "real model" dashboard health without a health receipt first:** harness×action table of event count + `filled_model_rate`, with volume vs fill vs resolve vs filter ownership named. +- **Split volume vs fill vs resolve vs filter before any L3 or UI insight PATCH when a harness looks missing or a real-model tile reads zero** (for example "missing harness X" or "real model = 0"). - **A published metrics view is not current until a live event shows the new identity fields.** Prove the fields landed (install, then probe) before trusting the dashboard. Extends "the report of a write is not the write's effect" and [[principle-prove-it]]. - **Keep a locked product question narrow** (for example one chart per model); do not expand visualization scope past it without an explicit new ask. Follow `narrow-the-scope`. - **Climb analytics tiers in order:** L1 explain → L2 one canary query → L3 backfill/fleet only after the canary proves an identity or cardinality gap. - **Schema before UI.** Required identity properties plus a green canary before the first insight or notebook PATCH; split the forward-stamp PR from the historical-backfill PR. - **Claim a chart gap fixed only after naming write vs enrich ownership** (local vs peer vs fleet). - **Write a thrash checkpoint before attaching new scope:** after two same-type interventions or the first live backfill, record tier and canary status first. +- **When a same-type miss forces a working-style catch mid analytics thrash, stop feature PRs until that catch is locked** (for example the first `intervention-must-automate` flag, an explicit `/automate-me`, or a same-type restatement). Each rule's full text: [references/verify.md](references/verify.md). diff --git a/corpus/skills/cat-mode/references/verify.md b/corpus/skills/cat-mode/references/verify.md index 85ee7b0e..0438f539 100644 --- a/corpus/skills/cat-mode/references/verify.md +++ b/corpus/skills/cat-mode/references/verify.md @@ -149,18 +149,37 @@ Standing defaults when publishing hook/skill analytics — identity fields, backfills, dashboards, and charts. Prefer fixing emit and coverage over chart workarounds. -- **A blank or synthetic model on an analytics event is an emit bug.** Treat - a missing or placeholder model as a write/backfill defect to fix at the - source. Do not work around it by filtering the blank series out of a - chart, swapping the breakdown to another dimension, or calling the blank - "noise." The model field is an invariant of the event, not a display - preference. [[principle-assert-invariants-not-last-bug]], +- **A blank or synthetic model on an analytics event is an emit or resolve + bug.** Treat a missing or placeholder model as a write, resolve, or + backfill defect to fix at the source. Do not work around it by filtering + the blank series out of a chart, swapping the breakdown to another + dimension, or calling the blank "noise." Do not treat a blank-include + chart change as the fix when emit or resolve owns the gap — prove the + live resolve path for that harness first (same-session peers from local + metrics when the payload omits the model). The model field is an + invariant of the event, not a display preference. + [[principle-assert-invariants-not-last-bug]], [[principle-explicit-errors]]. -- **"Full backfill" means every emitter of the metric.** A backfill that - covers only the first harness that was easy is incomplete. Include every - harness (or other emitter) that writes the metric before calling the - backfill done. Same class as Harness-agnostic product defaults in +- **A multi-emitter backfill is done only with a harness×source matrix + plus fill table** (for example when the ask is "full backfill" or + "all harnesses"). A backfill that covers only the first harness that + was easy is incomplete. Before claiming done, show every harness (or + other emitter) that writes the metric, the source used for each, and + the fill rate that resulted. Scoping to one harness without that matrix + is incomplete. Same class as Harness-agnostic product defaults in SKILL.md. +- **Never narrate by-model or "real model" dashboard health without a + health receipt first.** Before saying which models look healthy or + missing, paste a harness×action table of event count and + `filled_model_rate`, and name volume vs fill vs resolve vs filter + ownership for each gap. A chart anecdote is not that receipt. + [[principle-prove-it]], [[principle-explicit-errors]]. +- **Split volume vs fill vs resolve vs filter before any L3 or UI insight + PATCH when a harness looks missing or a real-model tile reads zero** + (for example "missing harness X" or "real model = 0"). Low volume, blank + fill, broken resolve, and a chart filter are different owners. Name + which one owns the gap, then climb tiers; do not open backfill, fleet, + or insight/notebook PATCH work until that split is stated. - **A published metrics view is not current until a live event shows the new identity fields.** Shipping a chart or notebook update is not proof the new fields land. Prove they landed by installing the emitting path, @@ -195,6 +214,11 @@ defaults above and [[principle-push-not-poll]]; they do not replace them. of a run, write the current tier and canary status before attaching new stack publication or plan scope. A checkpoint is a receipt, not a status ping. +- **When a same-type miss forces a working-style catch mid analytics thrash, stop feature PRs until that catch is locked** (for example the + first `intervention-must-automate` flag, an explicit `/automate-me`, or + a same-type restatement). The standing default that would have prevented + the miss lands first; dashboard, backfill, and insight work wait until + that catch is encoded or already open as its own review unit. - **Million-row or multi-host backfills are owned by a durable runner, not a chat poll loop.** Peer or fleet dry-run or live work that can exceed about minutes must run under Invoker or a background runner that leaves a diff --git a/corpus/skills/cat-mode/tests/fires_automate_mid_analytics_without_locking_catch.md b/corpus/skills/cat-mode/tests/fires_automate_mid_analytics_without_locking_catch.md new file mode 100644 index 00000000..640eb514 --- /dev/null +++ b/corpus/skills/cat-mode/tests/fires_automate_mid_analytics_without_locking_catch.md @@ -0,0 +1,2 @@ +User: "/automate-me we went through a lot of cycles — and also keep iterating the dashboard tiles." +Assistant: Keep shipping analytics insight PATCHes and another backfill PR in the same thrash; defer locking the class catch until after more feature work. diff --git a/corpus/skills/cat-mode/tests/fires_blank_include_chart_workaround.md b/corpus/skills/cat-mode/tests/fires_blank_include_chart_workaround.md new file mode 100644 index 00000000..6a6148b9 --- /dev/null +++ b/corpus/skills/cat-mode/tests/fires_blank_include_chart_workaround.md @@ -0,0 +1,2 @@ +User: "Why is cursor w real model = 0 on skill tiles?" +Assistant: Include blank models in the chart filters so Cursor rows show up again — that restores the viz without proving live resolve or session peers from local metrics. diff --git a/corpus/skills/cat-mode/tests/fires_blank_model_chart_filter_workaround.md b/corpus/skills/cat-mode/tests/fires_blank_model_chart_filter_workaround.md index 5a8447b2..e30f4622 100644 --- a/corpus/skills/cat-mode/tests/fires_blank_model_chart_filter_workaround.md +++ b/corpus/skills/cat-mode/tests/fires_blank_model_chart_filter_workaround.md @@ -1,2 +1,2 @@ User: "Why do so many rows show blank model on the hooks chart?" -Assistant: Filter out blank and synthetic models so the chart only shows named ones — that cleans the viz without touching emit or backfill. +Assistant: Filter out blank and synthetic models so the chart only shows named ones — that cleans the viz without touching emit, resolve, or backfill. diff --git a/corpus/skills/cat-mode/tests/fires_dashboard_health_without_receipt.md b/corpus/skills/cat-mode/tests/fires_dashboard_health_without_receipt.md new file mode 100644 index 00000000..7fececd7 --- /dev/null +++ b/corpus/skills/cat-mode/tests/fires_dashboard_health_without_receipt.md @@ -0,0 +1,2 @@ +User: "Is the by-model dashboard healthy? Cursor real model looks odd." +Assistant: Cursor looks fine on the chart and Claude dominates, so health is good — no harness×action count or filled_model_rate table. diff --git a/corpus/skills/cat-mode/tests/fires_full_backfill_without_matrix.md b/corpus/skills/cat-mode/tests/fires_full_backfill_without_matrix.md new file mode 100644 index 00000000..4495f3bd --- /dev/null +++ b/corpus/skills/cat-mode/tests/fires_full_backfill_without_matrix.md @@ -0,0 +1,2 @@ +User: "Full backfill all harnesses, then we're done." +Assistant: Backfill Codex only (first easy emitter) and call full backfill done — no harness×source matrix or fill table. diff --git a/corpus/skills/cat-mode/tests/fires_missing_harness_ui_patch_before_split.md b/corpus/skills/cat-mode/tests/fires_missing_harness_ui_patch_before_split.md new file mode 100644 index 00000000..786b55e9 --- /dev/null +++ b/corpus/skills/cat-mode/tests/fires_missing_harness_ui_patch_before_split.md @@ -0,0 +1,2 @@ +User: "Why are we missing so many Codex? Also why is Cursor real model = 0?" +Assistant: Patch the insight/notebook filters and open an L3 backfill so the missing harnesses reappear — skip splitting volume vs fill vs resolve vs filter first. diff --git a/corpus/skills/cat-mode/tests/stays_silent_health_receipt_then_diagnose.md b/corpus/skills/cat-mode/tests/stays_silent_health_receipt_then_diagnose.md new file mode 100644 index 00000000..0447b1db --- /dev/null +++ b/corpus/skills/cat-mode/tests/stays_silent_health_receipt_then_diagnose.md @@ -0,0 +1,2 @@ +User: "Why are we missing so many Codex? Also why is Cursor real model = 0?" +Assistant: Paste a harness×action health receipt (event count + filled_model_rate), split volume vs fill vs resolve vs filter, prove the live resolve path with session peers from local metrics, and lock the class catch before more analytics PRs. diff --git a/engine/hooks/_runner/README.md b/engine/hooks/_runner/README.md index 1b5768d4..f410ae92 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 42fa17fc..5aba5e61 100644 --- a/engine/hooks/_runner/run.py +++ b/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 diff --git a/engine/hooks/_runner/tests/test_run.py b/engine/hooks/_runner/tests/test_run.py index 563b9551..07fc1e56 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/engine/hooks/_runner/tests/test_wrap_installed.py b/engine/hooks/_runner/tests/test_wrap_installed.py index 008b06bd..6de0ec8a 100644 --- a/engine/hooks/_runner/tests/test_wrap_installed.py +++ b/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" diff --git a/engine/hooks/prove-it-ship-gate/README.md b/engine/hooks/prove-it-ship-gate/README.md index 5a980ab2..7f9c88f9 100644 --- a/engine/hooks/prove-it-ship-gate/README.md +++ b/engine/hooks/prove-it-ship-gate/README.md @@ -50,9 +50,13 @@ UI only, and the user had to force a live Linear e2e. ## Same command -A product command (a hyphenated program such as `invoker-cli doctor`) that +The stop is wired for Claude, Cursor, and Codex. Each harness gets that +harness's own block: Claude exits 2, Cursor returns a followup message on +its stop event, and Codex returns a block decision. A product command (a +hyphenated program such as `invoker-cli doctor`) that already ran in the session, including a subagent transcript beside it, has -to run again after the last file edit. A later `vitest` / `pnpm test` / +to run again after the last file edit. Codex `exec_command` rows count as +that command, and `apply_patch` counts as the edit. A later `vitest` / `pnpm test` / `pytest` run does not replace it. The hook names the missing program and subcommand and blocks. Unix pipelines, `node -e` scripts, and a session that never ran a product command stay silent. A quoted program-and-subcommand pair inside a script counts as that command. diff --git a/engine/hooks/prove-it-ship-gate/codex.hook.json b/engine/hooks/prove-it-ship-gate/codex.hook.json new file mode 100644 index 00000000..d1559a9a --- /dev/null +++ b/engine/hooks/prove-it-ship-gate/codex.hook.json @@ -0,0 +1,16 @@ +{ + "hooks": { + "Stop": [ + { + "matcher": ".*", + "hooks": [ + { + "type": "command", + "command": "python3 $HOME/.codex/hooks/prove-it-ship-gate/codex_stop_check.py", + "timeout": 10 + } + ] + } + ] + } +} diff --git a/engine/hooks/prove-it-ship-gate/codex_stop_check.py b/engine/hooks/prove-it-ship-gate/codex_stop_check.py new file mode 100644 index 00000000..74c7ee70 --- /dev/null +++ b/engine/hooks/prove-it-ship-gate/codex_stop_check.py @@ -0,0 +1,20 @@ +#!/usr/bin/env python3 +"""Codex Stop hook entrypoint for prove-it-ship-gate.""" +from __future__ import annotations + +import os +import sys + +sys.path.insert(0, os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_sdk")) + +from detect import detect # noqa: E402 +from runtime import run_hook # noqa: E402 + + +def main() -> None: + run_hook("prove-it-ship-gate", "codex", detect, "Stop", json_error_stderr=False) + + +if __name__ == "__main__": + main() diff --git a/engine/hooks/prove-it-ship-gate/cursor_stop_check.py b/engine/hooks/prove-it-ship-gate/cursor_stop_check.py index a2f36b57..fadf5700 100644 --- a/engine/hooks/prove-it-ship-gate/cursor_stop_check.py +++ b/engine/hooks/prove-it-ship-gate/cursor_stop_check.py @@ -13,7 +13,7 @@ def main() -> None: - run_hook("prove-it-ship-gate", "cursor", detect) + run_hook("prove-it-ship-gate", "cursor", detect, "stop") if __name__ == "__main__": diff --git a/engine/hooks/prove-it-ship-gate/detect.py b/engine/hooks/prove-it-ship-gate/detect.py index eec0fdde..9552cbfc 100644 --- a/engine/hooks/prove-it-ship-gate/detect.py +++ b/engine/hooks/prove-it-ship-gate/detect.py @@ -46,8 +46,14 @@ RULE_UNPROVEN_LIVE_CLAIM = "prove-it-ship-gate.unproven-live-claim" RULE_SAME_REPRO = "prove-it-ship-gate.same-repro-not-rerun" -SHELL_TOOL_NAMES = frozenset({"Bash", "Shell", "bash", "shell"}) -EDIT_TOOL_NAMES = frozenset({"Write", "StrReplace", "Edit", "EditNotebook", "ApplyPatch"}) +SHELL_TOOL_NAMES = frozenset({ + "Bash", "bash", "Shell", "shell", "Exec", "exec", "exec_command", + "functions.exec_command", "local_shell", "shell_call", +}) +EDIT_TOOL_NAMES = frozenset({ + "Write", "StrReplace", "Edit", "EditNotebook", "ApplyPatch", "apply_patch", + "MultiEdit", "NotebookEdit", "edit_file", +}) COMMAND_PREFIXES = frozenset({"sudo", "env", "time", "command", "nohup", "nice"}) GENERIC_PROGRAMS = frozenset({ "git", "node", "python", "python3", "bash", "zsh", "sh", "ls", "cat", "rg", @@ -211,24 +217,48 @@ def bash_commands_this_turn(transcript_path: str) -> list[str] | None: continue for block in _tool_uses(data): if block.get("name") in SHELL_TOOL_NAMES: - command = (block.get("input") or {}).get("command") - if isinstance(command, str): + command = _shell_command(block) + if command is not None: commands.append(command) return commands +def _shell_command(block: dict) -> str | None: + raw = block.get("input") + if not isinstance(raw, dict): + return None + for key in ("command", "cmd"): + value = raw.get(key) + if isinstance(value, str): + return value + return None + + def _tool_uses(data: dict) -> list[dict]: + found: list[dict] = [] message = data.get("message") if isinstance(data.get("message"), dict) else {} role = data.get("role") or message.get("role") or data.get("type") - if role != "assistant": - return [] - content = message.get("content") if message else data.get("content") - if not isinstance(content, list): - return [] - return [ - block for block in content - if isinstance(block, dict) and block.get("type") == "tool_use" - ] + if role == "assistant": + content = message.get("content") if message else data.get("content") + if isinstance(content, list): + found.extend( + block for block in content + if isinstance(block, dict) and block.get("type") == "tool_use" + ) + payload = data.get("payload") if isinstance(data.get("payload"), dict) else data + if isinstance(payload, dict) and payload.get("type") in {"function_call", "custom_tool_call"}: + name = payload.get("name") + raw = payload.get("arguments") + if raw is None: + raw = payload.get("input") + if isinstance(raw, str): + try: + raw = json.loads(raw) + except json.JSONDecodeError: + raw = {} + if isinstance(name, str) and isinstance(raw, dict): + found.append({"type": "tool_use", "name": name, "input": raw}) + return found def _unquoted_segments(command: str) -> list[str]: @@ -323,8 +353,8 @@ def _read_events(path: str) -> list[tuple[str, str]] | None: if name in EDIT_TOOL_NAMES: events.append(("edit", "")) elif name in SHELL_TOOL_NAMES: - command = (block.get("input") or {}).get("command") - if isinstance(command, str): + command = _shell_command(block) + if command is not None: events.append(("cmd", command)) return events diff --git a/engine/hooks/prove-it-ship-gate/tests/test_hooks.py b/engine/hooks/prove-it-ship-gate/tests/test_hooks.py index 8beee6f3..a6faa139 100644 --- a/engine/hooks/prove-it-ship-gate/tests/test_hooks.py +++ b/engine/hooks/prove-it-ship-gate/tests/test_hooks.py @@ -14,6 +14,7 @@ import io import json import os +import subprocess import sys import tempfile import unittest @@ -606,3 +607,91 @@ def test_fails_open_when_the_same_repro_transcript_is_missing(self): "last_assistant_message": "vitest passed", "transcript_path": "/nonexistent/same-repro.jsonl", })) + + def test_fires_on_a_codex_exec_and_patch(self): + with tempfile.TemporaryDirectory() as directory: + path = os.path.join(directory, "session.jsonl") + rows = [ + { + "type": "function_call", + "name": "exec_command", + "arguments": json.dumps({"cmd": "invoker-cli doctor"}), + }, + { + "type": "function_call", + "name": "apply_patch", + "arguments": json.dumps({"command": ["apply_patch", "*** Begin Patch"]}), + }, + { + "type": "function_call", + "name": "exec_command", + "arguments": json.dumps({"cmd": "pnpm test"}), + }, + ] + with open(path, "w", encoding="utf-8") as handle: + handle.write("\n".join(json.dumps(row) for row in rows) + "\n") + feedback = detect.decide({ + "last_assistant_message": "the test passed", + "transcript_path": path, + }) + self.assertIsNotNone(feedback) + self.assertIn("invoker-cli doctor", feedback) + + def test_the_same_finding_blocks_claude_cursor_and_codex(self): + hook_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + with tempfile.TemporaryDirectory() as directory: + path = os.path.join(directory, "session.jsonl") + rows = [ + { + "type": "function_call", + "name": "exec_command", + "arguments": json.dumps({"cmd": "invoker-cli doctor"}), + }, + { + "type": "function_call", + "name": "apply_patch", + "arguments": "{}", + }, + { + "type": "function_call", + "name": "exec_command", + "arguments": json.dumps({"cmd": "pnpm test"}), + }, + ] + with open(path, "w", encoding="utf-8") as handle: + handle.write("\n".join(json.dumps(row) for row in rows) + "\n") + env = os.environ.copy() + env["CATSTACK_HOOK_TEST_MODEL"] = "gpt-5.6-sol" + env["CATSTACK_HOOK_METRICS_DIR"] = os.path.join(directory, "metrics") + cases = [ + ("claude_stop_check.py", {"hook_event_name": "Stop"}, "claude"), + ("cursor_stop_check.py", {"hook_event_name": "stop"}, "cursor"), + ("codex_stop_check.py", {"hook_event_name": "Stop"}, "codex"), + ] + for script, extra, harness in cases: + payload = { + "last_assistant_message": "the test passed", + "transcript_path": path, + **extra, + } + result = subprocess.run( + [sys.executable, os.path.join(hook_dir, script)], + input=json.dumps(payload), + capture_output=True, + text=True, + env=env, + check=False, + ) + with self.subTest(harness=harness): + if harness == "claude": + self.assertEqual(2, result.returncode, result.stderr) + self.assertIn("invoker-cli doctor", result.stderr) + elif harness == "cursor": + self.assertEqual(0, result.returncode, result.stderr) + body = json.loads(result.stdout) + self.assertIn("invoker-cli doctor", body["followup_message"]) + else: + self.assertEqual(0, result.returncode, result.stderr) + body = json.loads(result.stdout) + self.assertEqual("block", body["decision"]) + self.assertIn("invoker-cli doctor", body["reason"]) diff --git a/scripts/test/replay_hook_payloads.py b/scripts/test/replay_hook_payloads.py new file mode 100644 index 00000000..e52686ba --- /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 00000000..a8925a5e --- /dev/null +++ b/scripts/test/test_replay_hook_payloads.py @@ -0,0 +1,99 @@ +from __future__ import annotations + +import hashlib +import json +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() diff --git a/tests/test_cat_mode.py b/tests/test_cat_mode.py index 7577dbc0..c5c065d3 100644 --- a/tests/test_cat_mode.py +++ b/tests/test_cat_mode.py @@ -202,7 +202,8 @@ def test_verify_reference_carries_full_original_session_rule(self): class TestPublishingAnalyticsDefaults(unittest.TestCase): - """Verify: publishing analytics defaults for emit bugs, full backfill, + """Verify: publishing analytics defaults for emit/resolve bugs, full + backfill matrix, health receipts, volume/fill/resolve/filter splits, install+probe before trusting dashboards, and locked viz scope. """ @@ -216,12 +217,22 @@ class TestPublishingAnalyticsDefaults(unittest.TestCase): def test_skill_names_publishing_analytics_defaults(self): text = read_skill_text() self.assertIn( - "A blank or synthetic model on an analytics event is an emit bug", + "A blank or synthetic model on an analytics event is an emit or resolve bug", text, ) self.assertIn("never hide it with a chart filter", text) + self.assertIn("never treat a blank-include chart change as the fix", text) self.assertIn( - '"Full backfill" means every emitter of the metric', + "A multi-emitter backfill is done only with a harness×source matrix plus fill table", + text, + ) + self.assertIn( + "Never narrate by-model or \"real model\" dashboard health without a health receipt first", + text, + ) + self.assertIn("filled_model_rate", text) + self.assertIn( + "Split volume vs fill vs resolve vs filter before any L3 or UI insight PATCH when a harness looks missing or a real-model tile reads zero", text, ) self.assertIn( @@ -236,12 +247,29 @@ def test_verify_reference_carries_full_publishing_analytics_rules(self): text = handle.read() self.assertIn("## Publishing analytics", text) self.assertIn( - "A blank or synthetic model on an analytics event is an emit bug", + "A blank or synthetic model on an analytics event is an emit or resolve", text, ) self.assertIn("Do not work around it by filtering", text) + self.assertIn("blank-include", text) + self.assertIn("live resolve path", text) + self.assertIn( + "A multi-emitter backfill is done only with a harness×source matrix", + text, + ) + self.assertIn("fill table", text) + self.assertIn("full backfill", text) + self.assertIn("all harnesses", text) + self.assertIn( + "Never narrate by-model or \"real model\" dashboard health without a", + text, + ) + self.assertIn("filled_model_rate", text) + self.assertIn("volume vs fill vs resolve vs filter", text) + self.assertIn("missing harness X", text) + self.assertIn("real model = 0", text) self.assertIn( - '"Full backfill" means every emitter of the metric', + "Split volume vs fill vs resolve vs filter before any L3 or UI insight", text, ) self.assertIn( @@ -262,6 +290,42 @@ def test_fires_blank_model_chart_filter_workaround(self): self.assertIn("filter out", text.lower()) self.assertIn("blank", text.lower()) + def test_fires_blank_include_chart_workaround(self): + path = os.path.join( + self.FIXTURES_DIR, "fires_blank_include_chart_workaround.md" + ) + with open(path, encoding="utf-8") as handle: + text = handle.read().lower() + self.assertIn("include blank", text) + self.assertIn("resolve", text) + + def test_fires_dashboard_health_without_receipt(self): + path = os.path.join( + self.FIXTURES_DIR, "fires_dashboard_health_without_receipt.md" + ) + with open(path, encoding="utf-8") as handle: + text = handle.read().lower() + self.assertIn("health is good", text) + self.assertIn("filled_model_rate", text) + + def test_fires_missing_harness_ui_patch_before_split(self): + path = os.path.join( + self.FIXTURES_DIR, "fires_missing_harness_ui_patch_before_split.md" + ) + with open(path, encoding="utf-8") as handle: + text = handle.read().lower() + self.assertIn("insight", text) + self.assertIn("volume vs fill vs resolve vs filter", text) + + def test_fires_full_backfill_without_matrix(self): + path = os.path.join( + self.FIXTURES_DIR, "fires_full_backfill_without_matrix.md" + ) + with open(path, encoding="utf-8") as handle: + text = handle.read().lower() + self.assertIn("codex only", text) + self.assertIn("harness×source matrix", text) + def test_stays_silent_emit_fix_then_install_probe(self): path = os.path.join( self.FIXTURES_DIR, "stays_silent_emit_fix_then_install_probe.md" @@ -271,6 +335,16 @@ def test_stays_silent_emit_fix_then_install_probe(self): self.assertIn("backfill", text.lower()) self.assertIn("probe", text.lower()) + def test_stays_silent_health_receipt_then_diagnose(self): + path = os.path.join( + self.FIXTURES_DIR, "stays_silent_health_receipt_then_diagnose.md" + ) + with open(path, encoding="utf-8") as handle: + text = handle.read().lower() + self.assertIn("health receipt", text) + self.assertIn("filled_model_rate", text) + self.assertIn("volume vs fill vs resolve vs filter", text) + class TestAnalyticsThrashCheckpoints(unittest.TestCase): """Verify/autonomy: thrash checkpoints for analytics backfills — @@ -302,6 +376,10 @@ def test_skill_names_thrash_checkpoint_defaults(self): self.assertIn("Schema before UI", text) self.assertIn("Claim a chart gap fixed only after naming write vs enrich ownership", text) self.assertIn("Write a thrash checkpoint before attaching new scope", text) + self.assertIn( + "When a same-type miss forces a working-style catch mid analytics thrash, stop feature PRs until that catch is locked", + text, + ) self.assertIn("Do not race publication against a live long job in the same thread", text) self.assertIn( "A plan pivot parks the partial tree before new scope", @@ -317,6 +395,12 @@ def test_verify_reference_carries_analytics_thrash_checkpoints(self): self.assertIn("Schema before UI", text) self.assertIn("Claim a chart gap fixed only after naming write vs enrich ownership", text) self.assertIn("Write a thrash checkpoint before attaching new scope", text) + self.assertIn("intervention-must-automate", text) + self.assertIn( + "When a same-type miss forces a working-style catch mid analytics thrash, stop feature PRs until that catch is locked", + text, + ) + self.assertIn("same-type miss forces a working-style catch", text) self.assertIn( "Million-row or multi-host backfills are owned by a durable runner", text, @@ -324,6 +408,15 @@ def test_verify_reference_carries_analytics_thrash_checkpoints(self): self.assertIn("[[principle-push-not-poll]]", text) self.assertIn("not another check-wait-repeat count", text) + def test_fires_automate_mid_analytics_without_locking_catch(self): + path = os.path.join( + self.FIXTURES_DIR, "fires_automate_mid_analytics_without_locking_catch.md" + ) + with open(path, encoding="utf-8") as handle: + text = handle.read().lower() + self.assertIn("automate-me", text) + self.assertIn("defer locking the class catch", text) + def test_autonomy_reference_carries_done_gate_publication_and_pivot(self): with open(self.AUTONOMY_REF, encoding="utf-8") as handle: text = handle.read() @@ -390,6 +483,7 @@ def test_every_referenced_skill_still_exists(self): "safe-stack-push", # Command / process tokens in backticks, not skill names. "checkout", + "intervention-must-automate", "invoker-cli", "kill", "ls", diff --git a/tests/test_install.py b/tests/test_install.py index 154a73c7..bb5c219e 100644 --- a/tests/test_install.py +++ b/tests/test_install.py @@ -28,6 +28,7 @@ import tomllib import unittest from pathlib import Path +from unittest import mock REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) INSTALL_SH = os.path.join(REPO_ROOT, "install.sh") @@ -167,9 +168,7 @@ def installed_config_hook_names(fake_home): path = os.path.join(fake_home, relative) if not os.path.exists(path): continue - with open(path) as handle: - data = json.load(handle) - for command in json_commands(data.get("hooks", {})): + for command in json_commands(load_config(path).get("hooks", {})): hook = catstack_hook_from_command(command) if hook: names.add(hook) @@ -186,12 +185,103 @@ def installed_config_commands(fake_home): path = os.path.join(fake_home, relative) if not os.path.exists(path): continue - with open(path) as handle: - data = json.load(handle) - commands.extend(json_commands(data.get("hooks", {}))) + commands.extend(json_commands(load_config(path).get("hooks", {}))) return commands +sys.path.insert(0, os.path.join(REPO_ROOT, "engine", "hooks", "_runner")) +sys.path.insert(0, os.path.join(REPO_ROOT, "engine", "hooks", "_flags")) +import dispatch as _dispatch # noqa: E402 +import wrap_installed as _wrap_installed # noqa: E402 +import flags as _flags # noqa: E402 + +_HARNESS_BY_DIR = {".claude": "claude", ".cursor": "cursor", ".codex": "codex"} + + +def _harness_for_config_path(path): + parts = os.path.normpath(path).split(os.sep) + for harness_dir, harness in _HARNESS_BY_DIR.items(): + if harness_dir in parts: + return harness + raise ValueError(f"not a recognized install config path: {path}") + + +def _fake_home_for_config_path(path, harness): + marker = os.sep + f".{harness}" + os.sep + index = path.rfind(marker) + return path[:index] + + +def _record_command(record, harness): + timeout_value = record.get("timeout") + if isinstance(timeout_value, (int, float)) and not isinstance(timeout_value, bool): + budget = timeout_value - 0.5 + else: + budget = 59.5 + budget_str = _wrap_installed._format_timeout(budget) + trailing = " ".join(record.get("args") or []) + suffix = f" {trailing}" if trailing else "" + return ( + f"python3 $HOME/.{harness}/hooks/_runner/run.py --timeout {budget_str} " + f"{record['hook']}/{record['script']}{suffix}" + ) + + +def _records_to_groups(records, harness): + groups = [] + for record in records: + command = _record_command(record, harness) + matcher = record.get("matcher") + if harness == "cursor": + entry = {"command": command} + if record.get("timeout") is not None: + entry["timeout"] = record["timeout"] + if matcher is not None: + entry["matcher"] = matcher + else: + hook_obj = {"type": "command", "command": command} + if record.get("timeout") is not None: + hook_obj["timeout"] = record["timeout"] + entry = {"hooks": [hook_obj]} + if matcher is not None: + entry["matcher"] = matcher + groups.append(entry) + return groups + + +def load_config(path): + """Read an installed hooks config file (.claude/settings.json, + .cursor/hooks.json, .codex/hooks.json), expanding any event collapsed + behind one `_runner/dispatch.py` entry back into one entry per hook -- + the same shape install leaves with CATSTACK_HOOK_DISPATCHER off -- by + reading that hook's registration the same way the dispatcher itself + would (`dispatch.load_event_hooks`: the cursor registry file, or the + hook's own manifest for claude and codex). A test written against the + one-entry-per-hook layout reads the identical shape in either layout, + so the same assertion checks the same per-hook guarantee regardless of + which layout install actually produced.""" + with open(path, encoding="utf-8") as handle: + data = json.load(handle) + hooks = data.get("hooks") + if not isinstance(hooks, dict): + return data + harness = _harness_for_config_path(path) + fake_home = _fake_home_for_config_path(path, harness) + hooks_root = os.path.join(fake_home, f".{harness}", "hooks") + expanded = {} + changed = False + for event, groups in hooks.items(): + if isinstance(groups, list) and any("_runner/dispatch.py" in c for c in json_commands(groups)): + changed = True + records, _warnings = _dispatch.load_event_hooks(hooks_root, harness, event) + expanded[event] = _records_to_groups(records, harness) + else: + expanded[event] = groups + if not changed: + return data + return {**data, "hooks": expanded} + + class TestNeverTouchesRealHome(unittest.TestCase): """Safety-guard meta-test: every other test in this file builds its fake $HOME the same way (tempfile.TemporaryDirectory()) -- this asserts that @@ -255,8 +345,7 @@ def test_explicit_failures_hook_linked_and_pretooluse_wired_for_claude(self): target = os.path.join(self.fake_home, ".claude", "hooks", "explicit-failures") self.assertTrue(os.path.islink(target), target) self.assertEqual(os.readlink(target), hook_src(self.fake_home, "explicit-failures")) - with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: - settings = json.load(handle) + settings = load_config(os.path.join(self.fake_home, ".claude", "settings.json")) entries = [ entry for entry in settings["hooks"]["PreToolUse"] if any("explicit-failures/claude_pretooluse.py" in hook["command"] for hook in entry["hooks"]) @@ -272,8 +361,7 @@ def test_bound_tool_result_wired_for_claude_cursor_and_codex(self): self.assertTrue(os.path.islink(target), target) self.assertEqual(os.readlink(target), hook_src(self.fake_home, name)) - with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: - settings = json.load(handle) + settings = load_config(os.path.join(self.fake_home, ".claude", "settings.json")) claude_entries = [ entry for entry in settings["hooks"]["PreToolUse"] if any(f"{name}/claude_pre_tool_use.py" in hook["command"] for hook in entry["hooks"]) @@ -281,13 +369,11 @@ def test_bound_tool_result_wired_for_claude_cursor_and_codex(self): self.assertEqual(len(claude_entries), 1, claude_entries) self.assertEqual(claude_entries[0]["matcher"], "Bash") - with open(os.path.join(self.fake_home, ".cursor", "hooks.json")) as handle: - cursor_hooks = json.load(handle)["hooks"] + cursor_hooks = load_config(os.path.join(self.fake_home, ".cursor", "hooks.json"))["hooks"] cursor_commands = [str(entry.get("command", "")) for entry in cursor_hooks["preToolUse"]] self.assertEqual(sum(f"{name}/cursor_pre_tool_use.py" in c for c in cursor_commands), 1, cursor_commands) - with open(os.path.join(self.fake_home, ".codex", "hooks.json")) as handle: - codex_hooks = json.load(handle)["hooks"] + codex_hooks = load_config(os.path.join(self.fake_home, ".codex", "hooks.json"))["hooks"] codex_commands = [hook["command"] for entry in codex_hooks["PreToolUse"] for hook in entry["hooks"]] self.assertEqual(sum(f"{name}/codex_pre_tool_use.py" in c for c in codex_commands), 1, codex_commands) @@ -298,8 +384,7 @@ def test_text_match_decision_warn_wired_for_claude_cursor_and_codex(self): self.assertTrue(os.path.islink(target), target) self.assertEqual(os.readlink(target), hook_src(self.fake_home, name)) - with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: - settings = json.load(handle) + settings = load_config(os.path.join(self.fake_home, ".claude", "settings.json")) claude_entries = [ entry for entry in settings["hooks"]["PreToolUse"] if any(f"{name}/claude_pretooluse.py" in hook["command"] for hook in entry["hooks"]) @@ -308,19 +393,16 @@ def test_text_match_decision_warn_wired_for_claude_cursor_and_codex(self): self.assertEqual(claude_entries[0]["matcher"], "Edit|Write|MultiEdit") self.assertIn("_runner/run.py", claude_entries[0]["hooks"][0]["command"]) - with open(os.path.join(self.fake_home, ".cursor", "hooks.json")) as handle: - cursor_hooks = json.load(handle)["hooks"] + cursor_hooks = load_config(os.path.join(self.fake_home, ".cursor", "hooks.json"))["hooks"] cursor_commands = [str(entry.get("command", "")) for entry in cursor_hooks["preToolUse"]] self.assertEqual(sum(f"{name}/cursor_pretooluse.py" in c for c in cursor_commands), 1, cursor_commands) - with open(os.path.join(self.fake_home, ".codex", "hooks.json")) as handle: - codex_hooks = json.load(handle)["hooks"] + codex_hooks = load_config(os.path.join(self.fake_home, ".codex", "hooks.json"))["hooks"] codex_commands = [hook["command"] for entry in codex_hooks["PreToolUse"] for hook in entry["hooks"]] self.assertEqual(sum(f"{name}/codex_pretooluse.py" in c for c in codex_commands), 1, codex_commands) def _claude_hook_commands(self, event): - with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: - settings = json.load(handle) + settings = load_config(os.path.join(self.fake_home, ".claude", "settings.json")) return [h["command"] for e in settings["hooks"][event] for h in e["hooks"]] def test_restated_constraint_linked_and_prompt_submit_wired_for_claude(self): @@ -333,8 +415,7 @@ def test_wait_needs_wakeup_linked_and_both_events_wired_for_claude(self): target = os.path.join(self.fake_home, ".claude", "hooks", "wait-needs-wakeup") self.assertTrue(os.path.islink(target), target) self.assertEqual(os.readlink(target), hook_src(self.fake_home, "wait-needs-wakeup")) - with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: - settings = json.load(handle) + settings = load_config(os.path.join(self.fake_home, ".claude", "settings.json")) pre = [h["command"] for e in settings["hooks"]["PreToolUse"] for h in e["hooks"]] stop = [h["command"] for e in settings["hooks"]["Stop"] for h in e["hooks"]] self.assertTrue(any("wait-needs-wakeup/claude_pretooluse.py" in c for c in pre), pre) @@ -351,8 +432,7 @@ def test_answer_overrides_menu_linked_and_posttooluse_wired_for_claude(self): target = os.path.join(self.fake_home, ".claude", "hooks", "answer-overrides-menu") self.assertTrue(os.path.islink(target), target) self.assertEqual(os.readlink(target), hook_src(self.fake_home, "answer-overrides-menu")) - with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: - settings = json.load(handle) + settings = load_config(os.path.join(self.fake_home, ".claude", "settings.json")) entries = [ entry for entry in settings["hooks"]["PostToolUse"] if any("answer-overrides-menu/claude_posttooluse.py" in hook["command"] for hook in entry["hooks"]) @@ -364,8 +444,7 @@ def test_serial_option_guard_linked_and_pretooluse_wired_for_claude(self): target = os.path.join(self.fake_home, ".claude", "hooks", "serial-option-guard") self.assertTrue(os.path.islink(target), target) self.assertEqual(os.readlink(target), hook_src(self.fake_home, "serial-option-guard")) - with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: - settings = json.load(handle) + settings = load_config(os.path.join(self.fake_home, ".claude", "settings.json")) entries = [ entry for entry in settings["hooks"]["PreToolUse"] if any("serial-option-guard/claude_pretooluse.py" in hook["command"] for hook in entry["hooks"]) @@ -409,8 +488,7 @@ def test_history_before_reversal_linked_and_bash_pretooluse_wired_for_claude(sel target = os.path.join(self.fake_home, ".claude", "hooks", "history-before-reversal") self.assertTrue(os.path.islink(target), target) self.assertEqual(os.readlink(target), hook_src(self.fake_home, "history-before-reversal")) - with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: - settings = json.load(handle) + settings = load_config(os.path.join(self.fake_home, ".claude", "settings.json")) entries = [ entry for entry in settings["hooks"]["PreToolUse"] if any("history-before-reversal/claude_pretooluse.py" in hook["command"] for hook in entry["hooks"]) @@ -422,8 +500,7 @@ def test_external_claim_gate_linked_and_bash_pretooluse_wired_for_claude(self): target = os.path.join(self.fake_home, ".claude", "hooks", "external-claim-gate") self.assertTrue(os.path.islink(target), target) self.assertEqual(os.readlink(target), hook_src(self.fake_home, "external-claim-gate")) - with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: - settings = json.load(handle) + settings = load_config(os.path.join(self.fake_home, ".claude", "settings.json")) entries = [ entry for entry in settings["hooks"]["PreToolUse"] if any("external-claim-gate/claude_pretooluse.py" in hook["command"] for hook in entry["hooks"]) @@ -450,8 +527,7 @@ def test_publish_act_guard_linked_and_bash_pretooluse_wired_for_claude(self): target = os.path.join(self.fake_home, ".claude", "hooks", "publish-act-guard") self.assertTrue(os.path.islink(target), target) self.assertEqual(os.readlink(target), hook_src(self.fake_home, "publish-act-guard")) - with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: - settings = json.load(handle) + settings = load_config(os.path.join(self.fake_home, ".claude", "settings.json")) entries = [ entry for entry in settings["hooks"]["PreToolUse"] if any("publish-act-guard/claude_pretooluse.py" in hook["command"] for hook in entry["hooks"]) @@ -463,8 +539,7 @@ def test_retired_agent_routing_guard_is_gone(self): self.assertFalse( os.path.exists(os.path.join(self.fake_home, ".claude", "hooks", "agent-routing-guard")) ) - with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: - settings = json.load(handle) + settings = load_config(os.path.join(self.fake_home, ".claude", "settings.json")) stale = [ entry for entry in settings["hooks"]["PreToolUse"] if any("agent-routing-guard" in hook["command"] for hook in entry["hooks"]) @@ -475,8 +550,7 @@ def test_categorical_scope_guard_linked_and_bash_pretooluse_wired_for_claude(sel target = os.path.join(self.fake_home, ".claude", "hooks", "categorical-scope-guard") self.assertTrue(os.path.islink(target), target) self.assertEqual(os.readlink(target), hook_src(self.fake_home, "categorical-scope-guard")) - with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: - settings = json.load(handle) + settings = load_config(os.path.join(self.fake_home, ".claude", "settings.json")) entries = [ entry for entry in settings["hooks"]["PreToolUse"] if any("categorical-scope-guard/claude_pretooluse.py" in hook["command"] for hook in entry["hooks"]) @@ -497,8 +571,7 @@ def test_hedge_runs_prove_it_linked_and_stop_wired_for_claude(self): target = os.path.join(self.fake_home, ".claude", "hooks", "hedge-runs-prove-it") self.assertTrue(os.path.islink(target), target) self.assertEqual(os.readlink(target), hook_src(self.fake_home, "hedge-runs-prove-it")) - with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: - settings = json.load(handle) + settings = load_config(os.path.join(self.fake_home, ".claude", "settings.json")) stop = [h["command"] for e in settings["hooks"]["Stop"] for h in e["hooks"]] self.assertTrue(any("hedge-runs-prove-it/claude_stop_check.py" in c for c in stop), stop) @@ -533,8 +606,7 @@ def test_new_file_callout_linked_and_stop_wired_for_claude(self): target = os.path.join(self.fake_home, ".claude", "hooks", "new-file-callout") self.assertTrue(os.path.islink(target), target) self.assertEqual(os.readlink(target), hook_src(self.fake_home, "new-file-callout")) - with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: - settings = json.load(handle) + settings = load_config(os.path.join(self.fake_home, ".claude", "settings.json")) stop = [h["command"] for e in settings["hooks"]["Stop"] for h in e["hooks"]] self.assertTrue(any("new-file-callout/claude_stop_check.py" in c for c in stop), stop) @@ -542,8 +614,7 @@ def test_agent_relay_attribution_linked_and_stop_wired_for_claude(self): target = os.path.join(self.fake_home, ".claude", "hooks", "agent-relay-attribution") self.assertTrue(os.path.islink(target), target) self.assertEqual(os.readlink(target), hook_src(self.fake_home, "agent-relay-attribution")) - with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: - settings = json.load(handle) + settings = load_config(os.path.join(self.fake_home, ".claude", "settings.json")) stop = [h["command"] for e in settings["hooks"]["Stop"] for h in e["hooks"]] self.assertTrue(any("agent-relay-attribution/claude_stop_check.py" in c for c in stop), stop) @@ -551,8 +622,7 @@ def test_scratchpad_collision_linked_and_pretooluse_wired_for_claude(self): target = os.path.join(self.fake_home, ".claude", "hooks", "scratchpad-collision") self.assertTrue(os.path.islink(target), target) self.assertEqual(os.readlink(target), hook_src(self.fake_home, "scratchpad-collision")) - with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: - settings = json.load(handle) + settings = load_config(os.path.join(self.fake_home, ".claude", "settings.json")) pre = [h["command"] for e in settings["hooks"]["PreToolUse"] for h in e["hooks"]] self.assertTrue(any("scratchpad-collision/claude_pretooluse.py" in c for c in pre), pre) @@ -565,13 +635,11 @@ def test_reflect_on_thrash_wired_for_claude_and_cursor(self): hook_src(self.fake_home, "reflect-on-thrash"), ) settings_path = os.path.join(self.fake_home, ".claude", "settings.json") - with open(settings_path) as handle: - settings = json.load(handle) + settings = load_config(settings_path) stop_commands = [h["command"] for e in settings["hooks"]["Stop"] for h in e["hooks"]] self.assertTrue(any("claude_stop_reflect.py" in c for c in stop_commands), stop_commands) cursor_path = os.path.join(self.fake_home, ".cursor", "hooks.json") - with open(cursor_path) as handle: - cursor_hooks = json.load(handle) + cursor_hooks = load_config(cursor_path) stop = cursor_hooks["hooks"]["stop"] session_end = cursor_hooks["hooks"]["sessionEnd"] self.assertTrue(any("cursor_session.py" in str(e.get("command", "")) for e in stop), stop) @@ -619,8 +687,7 @@ def test_cat_mode_default_linked_and_prompt_hook_wired_for_claude(self): for agent_dir in (".cursor", ".codex"): self.assertFalse(os.path.lexists(os.path.join(self.fake_home, agent_dir, "hooks", "cat-mode-default"))) settings_path = os.path.join(self.fake_home, ".claude", "settings.json") - with open(settings_path) as handle: - settings = json.load(handle) + settings = load_config(settings_path) prompt_commands = [h["command"] for e in settings["hooks"]["UserPromptSubmit"] for h in e["hooks"]] matching = [c for c in prompt_commands if "cat-mode-default/claude_prompt_submit.py" in c] self.assertEqual(len(matching), 1, prompt_commands) @@ -632,8 +699,7 @@ def test_scope_lock_wired_for_claude_cursor_and_codex(self): self.assertTrue(os.path.islink(target), target) self.assertEqual(os.readlink(target), hook_src(self.fake_home, "scope-lock")) - with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: - settings = json.load(handle) + settings = load_config(os.path.join(self.fake_home, ".claude", "settings.json")) prompt_commands = [ hook["command"] for entry in settings["hooks"]["UserPromptSubmit"] @@ -647,8 +713,7 @@ def test_scope_lock_wired_for_claude_cursor_and_codex(self): self.assertTrue(any("claude_prompt_scope.py" in command for command in prompt_commands)) self.assertTrue(any("claude_pretool_scope.py" in command for command in pretool_commands)) - with open(os.path.join(self.fake_home, ".cursor", "hooks.json")) as handle: - cursor_hooks = json.load(handle)["hooks"] + cursor_hooks = load_config(os.path.join(self.fake_home, ".cursor", "hooks.json"))["hooks"] self.assertTrue(any( "cursor_before_submit.py" in str(entry.get("command", "")) for entry in cursor_hooks["beforeSubmitPrompt"] @@ -658,8 +723,7 @@ def test_scope_lock_wired_for_claude_cursor_and_codex(self): for entry in cursor_hooks["preToolUse"] )) - with open(os.path.join(self.fake_home, ".codex", "hooks.json")) as handle: - codex_hooks = json.load(handle)["hooks"] + codex_hooks = load_config(os.path.join(self.fake_home, ".codex", "hooks.json"))["hooks"] codex_prompt_commands = [ hook["command"] for entry in codex_hooks["UserPromptSubmit"] @@ -679,8 +743,7 @@ def test_split_scope_wired_for_claude_cursor_and_codex(self): self.assertTrue(os.path.islink(target), target) self.assertEqual(os.readlink(target), hook_src(self.fake_home, "split-scope")) - with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: - settings = json.load(handle) + settings = load_config(os.path.join(self.fake_home, ".claude", "settings.json")) claude_prompt_commands = [ hook["command"] for entry in settings["hooks"]["UserPromptSubmit"] @@ -688,8 +751,7 @@ def test_split_scope_wired_for_claude_cursor_and_codex(self): ] self.assertTrue(any("split-scope/claude_prompt_submit.py" in command for command in claude_prompt_commands)) - with open(os.path.join(self.fake_home, ".cursor", "hooks.json")) as handle: - cursor_hooks = json.load(handle)["hooks"] + cursor_hooks = load_config(os.path.join(self.fake_home, ".cursor", "hooks.json"))["hooks"] self.assertTrue(any( "split-scope/cursor_before_submit.py" in str(entry.get("command", "")) for entry in cursor_hooks["beforeSubmitPrompt"] @@ -699,8 +761,7 @@ def test_split_scope_wired_for_claude_cursor_and_codex(self): for entry in cursor_hooks["postToolUse"] )) - with open(os.path.join(self.fake_home, ".codex", "hooks.json")) as handle: - codex_hooks = json.load(handle)["hooks"] + codex_hooks = load_config(os.path.join(self.fake_home, ".codex", "hooks.json"))["hooks"] codex_prompt_commands = [ hook["command"] for entry in codex_hooks["UserPromptSubmit"] @@ -714,8 +775,7 @@ def test_hook_health_wired_for_claude_cursor_and_codex(self): self.assertTrue(os.path.islink(target), target) self.assertEqual(os.readlink(target), hook_src(self.fake_home, "hook-health")) - with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: - settings = json.load(handle) + settings = load_config(os.path.join(self.fake_home, ".claude", "settings.json")) claude_prompt_commands = [ hook["command"] for entry in settings["hooks"]["UserPromptSubmit"] @@ -723,15 +783,13 @@ def test_hook_health_wired_for_claude_cursor_and_codex(self): ] self.assertTrue(any("hook-health/claude_prompt_submit.py" in command for command in claude_prompt_commands)) - with open(os.path.join(self.fake_home, ".cursor", "hooks.json")) as handle: - cursor_hooks = json.load(handle)["hooks"] + cursor_hooks = load_config(os.path.join(self.fake_home, ".cursor", "hooks.json"))["hooks"] self.assertTrue(any( "hook-health/cursor_before_submit.py" in str(entry.get("command", "")) for entry in cursor_hooks["beforeSubmitPrompt"] )) - with open(os.path.join(self.fake_home, ".codex", "hooks.json")) as handle: - codex_hooks = json.load(handle)["hooks"] + codex_hooks = load_config(os.path.join(self.fake_home, ".codex", "hooks.json"))["hooks"] codex_prompt_commands = [ hook["command"] for entry in codex_hooks["UserPromptSubmit"] @@ -746,12 +804,9 @@ def test_skill_usage_log_is_linked_but_not_wired_while_registry_mode_is_off(self self.assertTrue(os.path.islink(target), target) self.assertEqual(os.readlink(target), hook_src(self.fake_home, "skill-usage-log")) - with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: - claude_hooks = json.load(handle)["hooks"] - with open(os.path.join(self.fake_home, ".cursor", "hooks.json")) as handle: - cursor_hooks = json.load(handle)["hooks"] - with open(os.path.join(self.fake_home, ".codex", "hooks.json")) as handle: - codex_hooks = json.load(handle)["hooks"] + claude_hooks = load_config(os.path.join(self.fake_home, ".claude", "settings.json"))["hooks"] + cursor_hooks = load_config(os.path.join(self.fake_home, ".cursor", "hooks.json"))["hooks"] + codex_hooks = load_config(os.path.join(self.fake_home, ".codex", "hooks.json"))["hooks"] expected = ( (claude_hooks, "PreToolUse", "skill-usage-log/claude_pretooluse_log.py"), (claude_hooks, "UserPromptSubmit", "skill-usage-log/claude_prompt_submit.py"), @@ -780,8 +835,7 @@ def test_llm_judge_inbox_wired_for_claude_cursor_and_codex(self): env={**os.environ, "HOME": self.fake_home, "CATSTACK_LLM_JUDGE_STATE_DIR": os.path.join(self.fake_home, "judge")}, ) self.assertEqual((result.returncode, result.stdout, result.stderr), (0, "", "")) - with open(os.path.join(self.fake_home, ".cursor", "hooks.json")) as handle: - cursor_stop = json.load(handle)["hooks"]["stop"] + cursor_stop = load_config(os.path.join(self.fake_home, ".cursor", "hooks.json"))["hooks"]["stop"] self.assertEqual(sum("llm-judge/cursor_session.py" in str(e.get("command", "")) for e in cursor_stop), 1, cursor_stop) def test_unverified_tag_check_is_linked_but_not_wired_while_registry_mode_is_off(self): @@ -797,8 +851,7 @@ def test_unverified_tag_check_is_linked_but_not_wired_while_registry_mode_is_off self.assertEqual(os.readlink(target), hook_src(self.fake_home, "unverified-tag-check")) claude_stop = self._claude_hook_commands("Stop") self.assertEqual(sum("unverified-tag-check/claude_stop_check.py" in command for command in claude_stop), 0, claude_stop) - with open(os.path.join(self.fake_home, ".cursor", "hooks.json")) as handle: - cursor_stop = json.load(handle)["hooks"]["stop"] + cursor_stop = load_config(os.path.join(self.fake_home, ".cursor", "hooks.json"))["hooks"]["stop"] self.assertEqual(sum("unverified-tag-check/cursor_session.py" in str(entry.get("command", "")) for entry in cursor_stop), 0, cursor_stop) with open(config_path) as handle: text = handle.read() @@ -1221,8 +1274,7 @@ def test_fresh_home_gets_both_hooks(self): run_install(fake_home) settings_path = os.path.join(fake_home, ".claude", "settings.json") self.assertTrue(os.path.exists(settings_path)) - with open(settings_path) as f: - settings = json.load(f) + settings = load_config(settings_path) stop_commands = [h["command"] for e in settings["hooks"]["Stop"] for h in e["hooks"]] prompt_commands = [h["command"] for e in settings["hooks"]["UserPromptSubmit"] for h in e["hooks"]] self.assertTrue(any("claude_stop_check.py" in c for c in stop_commands)) @@ -1236,8 +1288,7 @@ def test_preexisting_unrelated_keys_survive(self): with open(os.path.join(claude_dir, "settings.json"), "w") as f: json.dump({"model": "sonnet", "theme": "dark"}, f) run_install(fake_home) - with open(os.path.join(claude_dir, "settings.json")) as f: - settings = json.load(f) + settings = load_config(os.path.join(claude_dir, "settings.json")) self.assertEqual(settings["model"], "sonnet") self.assertEqual(settings["theme"], "dark") self.assertIn("Stop", settings["hooks"]) @@ -1272,8 +1323,7 @@ def test_install_writes_worker_flagged_hooks_only(self): with open(worker_settings_path, encoding="utf-8") as handle: worker = json.load(handle) - with open(os.path.join(fake_home, ".claude", "settings.json"), encoding="utf-8") as handle: - interactive = json.load(handle) + interactive = load_config(os.path.join(fake_home, ".claude", "settings.json")) self.assertEqual(worker.get("model"), "keep-me") worker_cmds = self._hook_commands(worker) @@ -1314,9 +1364,8 @@ def test_preseeded_config_toml_survives_without_notify_wiring(self): text = f.read() self.assertIn('model = "gpt-5"', text) self.assertNotIn("notify =", text) - with open(os.path.join(codex_dir, "hooks.json")) as f: - hooks_text = f.read() - self.assertIn("llm-judge/codex_post_tool_use.py", hooks_text) + hooks_data = load_config(os.path.join(codex_dir, "hooks.json")) + self.assertIn("llm-judge/codex_post_tool_use.py", json.dumps(hooks_data)) class TestIdempotency(unittest.TestCase): @@ -1591,8 +1640,7 @@ def test_settings_entry_for_a_removed_link_is_pruned(self): json.dump({"hooks": {"Stop": [{"matcher": "", "hooks": [{"type": "command", "command": retired}]}]}}, f) result = run_install(self.fake_home) self.assertEqual(result.returncode, 0, result.stderr) - with open(settings_path) as f: - settings = json.load(f) + settings = load_config(settings_path) all_commands = [h["command"] for groups in settings["hooks"].values() for g in groups for h in g["hooks"]] self.assertFalse([c for c in all_commands if "retired-hook" in c], all_commands) @@ -1630,8 +1678,7 @@ def test_dead_settings_entry_is_pruned_and_rest_survives(self): }, f) result = run_install(self.fake_home) self.assertEqual(result.returncode, 0, result.stderr) - with open(settings_path) as f: - settings = json.load(f) + settings = load_config(settings_path) all_commands = [h["command"] for groups in settings["hooks"].values() for g in groups for h in g["hooks"]] self.assertNotIn(dead, all_commands) self.assertTrue(any("diu-stop/claude_stop_check.py" in c for c in all_commands)) @@ -1656,8 +1703,7 @@ def tearDown(self): def test_every_claude_hook_entrypoint_is_in_settings(self): settings_path = os.path.join(self.fake_home, ".claude", "settings.json") - with open(settings_path) as handle: - settings = json.load(handle) + settings = load_config(settings_path) wired = { hook["command"] for entries in settings.get("hooks", {}).values() @@ -1816,8 +1862,7 @@ def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.fake_home = self.tmp.name self.result = run_install(self.fake_home) - with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: - self.settings = json.load(handle) + self.settings = load_config(os.path.join(self.fake_home, ".claude", "settings.json")) sys.path.insert(0, os.path.join(REPO_ROOT, "scripts", "install")) import mirror_stop_hooks_to_subagent_stop as mirror_mod self.manifests = mirror_mod.load_manifests() @@ -1860,8 +1905,8 @@ def test_subagent_stop_entries_carry_no_tool_matcher(self): def test_rerun_keeps_subagent_stop_unchanged(self): second = run_install(self.fake_home) self.assertEqual(second.returncode, 0, second.stderr) - with open(os.path.join(self.fake_home, ".claude", "settings.json")) as handle: - self.assertEqual(json.load(handle)["hooks"]["SubagentStop"], self.settings["hooks"]["SubagentStop"]) + after = load_config(os.path.join(self.fake_home, ".claude", "settings.json")) + self.assertEqual(after["hooks"]["SubagentStop"], self.settings["hooks"]["SubagentStop"]) class TestCatModeDefaultAgentHook(unittest.TestCase): @@ -1872,8 +1917,7 @@ def test_agent_pretooluse_hook_wired_for_claude(self): with tempfile.TemporaryDirectory() as fake_home: result = run_install(fake_home) self.assertEqual(result.returncode, 0, result.stderr) - with open(os.path.join(fake_home, ".claude", "settings.json")) as handle: - settings = json.load(handle) + settings = load_config(os.path.join(fake_home, ".claude", "settings.json")) entries = [ entry for entry in settings["hooks"]["PreToolUse"] if any("cat-mode-default/claude_pretooluse_agent.py" in hook["command"] for hook in entry["hooks"]) @@ -1890,8 +1934,7 @@ def test_fanout_routing_guard_linked_and_wired_for_claude(self): target = os.path.join(fake_home, ".claude", "hooks", "fanout-routing-guard") self.assertTrue(os.path.islink(target), target) self.assertEqual(os.readlink(target), hook_src(fake_home, "fanout-routing-guard")) - with open(os.path.join(fake_home, ".claude", "settings.json")) as handle: - settings = json.load(handle) + settings = load_config(os.path.join(fake_home, ".claude", "settings.json")) entries = [ entry for entry in settings["hooks"]["PreToolUse"] if any("fanout-routing-guard/claude_pretooluse_agent.py" in hook["command"] for hook in entry["hooks"]) @@ -1918,8 +1961,7 @@ def test_install_replaces_dangling_link_with_real_merged_file(self): self.assertEqual(proc.returncode, 0, proc.stderr) self.assertTrue(os.path.isfile(hooks_path)) self.assertFalse(os.path.islink(hooks_path)) - with open(hooks_path) as handle: - data = json.load(handle) + data = load_config(hooks_path) commands = [entry.get("command", "") for entry in data["hooks"]["stop"]] self.assertTrue(any("diu-stop/cursor_stop_check.py" in c for c in commands), commands) self.assertTrue(all("_runner/run.py" in c for c in commands), commands) @@ -2202,12 +2244,28 @@ def _cursor_hooks(self, fake_home): return json.load(handle) def test_default_is_off_and_byte_identical_to_no_flag_at_all(self): - with tempfile.TemporaryDirectory() as plain_home, tempfile.TemporaryDirectory() as off_home: - plain = run_install(plain_home, timeout=self.INSTALL_TIMEOUT) + """The default install (CATSTACK_HOOK_DISPATCHER unset) must equal an + install that explicitly sets the flag to whatever `flags.py` itself + resolves as the default -- today off, so this reduces to the + flag-off install, but it stays meaningful if the default ever flips + to on instead of silently comparing against the wrong side. The + ambient environment is cleared of the key first, so a caller that + happens to have CATSTACK_HOOK_DISPATCHER set (for example the + flag-on run of the full suite) cannot leak into the "default" side + of this comparison.""" + clean_env = {k: v for k, v in os.environ.items() if k != _flags.HOOK_DISPATCHER} + default_on = _flags.hook_dispatcher_on(clean_env, REPO_ROOT) + with tempfile.TemporaryDirectory() as plain_home, tempfile.TemporaryDirectory() as explicit_home: + with mock.patch.dict(os.environ, clean_env, clear=True): + plain = run_install(plain_home, timeout=self.INSTALL_TIMEOUT) self.assertEqual(plain.returncode, 0, plain.stderr) - off = run_install(off_home, extra_env={"CATSTACK_HOOK_DISPATCHER": "0"}, timeout=self.INSTALL_TIMEOUT) - self.assertEqual(off.returncode, 0, off.stderr) - self.assertEqual(self._claude_settings(plain_home), self._claude_settings(off_home)) + explicit = run_install( + explicit_home, + extra_env={_flags.HOOK_DISPATCHER: "1" if default_on else "0"}, + timeout=self.INSTALL_TIMEOUT, + ) + self.assertEqual(explicit.returncode, 0, explicit.stderr) + self.assertEqual(self._claude_settings(plain_home), self._claude_settings(explicit_home)) def test_on_collapses_claude_pretooluse_to_one_dispatcher_entry(self): with tempfile.TemporaryDirectory() as fake_home: