From 5a9e5c1b6e5ec6032e520a182892ba20a675846f Mon Sep 17 00:00:00 2001 From: CI Bot Date: Sun, 11 Oct 2026 14:27:01 +0000 Subject: [PATCH 1/2] Let the install tests read either hook layout. tests/test_install.py and engine/hooks/_runner/tests/test_wrap_installed.py asserted one settings entry per hook, so every per-hook guarantee they check failed once CATSTACK_HOOK_DISPATCHER (#1122) collapsed an event to a single _runner/dispatch.py entry: 70 failures and 1 error on #1122's head, all in these two files. load_config() in tests/test_install.py now reads an installed config and expands a collapsed event back into one entry per hook, resolving each hook's registration the way dispatch.py itself does (dispatch.load_event_hooks: the cursor registry, or the hook's own manifest for claude and codex). Each assertion therefore sees the identical shape in both layouts and keeps checking the same guarantee -- right event, right harness, Stop mirrored to SubagentStop or opted out, dead entries pruned, non-catstack entries untouched. test_wrap_installed.py's fixture pins the flag off, since it asserts the settings bytes install writes rather than a hook's registration. #1122's own test_default_is_off_and_byte_identical_to_no_flag_at_all now compares the default install against an install that sets the flag to whatever flags.py resolves as the default, and clears the key from its own environment first, so the flag-on run of the full suite cannot leak into the "default" side of the comparison. The CATSTACK_HOOK_DISPATCHER default is left off. Flipping it on is gated on both full-suite runs passing, and the source branch plan/hook-backlog-1c-install-tests-accept-the-one-entry-per-event-layout-default-on-only-if-both-suites-pass never changed engine/hooks/_flags/flags.py, so that gate was not met there and this commit does not claim it. Cherry-picked from 6330abf291561cace7301948bda714d0b8795f43 on that branch. The branch's two other content commits (engine/hooks/_runner/doctor.py, scripts/test/run_skill_scenarios.py) are unrelated to this claim and are left behind. --- .../_runner/tests/test_wrap_installed.py | 3 + tests/test_install.py | 278 +++++++++++------- 2 files changed, 171 insertions(+), 110 deletions(-) diff --git a/engine/hooks/_runner/tests/test_wrap_installed.py b/engine/hooks/_runner/tests/test_wrap_installed.py index 008b06bd4..6de0ec8a3 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/tests/test_install.py b/tests/test_install.py index e065f2d94..675124847 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() @@ -1809,8 +1855,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() @@ -1847,8 +1892,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): @@ -1859,8 +1904,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"]) @@ -1877,8 +1921,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"]) @@ -1905,8 +1948,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) @@ -2189,12 +2231,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: From 1b46a91cadcb9c96ca915f5886077253ef6a36bb Mon Sep 17 00:00:00 2001 From: Invoker Bot Date: Sun, 11 Oct 2026 14:43:08 +0000 Subject: [PATCH 2/2] =?UTF-8?q?invoker:=20wf-1791728053338-10/publish-both?= =?UTF-8?q?-layout-install-tests=20=E2=80=94=20Review=20claim:=20tests.tes?= =?UTF-8?q?t=5Finstall=20and=20test=5Fwrap=5Finstalled=20accept=20both=20i?= =?UTF-8?q?nstall=20layouts=20(one=20entry=20per=20hook=20with=20CATSTACK?= =?UTF-8?q?=5FHOOK=5FDISPATCHER=20off,=20one=20dispatcher=20entry=20per=20?= =?UTF-8?q?event=20with=20it=20on),=20and=20the=20flag=20defaults=20to=20o?= =?UTF-8?q?n=20only=20if=20bash=20scripts/test/run=5Fall=5Ftests.sh=20pass?= =?UTF-8?q?es=20with=20the=20flag=20on=20and=20off.=20Review=20lane:=20beh?= =?UTF-8?q?avior=20Safety=20invariant:=20If=20either=20suite=20run=20fails?= =?UTF-8?q?,=20the=20default=20stays=20off.=20Per-hook=20guarantees=20stay?= =?UTF-8?q?=20checked=20in=20the=20dispatcher=20layout.=20Files:=20tests/t?= =?UTF-8?q?est=5Finstall.py,=20tests/test=5Fwrap=5Finstalled.py=20Change?= =?UTF-8?q?=20types:=20test=20Acceptance=20criteria:=20This=20workflow=20p?= =?UTF-8?q?ublishes=20one=20pull=20request=20for=20the=20install-layout=20?= =?UTF-8?q?claim=20only.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Exit code: 0 Invoker-Finalize-Id: f459f9c9-a623-4fd7-9708-21cc3ae4cbda