From 710df8288b4d7d7a172ac915225cec5891ebaf09 Mon Sep 17 00:00:00 2001 From: Edbert Chan Date: Sat, 12 Sep 2026 12:11:41 -0700 Subject: [PATCH] hooks: reflect/automate-me enforcement is opt-in behind one flag Four hooks push the user toward /reflect and automate-me: scope-lock (which stops every tool after a second scope correction), reflect-on-thrash, wrong-check-reflect and verdict-flip-watch. None of them could be switched off. They now do nothing unless CATSTACK_REFLECT_ENFORCEMENT is on, read from the environment, $CATSTACK_ENV_FILE, the repo's .env or ~/.catstack.env. The reader moves into engine/hooks/_flags so the four hooks share one lookup instead of four copies drifting apart. It also stops returning "not set" for an .env file that exists but cannot be read: that case raises inside the reader, is collected in FlagLookup.unreadable, and enforcement_gate names the file on stderr before failing closed. Each gate sits ahead of the hook's recorder, not just its output. A disabled scope-lock that kept counting corrections would hard stop the first tool call after the flag went on; a disabled reflect-on-thrash that kept writing its deferred marker would hand the prompt to a later session. install.sh links _flags into the Claude, Cursor and Codex hook folders, since scope-lock and wrong-check-reflect load it on all three. A missing link is an import error at hook start, which fails open and looks like "not opted in"; _flags/tests/test_installed_layout.py pins the link per harness. The scenario runner switches the flag on for each scenario. Without that, two expect_fire/expect_enqueue scenarios failed, and three expect_silent checks on these hooks passed only because the hook was off. A new test inverts two real scenarios with the flag set to 0 and requires both to be reported. frustration-watchdog is deliberately left alone: it enforces the live-demo "end the wait" rule and its output never mentions reflect or automate-me. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_011vy9sMw49j6PqS2NLRBYhs Change-Id: I82a634782fcb75591db4807d0d67e028c02ae252 --- docs/ecosystem.md | 5 +- engine/hooks/_flags/README.md | 86 +++++++ engine/hooks/_flags/flags.py | 229 +++++++++++++++++ engine/hooks/_flags/tests/test_flags.py | 239 ++++++++++++++++++ .../_flags/tests/test_installed_layout.py | 161 ++++++++++++ engine/hooks/reflect-on-thrash/README.md | 15 ++ engine/hooks/reflect-on-thrash/detect.py | 11 + .../reflect-on-thrash/tests/test_hooks.py | 62 +++++ engine/hooks/scope-lock/README.md | 16 ++ engine/hooks/scope-lock/detect.py | 27 ++ engine/hooks/scope-lock/tests/test_hooks.py | 81 ++++++ engine/hooks/verdict-flip-watch/README.md | 15 ++ engine/hooks/verdict-flip-watch/detect.py | 8 + .../verdict-flip-watch/tests/test_hooks.py | 51 ++++ engine/hooks/wrong-check-reflect/README.md | 15 ++ engine/hooks/wrong-check-reflect/detect.py | 8 + .../wrong-check-reflect/tests/test_hooks.py | 20 ++ install.sh | 3 + scripts/run_skill_scenarios.py | 22 +- tests/test_skill_scenarios.py | 24 ++ 20 files changed, 1096 insertions(+), 2 deletions(-) create mode 100644 engine/hooks/_flags/README.md create mode 100644 engine/hooks/_flags/flags.py create mode 100644 engine/hooks/_flags/tests/test_flags.py create mode 100644 engine/hooks/_flags/tests/test_installed_layout.py diff --git a/docs/ecosystem.md b/docs/ecosystem.md index 13380747..ada557da 100644 --- a/docs/ecosystem.md +++ b/docs/ecosystem.md @@ -79,7 +79,10 @@ again. | `named-verb-guard` | hook | | `plan-discipline` | hook (not always installed) | | `pr-schema-gate` | hook (advisory; PreToolUse on shell tools; checks direct PR text writes with the repo's own `scripts/validate-pr-body.mjs` and reminds about the stack follow-up; never blocks) | -| `reflect-on-thrash` | hook | +| `reflect-on-thrash` | hook (off unless `CATSTACK_REFLECT_ENFORCEMENT=1`) | +| `scope-lock` | hook (off unless `CATSTACK_REFLECT_ENFORCEMENT=1`; stops every tool after a second scope correction) | +| `wrong-check-reflect` | hook (off unless `CATSTACK_REFLECT_ENFORCEMENT=1`) | +| `verdict-flip-watch` | hook (off unless `CATSTACK_REFLECT_ENFORCEMENT=1`) | | `restart-risk-check` | hook | | `restated-constraint` | hook | | `repeat-error-stop` | hook (blocks blind repeated failures; nudges when one signature survives edit epochs) | diff --git a/engine/hooks/_flags/README.md b/engine/hooks/_flags/README.md new file mode 100644 index 00000000..7d7da181 --- /dev/null +++ b/engine/hooks/_flags/README.md @@ -0,0 +1,86 @@ +# _flags + +One reader for "is this catstack flag on", shared by every hook that can be +switched off. + +## The switch + +`CATSTACK_REFLECT_ENFORCEMENT` turns reflect/automate-me enforcement on. It is +**off unless something sets it**. Four hooks answer to it: + +| Hook | What it does when on | +|---|---| +| `scope-lock` | stops every tool after a second scope correction, until the user types `/reflect` and `automate-me` | +| `reflect-on-thrash` | asks for a reflect at the end of a thrashy session | +| `wrong-check-reflect` | queues a judge on a retraction-shaped reply | +| `verdict-flip-watch` | notes a verifier that passed and then failed | + +Turn it on for a machine: + +```sh +echo 'CATSTACK_REFLECT_ENFORCEMENT=1' >> ~/.catstack.env +``` + +or for one repo, in that repo's `.env`, or by exporting it in the shell. + +`frustration-watchdog` is deliberately **not** in the table. It enforces the +live-demo "end the wait" rule and never mentions reflect or automate-me; the +only reason it reads like a reflect hook is a comment saying a reflect pass is +what motivated it. + +## Where the value comes from + +First source that defines the key wins: + +1. the process environment +2. the file named by `$CATSTACK_ENV_FILE` +3. `/.env`, walking up from the hook payload's `cwd` +4. `~/.catstack.env` + +Files are read as plain `KEY=VALUE` lines, never sourced, and no key other +than the one asked for is kept. `1`, `true`, `yes` and `on` mean on; anything +else, including an empty value, means off. + +## Three outcomes, not two + +A lookup answers set-on, set-off, or could-not-tell. The third one is why this +module exists in the shape it does. A candidate `.env` file that is there and +cannot be read (wrong type, bad bytes, no permission) used to come back as +`None`, which every caller then read as "the flag is not set" -- a check that +could not run reporting clean. + +Now `read_flag_from_file` raises `UnreadableEnvFile` for that case, +`resolve_flag` collects those paths in `FlagLookup.unreadable`, and +`enforcement_gate` prints them to stderr before returning. The gate still +fails closed, because a quiet advisory is the safe direction for these hooks, +but the user is told which file could not be checked rather than being left to +read silence as consent. + +## Using it from a hook + +```python +sys.path.insert(0, os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_flags")) + +from flags import enforcement_gate # noqa: E402 + +if not enforcement_gate("my-hook", payload.get("cwd")): + return None +``` + +Two `dirname` calls, never `realpath`: install.sh links each hook directory +separately, so a hook finds this module as a sibling of its own symlink. +`realpath` would jump into the checkout and miss it in a partial install. + +install.sh links `_flags` into `~/.claude/hooks`, `~/.cursor/hooks` and +`~/.codex/hooks`, because `scope-lock` and `wrong-check-reflect` are installed +into all three and import this module at load time. A missing link is an +`ImportError` at hook start, which fails open and looks exactly like "the user +did not opt in" -- `tests/test_installed_layout.py` pins the link per harness +for that reason. + +## Tests + +```sh +python3 -m unittest discover -s engine/hooks/_flags/tests -v +``` diff --git a/engine/hooks/_flags/flags.py b/engine/hooks/_flags/flags.py new file mode 100644 index 00000000..db9e0570 --- /dev/null +++ b/engine/hooks/_flags/flags.py @@ -0,0 +1,229 @@ +#!/usr/bin/env python3 +"""One definition of "is this catstack flag on", shared by every flagged hook. + +A flag is off unless something turns it on. The value is looked up in this +order, and the first source that defines the key wins: + + 1. the process environment + 2. the file named by `$CATSTACK_ENV_FILE`, if that variable is set + 3. `/.env`, for the repo containing the hook's `cwd` + 4. `~/.catstack.env` + +Files are parsed as plain `KEY=VALUE` lines. They are never sourced, and no +key other than the one asked for is read back or printed. + +Three outcomes, not two. A lookup answers set-on, set-off, or could-not-tell, +and they are different answers: + + * `FlagLookup.value is None` -- nothing defined the key. Unset, not off. + * `FlagLookup.unreadable` -- a candidate file exists and could not be read + or decoded. The flag may well be set in there. A caller that treats this + as a clean "not set" is doing exactly what "a check that could not run is + not a pass" forbids, so `flag_on` returning False is never the whole + story: `unreadable` is non-empty and the caller has to say so. + +`flag_on` collapses the lookup to a bool for the common path and fails closed +-- an unreadable file leaves an opt-in flag off. That is the safe direction +for these hooks (an advisory that stays quiet), but it is a decision, not an +accident, and it is why the note has to reach the user some other way. + +Why a module and not a copy of the reader in each hook: cat-mode-default +grew this lookup first, and the reflect/automate-me hooks need exactly the +same one. Two readers over the same files drift apart silently -- one learns +about `~/.catstack.env` and the other does not, and the user who sets the +flag in one place finds half the hooks still running. + +Hooks are installed as sibling symlinks under $HOME/.claude/hooks/, so a +hook reaches this module by its own parent directory: + + sys.path.insert(0, os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_flags")) + +Two dirnames, not a "..": the hook's own directory is a symlink, so the OS +resolves it before applying "..", landing beside the checkout instead of +beside the other hooks. Stripping two segments textually cannot do that, and +abspath (never realpath) is what keeps the installed path in place. +""" +from __future__ import annotations + +import os +import re +import sys +from typing import NamedTuple + +ENV_FILE_VAR = "CATSTACK_ENV_FILE" +HOME_ENV_FILE = "~/.catstack.env" +TRUE_VALUES = frozenset({"1", "true", "yes", "on"}) +ENV_LINE_RE = re.compile(r"^\s*(?:export\s+)?([A-Za-z_][A-Za-z0-9_]*)\s*=\s*(.*?)\s*$") + +REFLECT_ENFORCEMENT = "CATSTACK_REFLECT_ENFORCEMENT" + + +class UnreadableEnvFile(Exception): + """A candidate .env file exists but could not be read or decoded.""" + + def __init__(self, path: str, reason: str) -> None: + super().__init__(f"{path}: {reason}") + self.path = path + self.reason = reason + + +class FlagLookup(NamedTuple): + value: str | None + source: str | None + unreadable: tuple[tuple[str, str], ...] = () + + @property + def on(self) -> bool: + return self.value is not None and self.value.strip().lower() in TRUE_VALUES + + def unreadable_note(self, key: str) -> str: + """One line naming the files that could not be checked, or "" when + every candidate was readable. Callers print this instead of letting an + unreadable file pass as "flag not set".""" + if not self.unreadable: + return "" + listed = "; ".join(f"{path} ({reason})" for path, reason in self.unreadable) + return ( + f"catstack: could not read {listed} while looking up {key}. " + f"Treating {key} as off -- set it in the environment to be sure." + ) + + +def repo_root(start: str | None) -> str | None: + """The nearest ancestor of `start` holding a `.git`, or None. + + None means "no repo to look in" -- either no cwd was supplied or the walk + reached the filesystem root. Both are ordinary: the caller drops the + `/.env` candidate and keeps the others. This is not a read + failure, which is what `UnreadableEnvFile` is for. + """ + if not start: + return None + current = os.path.abspath(start) + while True: + if os.path.exists(os.path.join(current, ".git")): + return current + parent = os.path.dirname(current) + if parent == current: + return None + current = parent + + +def env_file_candidates(environ: dict, cwd: str | None, home: str | None = None) -> list[str]: + candidates: list[str] = [] + explicit = environ.get(ENV_FILE_VAR) + if explicit: + candidates.append(os.path.expanduser(explicit)) + root = repo_root(cwd) + if root: + candidates.append(os.path.join(root, ".env")) + home_dir = home or environ.get("HOME") or os.path.expanduser("~") + candidates.append(os.path.join(home_dir, HOME_ENV_FILE.replace("~/", "", 1))) + return candidates + + +def _unquote(value: str) -> str: + if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'): + return value[1:-1] + return value + + +def read_flag_from_file(path: str, key: str) -> str | None: + """Return the value of `key` from a KEY=VALUE file. + + None means the file is not there, or is there and does not define the key + -- both are honest "not set here" answers. A file that exists and cannot + be read or decoded raises `UnreadableEnvFile` rather than returning None, + because "I could not look" and "I looked and it is absent" are different + answers and only the second one is a clean miss. + + A later line for the same key wins, as a shell would. Blank lines and + `#` comments define nothing and are skipped; that is the file format, not + a parse failure. + """ + try: + with open(path, encoding="utf-8") as handle: + lines = handle.read().splitlines() + except FileNotFoundError: + return None + except (OSError, UnicodeDecodeError) as exc: + raise UnreadableEnvFile(path, type(exc).__name__) from exc + found: str | None = None + for raw in lines: + stripped = raw.strip() + if not stripped or stripped.startswith("#"): + continue + match = ENV_LINE_RE.match(raw) + if match and match.group(1) == key: + found = _unquote(match.group(2)) + return found + + +def resolve_flag( + key: str, environ: dict, cwd: str | None, home: str | None = None +) -> FlagLookup: + """Look `key` up across every source. `source` is "env" or the path of the + file that defined it; `value` is None when nothing did. Files that exist + but could not be read land in `unreadable` and do not stop the walk -- a + later candidate may still define the key.""" + if key in environ: + return FlagLookup(environ[key], "env") + unreadable: list[tuple[str, str]] = [] + for path in env_file_candidates(environ, cwd, home): + try: + value = read_flag_from_file(path, key) + except UnreadableEnvFile as exc: + unreadable.append((exc.path, exc.reason)) + continue + if value is not None: + return FlagLookup(value, path, tuple(unreadable)) + return FlagLookup(None, None, tuple(unreadable)) + + +def flag_on(key: str, environ: dict, cwd: str | None, home: str | None = None) -> bool: + """True only when a source says so. Fails closed: an unreadable candidate + file leaves the flag off. Use `resolve_flag` when you need to tell the + user that a file could not be checked.""" + return resolve_flag(key, environ, cwd, home).on + + +def reflect_enforcement(environ: dict, cwd: str | None, home: str | None = None) -> FlagLookup: + """Full lookup for the reflect/automate-me flag, so a caller can report an + unreadable candidate file instead of silently staying quiet. + + The key is defined once, here beside the reader, because several hooks + answer to it and a key spelled from memory in one of them is a hook that + never turns on. Off unless a source sets it: these hooks push the user + into /reflect and automate-me, up to stopping every tool, so opting in is + the user's call. + """ + return resolve_flag(REFLECT_ENFORCEMENT, environ, cwd, home) + + +def reflect_enforcement_on(environ: dict, cwd: str | None, home: str | None = None) -> bool: + return reflect_enforcement(environ, cwd, home).on + + +def enforcement_gate( + hook_name: str, + cwd: str | None, + environ: dict | None = None, + stderr=None, +) -> bool: + """The one gate every reflect/automate-me hook calls. + + True only when the user opted in. A candidate .env file that exists and + could not be read is named on stderr first, because the honest answer + there is "I could not check", and a hook that just went quiet would be + reporting that as "the user did not opt in". + + `cwd` comes from the hook payload; it is what lets one repo turn the + class on through its own `.env`. None is fine -- the process environment + and `~/.catstack.env` are still consulted. + """ + found = reflect_enforcement(os.environ if environ is None else environ, cwd) + note = found.unreadable_note(REFLECT_ENFORCEMENT) + if note: + (stderr or sys.stderr).write(f"{hook_name}: {note}\n") + return found.on diff --git a/engine/hooks/_flags/tests/test_flags.py b/engine/hooks/_flags/tests/test_flags.py new file mode 100644 index 00000000..28d64abb --- /dev/null +++ b/engine/hooks/_flags/tests/test_flags.py @@ -0,0 +1,239 @@ +#!/usr/bin/env python3 +"""A flag lookup has three answers, and the third one is the point. + +set-on and set-off are easy. The answer this suite exists for is +could-not-tell: a candidate .env file that is there and cannot be read. The +old reader returned None for that, which every caller then read as "the flag +is not set" -- a check that could not run reporting clean. These tests pin +that such a file lands in `unreadable`, that the walk carries on past it, and +that `unreadable_note` gives the caller something to say. + +Run: python3 -m unittest discover -s engine/hooks/_flags/tests -v +""" +from __future__ import annotations + +import io +import os +import sys +import tempfile +import unittest + +FLAGS_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, FLAGS_DIR) + +import flags # noqa: E402 + +KEY = "CATSTACK_TEST_FLAG" + + +class Sandbox: + """A home directory, and a git repo to act as cwd, both throwaway.""" + + def __init__(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.home = os.path.join(self.tmp.name, "home") + self.repo = os.path.join(self.tmp.name, "repo") + os.makedirs(os.path.join(self.repo, ".git")) + self.cwd = os.path.join(self.repo, "nested", "deeper") + os.makedirs(self.cwd) + os.makedirs(self.home) + + def write(self, path: str, text: str) -> str: + with open(path, "w", encoding="utf-8") as handle: + handle.write(text) + return path + + @property + def home_env(self) -> str: + return os.path.join(self.home, ".catstack.env") + + @property + def repo_env(self) -> str: + return os.path.join(self.repo, ".env") + + def environ(self, extra: dict | None = None) -> dict: + env = {"HOME": self.home} + env.update(extra or {}) + return env + + def cleanup(self) -> None: + self.tmp.cleanup() + + +class FlagsTest(unittest.TestCase): + def setUp(self) -> None: + self.box = Sandbox() + self.addCleanup(self.box.cleanup) + + def resolve(self, environ: dict | None = None, cwd: str | None = None): + return flags.resolve_flag( + KEY, self.box.environ(environ), cwd if cwd is not None else self.box.cwd, self.box.home + ) + + + def test_nothing_defines_it(self): + found = self.resolve() + self.assertEqual((found.value, found.source, found.unreadable), (None, None, ())) + self.assertFalse(found.on) + + def test_unset_and_explicit_off_are_different_answers(self): + unset = self.resolve() + self.box.write(self.box.repo_env, f"{KEY}=0\n") + off = self.resolve() + self.assertIsNone(unset.value) + self.assertEqual(off.value, "0") + self.assertFalse(unset.on) + self.assertFalse(off.on) + + + def test_true_values(self): + for value in ("1", "true", "TRUE", "yes", "on", " on "): + with self.subTest(value=value): + self.assertTrue(flags.flag_on(KEY, self.box.environ({KEY: value}), self.box.cwd, self.box.home)) + + def test_false_values(self): + for value in ("0", "false", "off", "no", "", "maybe"): + with self.subTest(value=value): + self.assertFalse(flags.flag_on(KEY, self.box.environ({KEY: value}), self.box.cwd, self.box.home)) + + + def test_environment_beats_every_file(self): + self.box.write(self.box.repo_env, f"{KEY}=1\n") + self.box.write(self.box.home_env, f"{KEY}=1\n") + found = self.resolve({KEY: "0"}) + self.assertEqual((found.value, found.source), ("0", "env")) + + def test_named_env_file_beats_repo_and_home(self): + named = self.box.write(os.path.join(self.box.tmp.name, "named.env"), f"{KEY}=1\n") + self.box.write(self.box.repo_env, f"{KEY}=0\n") + self.box.write(self.box.home_env, f"{KEY}=0\n") + found = self.resolve({flags.ENV_FILE_VAR: named}) + self.assertEqual((found.value, found.source), ("1", named)) + + def test_repo_env_beats_home_and_is_found_by_walking_up(self): + self.box.write(self.box.repo_env, f"{KEY}=1\n") + self.box.write(self.box.home_env, f"{KEY}=0\n") + found = self.resolve() + self.assertEqual((found.value, found.source), ("1", self.box.repo_env)) + + def test_home_env_is_the_last_resort(self): + self.box.write(self.box.home_env, f"{KEY}=1\n") + found = self.resolve() + self.assertEqual((found.value, found.source), ("1", self.box.home_env)) + + def test_no_cwd_drops_only_the_repo_candidate(self): + self.box.write(self.box.home_env, f"{KEY}=1\n") + found = self.resolve(cwd=None) + self.assertEqual((found.value, found.source), ("1", self.box.home_env)) + self.assertNotIn(self.box.repo_env, flags.env_file_candidates(self.box.environ(), None, self.box.home)) + + + def test_later_line_wins_and_comments_are_skipped(self): + self.box.write(self.box.repo_env, f"# {KEY}=1\n{KEY}=0\n\n{KEY}=1\n") + self.assertEqual(self.resolve().value, "1") + + def test_quotes_and_export_prefix(self): + self.box.write(self.box.repo_env, f'export {KEY}="1"\n') + self.assertEqual(self.resolve().value, "1") + + def test_other_keys_are_not_read_back(self): + self.box.write(self.box.repo_env, f"SECRET=hunter2\n{KEY}=1\n") + found = self.resolve() + self.assertEqual(found.value, "1") + self.assertNotIn("hunter2", repr(found)) + + + def test_missing_file_is_a_clean_miss_not_an_unreadable(self): + found = self.resolve() + self.assertEqual(found.unreadable, ()) + self.assertEqual(found.unreadable_note(KEY), "") + + def test_a_directory_in_place_of_the_file_is_reported_not_swallowed(self): + os.makedirs(self.box.repo_env) + found = self.resolve() + self.assertEqual([path for path, _ in found.unreadable], [self.box.repo_env]) + self.assertFalse(found.on) + self.assertIn(self.box.repo_env, found.unreadable_note(KEY)) + self.assertIn(KEY, found.unreadable_note(KEY)) + + def test_undecodable_file_is_reported_not_swallowed(self): + with open(self.box.repo_env, "wb") as handle: + handle.write(b"\xff\xfe\x00\x00" + KEY.encode() + b"=1\n") + found = self.resolve() + self.assertEqual([reason for _, reason in found.unreadable], ["UnicodeDecodeError"]) + self.assertFalse(found.on) + + def test_an_unreadable_candidate_does_not_stop_the_walk(self): + os.makedirs(self.box.repo_env) + self.box.write(self.box.home_env, f"{KEY}=1\n") + found = self.resolve() + self.assertEqual((found.value, found.source), ("1", self.box.home_env)) + self.assertTrue(found.on) + self.assertEqual([path for path, _ in found.unreadable], [self.box.repo_env]) + + def test_read_flag_from_file_raises_rather_than_returning_none(self): + os.makedirs(self.box.repo_env) + with self.assertRaises(flags.UnreadableEnvFile) as caught: + flags.read_flag_from_file(self.box.repo_env, KEY) + self.assertEqual(caught.exception.path, self.box.repo_env) + + + def test_reflect_enforcement_key_and_default(self): + self.assertEqual(flags.REFLECT_ENFORCEMENT, "CATSTACK_REFLECT_ENFORCEMENT") + self.assertFalse(flags.reflect_enforcement_on(self.box.environ(), self.box.cwd, self.box.home)) + + def test_reflect_enforcement_reads_the_same_sources(self): + self.box.write(self.box.repo_env, f"{flags.REFLECT_ENFORCEMENT}=1\n") + self.assertTrue(flags.reflect_enforcement_on(self.box.environ(), self.box.cwd, self.box.home)) + + +class EnforcementGateTest(unittest.TestCase): + """The one gate the reflect/automate-me hooks call. + + Its job beyond on/off is the third outcome: when a candidate .env file + cannot be read, the gate says so by name before returning False, so a + silent hook is never the only evidence the user gets. + """ + + def setUp(self) -> None: + self.box = Sandbox() + self.addCleanup(self.box.cleanup) + self.err = io.StringIO() + + def gate(self, environ: dict | None = None) -> bool: + return flags.enforcement_gate( + "test-hook", self.box.cwd, self.box.environ(environ), self.err + ) + + def test_off_and_quiet_when_nothing_is_set(self): + self.assertFalse(self.gate()) + self.assertEqual(self.err.getvalue(), "") + + def test_on_when_the_environment_says_so(self): + self.assertTrue(self.gate({flags.REFLECT_ENFORCEMENT: "1"})) + self.assertEqual(self.err.getvalue(), "") + + def test_an_unreadable_env_file_is_named_not_swallowed(self): + os.makedirs(self.box.repo_env) + self.assertFalse(self.gate()) + note = self.err.getvalue() + self.assertTrue(note.startswith("test-hook: "), note) + self.assertIn(self.box.repo_env, note) + self.assertIn(flags.REFLECT_ENFORCEMENT, note) + self.assertTrue(note.endswith("\n"), repr(note)) + + def test_an_unreadable_file_is_still_reported_when_the_flag_is_on(self): + os.makedirs(self.box.repo_env) + self.box.write(self.box.home_env, f"{flags.REFLECT_ENFORCEMENT}=1\n") + self.assertTrue(self.gate()) + self.assertIn(self.box.repo_env, self.err.getvalue()) + + def test_no_cwd_is_not_a_failure(self): + self.assertFalse( + flags.enforcement_gate("test-hook", None, self.box.environ(), self.err) + ) + self.assertEqual(self.err.getvalue(), "") + + +if __name__ == "__main__": + unittest.main() diff --git a/engine/hooks/_flags/tests/test_installed_layout.py b/engine/hooks/_flags/tests/test_installed_layout.py new file mode 100644 index 00000000..0de8a58e --- /dev/null +++ b/engine/hooks/_flags/tests/test_installed_layout.py @@ -0,0 +1,161 @@ +#!/usr/bin/env python3 +"""The sibling-path import has to resolve where hooks actually run. + +install.sh links each hook directory separately into the harness hooks folder, +so a hook reaches this module through its own parent directory. That only +works because the path is built with abspath, which leaves the symlink in +place; realpath would jump into the checkout and find nothing beside it in a +partial install. This test builds that layout for real rather than trusting +the shape. + +Three harnesses, not one. scope-lock and wrong-check-reflect are linked into +~/.cursor/hooks and ~/.codex/hooks as well, and both now import this module at +load time. A `_flags` link that reaches only Claude would leave those two dead +on the other two harnesses -- and dead means the flag can never turn them on +there, which reads exactly like "the user did not opt in". + +Run: python3 -m unittest discover -s engine/hooks/_flags/tests -v +""" +import os +import re +import shutil +import subprocess +import sys +import tempfile +import unittest + +FLAGS_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +HOOKS_DIR = os.path.dirname(FLAGS_DIR) +REPO_DIR = os.path.dirname(os.path.dirname(HOOKS_DIR)) +INSTALL_SH = os.path.join(REPO_DIR, "install.sh") + +CONSUMER = ( + "import os, sys\n" + "sys.path.insert(0, os.path.join(" + "os.path.dirname(os.path.dirname(os.path.abspath(__file__))), '_flags'))\n" + "from flags import enforcement_gate, REFLECT_ENFORCEMENT\n" + "print(enforcement_gate('consumer', None))\n" +) + +HARNESS_DIRS = ( + ("claude", os.path.join(".claude", "hooks")), + ("cursor", os.path.join(".cursor", "hooks")), + ("codex", os.path.join(".codex", "hooks")), +) + + +class InstalledLayout(unittest.TestCase): + def setUp(self): + self.home = tempfile.mkdtemp(prefix="flags-install-") + self.addCleanup(shutil.rmtree, self.home, ignore_errors=True) + self.installed_hooks = os.path.join(self.home, ".claude", "hooks") + os.makedirs(self.installed_hooks) + + def _link(self, name, src): + target = os.path.join(self.installed_hooks, name) + os.symlink(src, target) + return target + + def _run_consumer(self, consumer_dir, env=None): + script = os.path.join(consumer_dir, "detect.py") + with open(script, "w", encoding="utf-8") as handle: + handle.write(CONSUMER) + environ = dict(os.environ) + environ["HOME"] = self.home + environ.pop("CATSTACK_REFLECT_ENFORCEMENT", None) + environ.pop("CATSTACK_ENV_FILE", None) + environ.update(env or {}) + return subprocess.run( + [sys.executable, os.path.join(self.installed_hooks, "consumer", "detect.py")], + capture_output=True, + text=True, + env=environ, + ) + + def _consumer(self): + real = tempfile.mkdtemp(prefix="flags-consumer-") + self.addCleanup(shutil.rmtree, real, ignore_errors=True) + self._link("consumer", real) + return real + + def test_consumer_imports_flags_through_symlinked_siblings(self): + self._link("_flags", FLAGS_DIR) + result = self._run_consumer(self._consumer()) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(result.stdout.strip(), "False") + + def test_the_flag_reaches_the_consumer_in_the_installed_layout(self): + self._link("_flags", FLAGS_DIR) + result = self._run_consumer( + self._consumer(), env={"CATSTACK_REFLECT_ENFORCEMENT": "1"} + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(result.stdout.strip(), "True") + + def test_missing_flags_link_fails_loudly_instead_of_passing_clean(self): + result = self._run_consumer(self._consumer()) + self.assertNotEqual(result.returncode, 0) + self.assertIn("ModuleNotFoundError", result.stderr) + + +class InstallerWiring(unittest.TestCase): + """A hook that imports _flags is dead wherever _flags was not linked. + + The link is what makes the flag reachable, so it is asserted per harness + rather than once -- a single `link_item "_flags"` anywhere in the file + would satisfy a looser check while two harnesses stayed broken. + """ + + def setUp(self): + with open(INSTALL_SH, encoding="utf-8") as handle: + self.body = handle.read() + + def test_install_sh_links_flags_into_every_harness_hooks_dir(self): + for harness, relative in HARNESS_DIRS: + with self.subTest(harness=harness): + expected = '"$HOME/{}/_flags"'.format(relative.replace(os.sep, "/")) + self.assertIn(expected, self.body) + + def test_every_hook_importing_flags_is_linked_where_it_is_installed(self): + """Names the gap rather than trusting the list above to stay current.""" + importers = set() + for entry in sorted(os.listdir(HOOKS_DIR)): + hook_dir = os.path.join(HOOKS_DIR, entry) + if not os.path.isdir(hook_dir) or entry.startswith("_"): + continue + for name in os.listdir(hook_dir): + if not name.endswith(".py"): + continue + with open(os.path.join(hook_dir, name), encoding="utf-8") as handle: + if "from flags import" in handle.read(): + importers.add(entry) + break + self.assertTrue(importers, "no hook imports flags; this test is now vacuous") + checked = 0 + for hook in sorted(importers): + for harness, relative in HARNESS_DIRS: + folder = relative.replace(os.sep, "/") + installed = re.search( + r'link_item "{}" "\$REPO_DIR/engine/hooks/{}" "\$HOME/{}/{}"'.format( + re.escape(hook), re.escape(hook), re.escape(folder), re.escape(hook) + ), + self.body, + ) + if not installed: + continue + checked += 1 + with self.subTest(hook=hook, harness=harness): + self.assertIn( + '"$HOME/{}/_flags"'.format(folder), + self.body, + f"{hook} is installed into {folder} but _flags is not", + ) + self.assertTrue( + checked, + "matched no link_item line for any flags-importing hook -- the regex " + "no longer matches install.sh, so this test checked nothing", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/engine/hooks/reflect-on-thrash/README.md b/engine/hooks/reflect-on-thrash/README.md index 6e855ca0..1547996b 100644 --- a/engine/hooks/reflect-on-thrash/README.md +++ b/engine/hooks/reflect-on-thrash/README.md @@ -31,3 +31,18 @@ re-read. One "I told you" defers; the same class twice forces the prompt. ```sh python3 -m unittest discover -s hooks/reflect-on-thrash/tests -v ``` + +## Off unless you opt in + +This hook is part of the reflect/automate-me class and does nothing unless +`CATSTACK_REFLECT_ENFORCEMENT` is on: + +```sh +echo 'CATSTACK_REFLECT_ENFORCEMENT=1' >> ~/.catstack.env +``` + +The environment, `$CATSTACK_ENV_FILE`, the repo's `.env` and `~/.catstack.env` +are all consulted, in that order. See `engine/hooks/_flags/README.md`. + +The gate sits ahead of the deferred marker, so a disabled session leaves +nothing behind for a later enabled session to deliver. diff --git a/engine/hooks/reflect-on-thrash/detect.py b/engine/hooks/reflect-on-thrash/detect.py index 0beabc99..e2b61bed 100644 --- a/engine/hooks/reflect-on-thrash/detect.py +++ b/engine/hooks/reflect-on-thrash/detect.py @@ -2,6 +2,10 @@ Uses engine/skills/reflect/scripts/token_audit.py as the source of truth. Fail-open: any parse/IO/import error means "no hit" so a broken hook never bricks a session. + +Off unless `CATSTACK_REFLECT_ENFORCEMENT` is on -- see engine/hooks/_flags. +The gate sits in `decide`, ahead of the deferred marker, so a disabled hook +leaves no marker for a later session to deliver. """ from __future__ import annotations @@ -19,6 +23,11 @@ REPO_DIR = os.path.dirname(os.path.dirname(os.path.dirname(HERE))) TOKEN_AUDIT_DIR = os.path.join(REPO_DIR, "engine", "skills", "reflect", "scripts") +sys.path.insert(0, os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_flags")) + +from flags import enforcement_gate # noqa: E402 + STATE_DIR = os.environ.get( "REFLECT_ON_THRASH_STATE_DIR", os.path.join(os.path.expanduser("~"), ".cache", "catstack-reflect-on-thrash"), @@ -328,6 +337,8 @@ def decide( is the exception: deliver immediately (Claude Stop exit 2 / Cursor followup) — do not wait for session end or for the user to re-prompt. """ + if not enforcement_gate("reflect-on-thrash", payload.get("cwd")): + return None if payload.get("stop_hook_active"): return None path = resolve_transcript(payload) diff --git a/engine/hooks/reflect-on-thrash/tests/test_hooks.py b/engine/hooks/reflect-on-thrash/tests/test_hooks.py index 06c383a7..97f1e28e 100644 --- a/engine/hooks/reflect-on-thrash/tests/test_hooks.py +++ b/engine/hooks/reflect-on-thrash/tests/test_hooks.py @@ -25,6 +25,29 @@ import cursor_session # noqa: E402 import detect # noqa: E402 +sys.path.insert(0, os.path.join(os.path.dirname(HOOKS_DIR), "_flags")) +import flags # noqa: E402 + + +_ENFORCEMENT = None + + +def setUpModule() -> None: + """Every test below is about what this hook does when it is switched on. + + The hook is off unless CATSTACK_REFLECT_ENFORCEMENT says so, so the suite + opts in the way a user would. TestEnforcementFlag clears the environment + again to pin the off path. + """ + global _ENFORCEMENT + _ENFORCEMENT = patch.dict(os.environ, {flags.REFLECT_ENFORCEMENT: "1"}) + _ENFORCEMENT.start() + + +def tearDownModule() -> None: + _ENFORCEMENT.stop() + + def fixture(name: str) -> str: return os.path.join(FIXTURES, name) @@ -400,3 +423,42 @@ def test_missing_agent_transcript_does_not_fall_back_to_the_parent(self): detect.resolve_transcript({"transcript_path": parent, "agent_transcript_path": gone}), "", ) + + +class TestEnforcementFlag(unittest.TestCase): + """reflect-on-thrash is off unless the user opts in. + + The gate sits ahead of the deferred marker on purpose. A disabled hook + that still recorded "this session thrashed" would hand the prompt to + whichever later session happened to have the flag on, for work the user + had opted out of watching. + """ + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + detect.STATE_DIR = self.tmp.name + self.path = fixture("token_thrash_session.jsonl") + + def env(self, value=None): + env = {"HOME": self.tmp.name, "REFLECT_ON_THRASH_STATE_DIR": self.tmp.name} + if value is not None: + env[flags.REFLECT_ENFORCEMENT] = value + return patch.dict(os.environ, env, clear=True) + + def test_unset_flag_says_nothing_and_defers_nothing(self): + with self.env(): + self.assertIsNone(detect.decide({"transcript_path": self.path}, deliver=False)) + self.assertFalse(detect.has_deferred(self.path)) + self.assertIsNone(detect.decide({"transcript_path": self.path}, deliver=True)) + + def test_explicit_off_says_nothing(self): + for value in ("0", "false", "off", "no"): + with self.subTest(value=value), self.env(value): + self.assertIsNone(detect.decide({"transcript_path": self.path}, deliver=True)) + + def test_flag_on_still_delivers(self): + with self.env("1"): + message = detect.decide({"transcript_path": self.path}, deliver=False) + self.assertIsNotNone(message) + self.assertIn("automate-me", message) diff --git a/engine/hooks/scope-lock/README.md b/engine/hooks/scope-lock/README.md index b4cc0fc5..483be112 100644 --- a/engine/hooks/scope-lock/README.md +++ b/engine/hooks/scope-lock/README.md @@ -73,3 +73,19 @@ trusting new or changed definitions through `/hooks` before they run. python3 -m unittest discover -s engine/hooks/scope-lock/tests -v python3 scripts/check_hook_test_coverage.py engine/hooks/scope-lock ``` + +## Off unless you opt in + +This hook is part of the reflect/automate-me class and does nothing unless +`CATSTACK_REFLECT_ENFORCEMENT` is on: + +```sh +echo 'CATSTACK_REFLECT_ENFORCEMENT=1' >> ~/.catstack.env +``` + +The environment, `$CATSTACK_ENV_FILE`, the repo's `.env` and `~/.catstack.env` +are all consulted, in that order. See `engine/hooks/_flags/README.md`. + +The gate covers the recorder as well as the stop. A disabled hook that still +counted corrections would hard stop the first tool call after the flag was +turned on, using corrections from a session the user had opted out of. diff --git a/engine/hooks/scope-lock/detect.py b/engine/hooks/scope-lock/detect.py index 92311128..0d0a340d 100644 --- a/engine/hooks/scope-lock/detect.py +++ b/engine/hooks/scope-lock/detect.py @@ -5,6 +5,12 @@ stops every tool until the user explicitly invokes both /reflect and automate-me, in either order, in one message or across several. State is keyed to the harness session, not the repository. + +Off unless `CATSTACK_REFLECT_ENFORCEMENT` is on -- see engine/hooks/_flags. +Stopping every tool is the strongest thing this repo does to its own user, so +it is opt-in. The flag gates the recorder as well as the gate: counting +corrections while disabled would hard stop the first tool call after the flag +was turned on, using corrections from a session that was opted out. """ from __future__ import annotations @@ -12,9 +18,15 @@ import json import os import re +import sys import tempfile from typing import Any +sys.path.insert(0, os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_flags")) + +from flags import enforcement_gate # noqa: E402 + STATE_DIR = os.environ.get( "SCOPE_LOCK_STATE_DIR", os.path.join(os.path.expanduser("~"), ".cache", "catstack-scope-lock"), @@ -141,6 +153,17 @@ on.""" +def enforcement_on(payload: dict[str, Any]) -> bool: + """Whether reflect/automate-me enforcement is switched on for this payload. + + The payload's `cwd` is what lets a single repo turn the hook on through its + own `.env`; without one, only the process environment and `~/.catstack.env` + are consulted. An `.env` file that exists and cannot be read leaves the + flag off and is named on stderr rather than passing as "not set". + """ + return enforcement_gate("scope-lock", payload.get("cwd")) + + def extract_prompt_text(payload: dict[str, Any]) -> str: for key in ("prompt", "user_prompt", "userPrompt", "message"): value = payload.get(key) @@ -296,6 +319,8 @@ def process_prompt(payload: dict[str, Any]) -> dict[str, Any]: means the next correction gets the first-stage contract again, not an instant stop, and a new hold starts with nothing recorded toward ending it. """ + if not enforcement_on(payload): + return {} prompt = extract_prompt_text(payload) state = load_state(payload) if not state_path(payload): @@ -373,6 +398,8 @@ def _tool_name(payload: dict[str, Any]) -> str: def tool_block_reason(payload: dict[str, Any]) -> tuple[bool, str]: """Return whether this tool is blocked and the deterministic reason.""" + if not enforcement_on(payload): + return False, "" state = load_state(payload) phase = state.get("phase") if phase == "hard_stop": diff --git a/engine/hooks/scope-lock/tests/test_hooks.py b/engine/hooks/scope-lock/tests/test_hooks.py index 2d4fcfdb..c17f97c1 100644 --- a/engine/hooks/scope-lock/tests/test_hooks.py +++ b/engine/hooks/scope-lock/tests/test_hooks.py @@ -14,6 +14,7 @@ HOOK_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) FIXTURE_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "fixtures") sys.path.insert(0, HOOK_DIR) +sys.path.insert(0, os.path.join(os.path.dirname(HOOK_DIR), "_flags")) import claude_pretool_scope # noqa: E402 import claude_prompt_scope # noqa: E402 @@ -22,6 +23,7 @@ import cursor_before_submit # noqa: E402 import cursor_pretool_scope # noqa: E402 import detect # noqa: E402 +import flags # noqa: E402 import install_claude_hook # noqa: E402 import install_codex_hook # noqa: E402 import install_cursor_hook # noqa: E402 @@ -51,6 +53,23 @@ def append_assistant(path: str, text: str) -> None: }) + "\n") +def enable_enforcement(case: unittest.TestCase) -> None: + """Turn the opt-in flag on for a case that is testing the lock itself. + + scope-lock does nothing unless CATSTACK_REFLECT_ENFORCEMENT is on, so + every test about the lock's behaviour has to opt in the way a user would. + `clear=True` keeps a real ~/.catstack.env or repo .env on the developer's + machine from deciding what these tests see. + """ + patcher = patch.dict( + os.environ, + {"HOME": case.tmp.name, flags.REFLECT_ENFORCEMENT: "1"}, + clear=True, + ) + patcher.start() + case.addCleanup(patcher.stop) + + def run_main(main, payload: dict) -> tuple[int, str, str]: out = io.StringIO() err = io.StringIO() @@ -77,6 +96,7 @@ def setUp(self) -> None: self.tmp = tempfile.TemporaryDirectory() self.saved_state_dir = detect.STATE_DIR detect.STATE_DIR = self.tmp.name + enable_enforcement(self) def tearDown(self) -> None: detect.STATE_DIR = self.saved_state_dir @@ -107,6 +127,7 @@ class ScopeLockCase(unittest.TestCase): def setUp(self) -> None: self.tmp = tempfile.TemporaryDirectory() detect.STATE_DIR = self.tmp.name + enable_enforcement(self) self.transcript = os.path.join(self.tmp.name, "session.jsonl") open(self.transcript, "w", encoding="utf-8").close() self.base = {"session_id": "session-1", "transcript_path": self.transcript} @@ -575,5 +596,65 @@ def test_codex_installer_migrates_legacy_pretool_without_dropping_it(self): self.assertEqual(merged["hooks"]["PreToolUse"][0]["matcher"], "Bash") +class TestEnforcementFlag(ScopeLockCase): + """scope-lock is off unless the user opts in. + + This hook stops every tool until the user types two specific phrases. That + is the most intrusive thing in the repo, so it does not run on a machine + that never asked for it. `CATSTACK_REFLECT_ENFORCEMENT` is the one switch + for the whole reflect/automate-me class, and unset means off. + + Off means off all the way down: no state is recorded either. A hook that + quietly counted corrections while disabled would slam a hard stop onto the + first tool call after the flag was turned on, using corrections from a + session the user had opted out of. + """ + + def drive_two_corrections(self) -> dict: + self.prompt("wtf are you doing? Just fix it locally.") + append_assistant(self.transcript, "SCOPE CONTRACT: Fix only local catstack support.") + return self.prompt( + "why did you expand into babysitting the merge queue? All I am asking is catstack support." + ) + + def test_unset_flag_blocks_nothing_and_records_nothing(self): + with patch.dict(os.environ, {"HOME": self.tmp.name}, clear=True): + state = self.drive_two_corrections() + self.assertNotEqual(state.get("phase"), "hard_stop") + self.assertEqual(detect.prompt_instruction(state), "") + for tool in ("Read", "Bash", "Write"): + self.assertEqual(self.tool(tool), (False, ""), tool) + + def test_explicit_off_blocks_nothing(self): + for value in ("0", "false", "off", "no"): + with self.subTest(value=value): + env = {"HOME": self.tmp.name, flags.REFLECT_ENFORCEMENT: value} + with patch.dict(os.environ, env, clear=True): + state = self.drive_two_corrections() + self.assertNotEqual(state.get("phase"), "hard_stop") + self.assertFalse(self.tool("Write")[0]) + + def test_flag_on_still_hard_stops(self): + env = {"HOME": self.tmp.name, flags.REFLECT_ENFORCEMENT: "1"} + with patch.dict(os.environ, env, clear=True): + state = self.drive_two_corrections() + self.assertEqual(state["phase"], "hard_stop") + blocked, reason = self.tool("Write") + self.assertTrue(blocked) + self.assertIn("/reflect", reason) + self.assertIn("automate-me", reason) + + def test_a_dot_env_file_in_the_repo_turns_it_on(self): + repo = os.path.join(self.tmp.name, "repo") + os.makedirs(os.path.join(repo, ".git")) + with open(os.path.join(repo, ".env"), "w", encoding="utf-8") as handle: + handle.write(f"{flags.REFLECT_ENFORCEMENT}=1\n") + self.base = {**self.base, "cwd": repo} + with patch.dict(os.environ, {"HOME": self.tmp.name}, clear=True): + state = self.drive_two_corrections() + self.assertEqual(state["phase"], "hard_stop") + self.assertTrue(self.tool("Write")[0]) + + if __name__ == "__main__": unittest.main() diff --git a/engine/hooks/verdict-flip-watch/README.md b/engine/hooks/verdict-flip-watch/README.md index da3b2366..8adc012b 100644 --- a/engine/hooks/verdict-flip-watch/README.md +++ b/engine/hooks/verdict-flip-watch/README.md @@ -58,3 +58,18 @@ per transcript per target. Fail-open on any parse or IO error. python3 -m unittest discover -s engine/hooks/verdict-flip-watch/tests -v python3 scripts/check_hook_test_coverage.py engine/hooks/verdict-flip-watch ``` + +## Off unless you opt in + +This hook is part of the reflect/automate-me class and does nothing unless +`CATSTACK_REFLECT_ENFORCEMENT` is on: + +```sh +echo 'CATSTACK_REFLECT_ENFORCEMENT=1' >> ~/.catstack.env +``` + +The environment, `$CATSTACK_ENV_FILE`, the repo's `.env` and `~/.catstack.env` +are all consulted, in that order. See `engine/hooks/_flags/README.md`. + +It belongs to this class because its message ends in "the admission is a +reflect trigger". diff --git a/engine/hooks/verdict-flip-watch/detect.py b/engine/hooks/verdict-flip-watch/detect.py index 8ad2180a..be68c05c 100644 --- a/engine/hooks/verdict-flip-watch/detect.py +++ b/engine/hooks/verdict-flip-watch/detect.py @@ -33,6 +33,12 @@ import json import os import re +import sys + +sys.path.insert(0, os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_flags")) + +from flags import enforcement_gate # noqa: E402 STATE_DIR = os.environ.get( "VERDICT_FLIP_WATCH_STATE_DIR", @@ -186,6 +192,8 @@ def mark_noted(transcript_path: str, target: str) -> None: def decide(payload: dict) -> str | None: + if not enforcement_gate("verdict-flip-watch", payload.get("cwd")): + return None if payload.get("stop_hook_active"): return None message = payload.get("last_assistant_message") or "" diff --git a/engine/hooks/verdict-flip-watch/tests/test_hooks.py b/engine/hooks/verdict-flip-watch/tests/test_hooks.py index 0a337958..1046509f 100644 --- a/engine/hooks/verdict-flip-watch/tests/test_hooks.py +++ b/engine/hooks/verdict-flip-watch/tests/test_hooks.py @@ -14,12 +14,36 @@ import sys import tempfile import unittest +from unittest.mock import patch HERE = os.path.dirname(os.path.abspath(__file__)) HOOK_DIR = os.path.dirname(HERE) sys.path.insert(0, HOOK_DIR) import detect # noqa: E402 +sys.path.insert(0, os.path.join(os.path.dirname(HOOK_DIR), "_flags")) +import flags # noqa: E402 + + +_ENFORCEMENT = None + + +def setUpModule() -> None: + """Every test below is about what this hook does when it is switched on. + + The hook is off unless CATSTACK_REFLECT_ENFORCEMENT says so, so the suite + opts in the way a user would. TestEnforcementFlag clears the environment + again to pin the off path. + """ + global _ENFORCEMENT + _ENFORCEMENT = patch.dict(os.environ, {flags.REFLECT_ENFORCEMENT: "1"}) + _ENFORCEMENT.start() + + +def tearDownModule() -> None: + _ENFORCEMENT.stop() + + FIXTURES = os.path.join(HERE, "fixtures") CLEAN_REPLY = "Opened the PR; here is the link." @@ -143,5 +167,32 @@ def test_malformed_stdin_fails_open(self): self.assertEqual(res.returncode, 0) +class TestEnforcementFlag(IsolatedState): + """verdict-flip-watch is off unless the user opts in. + + It ends in "the admission is a reflect trigger", so it belongs to the + reflect/automate-me class and answers to the same single switch. + """ + + def env(self, value=None): + env = {"HOME": self._tmp} + if value is not None: + env[flags.REFLECT_ENFORCEMENT] = value + return patch.dict(os.environ, env, clear=True) + + def test_unset_flag_says_nothing_about_a_real_flip(self): + with self.env(): + self.assertIsNone(detect.decide(payload("flip"))) + + def test_explicit_off_says_nothing(self): + for value in ("0", "false", "off", "no"): + with self.subTest(value=value), self.env(value): + self.assertIsNone(detect.decide(payload("flip"))) + + def test_flag_on_still_notes_the_flip(self): + with self.env("1"): + self.assertIsNotNone(detect.decide(payload("flip"))) + + if __name__ == "__main__": unittest.main() diff --git a/engine/hooks/wrong-check-reflect/README.md b/engine/hooks/wrong-check-reflect/README.md index 705f0aa3..72d6aeb1 100644 --- a/engine/hooks/wrong-check-reflect/README.md +++ b/engine/hooks/wrong-check-reflect/README.md @@ -66,3 +66,18 @@ pattern to this hook; the prose meaning belongs in the phrase dictionary. python3 -m unittest discover -s engine/hooks/wrong-check-reflect/tests -v python3 scripts/check_hook_test_coverage.py engine/hooks/wrong-check-reflect ``` + +## Off unless you opt in + +This hook is part of the reflect/automate-me class and does nothing unless +`CATSTACK_REFLECT_ENFORCEMENT` is on: + +```sh +echo 'CATSTACK_REFLECT_ENFORCEMENT=1' >> ~/.catstack.env +``` + +The environment, `$CATSTACK_ENV_FILE`, the repo's `.env` and `~/.catstack.env` +are all consulted, in that order. See `engine/hooks/_flags/README.md`. + +The gate sits inside `enqueue_judge`, so it covers the Claude Stop hook, the +Codex notify and the Cursor session hook with one check. diff --git a/engine/hooks/wrong-check-reflect/detect.py b/engine/hooks/wrong-check-reflect/detect.py index fd2b7d2c..d05b5410 100644 --- a/engine/hooks/wrong-check-reflect/detect.py +++ b/engine/hooks/wrong-check-reflect/detect.py @@ -6,8 +6,14 @@ import json import os import re +import sys import uuid +sys.path.insert(0, os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_flags")) + +from flags import enforcement_gate # noqa: E402 + HOOKS_DIR = os.path.dirname(os.path.abspath(__file__)) LLM_JUDGE_DIR = os.path.join(os.path.dirname(HOOKS_DIR), "llm-judge") LLM_JUDGE_PATH = os.path.join(LLM_JUDGE_DIR, "judge.py") @@ -182,6 +188,8 @@ def _phrases(): def enqueue_judge(payload: dict) -> str | None: if not isinstance(payload, dict) or payload.get("stop_hook_active"): return None + if not enforcement_gate("wrong-check-reflect", payload.get("cwd")): + return None path = resolve_transcript(payload) text = last_assistant_text(payload, path) key = path or text[:200] diff --git a/engine/hooks/wrong-check-reflect/tests/test_hooks.py b/engine/hooks/wrong-check-reflect/tests/test_hooks.py index 4165be66..c897d515 100644 --- a/engine/hooks/wrong-check-reflect/tests/test_hooks.py +++ b/engine/hooks/wrong-check-reflect/tests/test_hooks.py @@ -21,6 +21,9 @@ import cursor_session # noqa: E402 import detect # noqa: E402 +sys.path.insert(0, os.path.join(os.path.dirname(HOOKS_DIR), "_flags")) +import flags # noqa: E402 + sys.path.append(os.path.dirname(detect.LLM_JUDGE_PATH)) import inbox as judge_inbox # noqa: E402 import judge # noqa: E402 @@ -76,6 +79,7 @@ def setUp(self): self.reflect_state = tempfile.TemporaryDirectory() self.judge_state = tempfile.TemporaryDirectory() self.env = patch.dict(os.environ, { + flags.REFLECT_ENFORCEMENT: "1", "WRONG_CHECK_REFLECT_STATE_DIR": self.reflect_state.name, judge.STATE_ENV: self.judge_state.name, judge.RUNNERS_ENV: json.dumps([ANSWERS_HIT]), @@ -137,6 +141,22 @@ def test_dictionary_loads(self): def test_decide_no_longer_returns_pattern_hit(self): self.assertIsNone(detect.decide({"last_assistant_message": HIT_TEXT})) + def test_enqueue_is_off_unless_the_flag_is_on(self): + """No job is queued, and none is silently deferred either. + + The gate sits inside enqueue_judge rather than in the harness wrapper + so every entry point -- Claude Stop, the Codex notify, the Cursor + session hook -- is covered by the one check. + """ + path = self.write_transcript(("assistant", HIT_TEXT)) + for value in (None, "0", "false", "off", "no"): + env = {"HOME": self.reflect_state.name} + if value is not None: + env[flags.REFLECT_ENFORCEMENT] = value + with self.subTest(value=value), patch.dict(os.environ, env, clear=True): + self.assertIsNone(detect.enqueue_judge({"transcript_path": path})) + self.assertEqual(self.jobs(), []) + def test_claude_stop_queues_job_for_normal_reply(self): os.environ[judge.RUNNERS_ENV] = json.dumps([SLOW_CLEAN]) path = self.write_transcript(("assistant", HIT_TEXT)) diff --git a/install.sh b/install.sh index dfdd3347..434414f4 100755 --- a/install.sh +++ b/install.sh @@ -217,6 +217,7 @@ fi echo "--- claude hooks (\$HOME/.claude/hooks) ---" mkdir -p "$HOME/.claude/hooks" link_item "_markers" "$REPO_DIR/engine/hooks/_markers" "$HOME/.claude/hooks/_markers" +link_item "_flags" "$REPO_DIR/engine/hooks/_flags" "$HOME/.claude/hooks/_flags" link_item "diu-stop" "$REPO_DIR/engine/hooks/diu-stop" "$HOME/.claude/hooks/diu-stop" link_item "bug-complaint-leak" "$REPO_DIR/engine/hooks/bug-complaint-leak" "$HOME/.claude/hooks/bug-complaint-leak" link_item "demo-freeze" "$REPO_DIR/engine/hooks/demo-freeze" "$HOME/.claude/hooks/demo-freeze" @@ -266,6 +267,7 @@ bash "$REPO_DIR/scripts/install-git-template.sh" echo "--- cursor hooks dir (\$HOME/.cursor/hooks) ---" mkdir -p "$HOME/.cursor/hooks" +link_item "_flags" "$REPO_DIR/engine/hooks/_flags" "$HOME/.cursor/hooks/_flags" link_item "bug-complaint-leak" "$REPO_DIR/engine/hooks/bug-complaint-leak" "$HOME/.cursor/hooks/bug-complaint-leak" link_item "reflect-on-thrash" "$REPO_DIR/engine/hooks/reflect-on-thrash" "$HOME/.cursor/hooks/reflect-on-thrash" link_item "scope-lock" "$REPO_DIR/engine/hooks/scope-lock" "$HOME/.cursor/hooks/scope-lock" @@ -280,6 +282,7 @@ link_item "ui-input-guard" "$REPO_DIR/engine/hooks/ui-input-guard" "$HOME/.curso echo "--- codex hooks (\$HOME/.codex/hooks) ---" mkdir -p "$HOME/.codex/hooks" +link_item "_flags" "$REPO_DIR/engine/hooks/_flags" "$HOME/.codex/hooks/_flags" link_item "diu-stop" "$REPO_DIR/engine/hooks/diu-stop" "$HOME/.codex/hooks/diu-stop" link_item "scope-lock" "$REPO_DIR/engine/hooks/scope-lock" "$HOME/.codex/hooks/scope-lock" link_item "auto-pr" "$REPO_DIR/engine/hooks/auto-pr" "$HOME/.codex/hooks/auto-pr" diff --git a/scripts/run_skill_scenarios.py b/scripts/run_skill_scenarios.py index 3b703280..4f06e3b9 100755 --- a/scripts/run_skill_scenarios.py +++ b/scripts/run_skill_scenarios.py @@ -40,6 +40,7 @@ HOOK_STATE_ENV = "WRONG_CHECK_REFLECT_STATE_DIR" JUDGE_STATE_ENV = "CATSTACK_LLM_JUDGE_STATE_DIR" JUDGE_RUNNERS_ENV = "CATSTACK_LLM_JUDGE_RUNNERS" +REFLECT_ENFORCEMENT_ENV = "CATSTACK_REFLECT_ENFORCEMENT" FAKE_JUDGE_RUNNER = ["scenario-fake", [sys.executable, "-c", "print('{\"match\": false}')", "{prompt}"]] _DETECT_CACHE: dict[str, object] = {} @@ -170,7 +171,26 @@ def skill_frontmatter(name: str) -> str | None: def check_scenario(scenario: dict, verbose: bool = False) -> list[str]: - """Failures for one scenario. Empty list means it passed.""" + """Failures for one scenario. Empty list means it passed. + + Runs with reflect/automate-me enforcement switched on, whatever the + caller's environment says, and puts the caller's value back afterwards. + Those hooks do nothing while the flag is off, so without this an + expect_silent on one of them would pass because the hook is off, not + because the detector chose silence. + """ + saved = os.environ.get(REFLECT_ENFORCEMENT_ENV) + os.environ[REFLECT_ENFORCEMENT_ENV] = "1" + try: + return _check_scenario(scenario, verbose) + finally: + if saved is None: + os.environ.pop(REFLECT_ENFORCEMENT_ENV, None) + else: + os.environ[REFLECT_ENFORCEMENT_ENV] = saved + + +def _check_scenario(scenario: dict, verbose: bool) -> list[str]: failures: list[str] = [] path = write_transcript(transcript_for(scenario)) diff --git a/tests/test_skill_scenarios.py b/tests/test_skill_scenarios.py index b4d9c90b..20781bc1 100644 --- a/tests/test_skill_scenarios.py +++ b/tests/test_skill_scenarios.py @@ -8,9 +8,11 @@ """ from __future__ import annotations +import os import sys import unittest from pathlib import Path +from unittest.mock import patch REPO = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPO / "scripts")) @@ -51,6 +53,28 @@ def test_a_wrong_expectation_is_reported(self): self.assertEqual(len(failures), 1, failures) self.assertIn("expected SILENCE", failures[0]) + def test_a_wrong_silence_on_an_opt_in_hook_is_reported_with_the_flag_off(self): + """Silence must come from the detector, not from the hook being off. + + verdict-flip-watch and wrong-check-reflect do nothing unless + CATSTACK_REFLECT_ENFORCEMENT is on. If the runner left that to the + caller's environment, every expect_silent / expect_no_enqueue on those + hooks would pass on any machine without the flag -- a check that could + not run, reporting clean. Inverting two scenarios that really fire + proves the runner switches the flag on itself. + """ + by_name = {s["name"]: s for s in rs.load_scenarios()} + flip = dict(by_name["stale-green-caught-without-any-admission"]) + flip["expect_silent"] = flip.pop("expect_fire") + judge = dict(by_name["admission-in-unlisted-wording-still-asks-the-judge"]) + judge["expect_no_enqueue"] = judge.pop("expect_enqueue") + with patch.dict(os.environ, {"CATSTACK_REFLECT_ENFORCEMENT": "0"}): + flip_failures = rs.check_scenario(flip) + judge_failures = rs.check_scenario(judge) + self.assertEqual(os.environ["CATSTACK_REFLECT_ENFORCEMENT"], "0") + self.assertTrue(any("expected SILENCE" in f for f in flip_failures), flip_failures) + self.assertTrue(any("expected NO judge job" in f for f in judge_failures), judge_failures) + def test_unknown_skill_name_is_reported(self): failures = rs.check_scenario( {"name": "x", "reply": "hi", "expect_skill_auto": ["no-such-skill-here"]}