diff --git a/README.md b/README.md index dbdef2c..75337c1 100644 --- a/README.md +++ b/README.md @@ -167,6 +167,7 @@ Full sourcing notes, including what was left out and why: [docs/provenance.md](d | `new-file-callout` | A new untracked file at the repo root or under `scripts/`: the reply must name it and say why. | | `agent-relay-attribution` | Advisory: facts relayed from a subagent's report must say so or be re-verified. | | `scratchpad-collision` | Two agents writing the same scratchpad file within ten minutes: use a uniquely named file. | +| `ui-input-guard` | Synthetic keystrokes, clicks, or screen recording aimed at the user's own session: blocked unless a hands-off window is open, the screen is unlocked, and the user is idle. | | `hook-freshness` | Advisory: the catstack checkout behind `~/.claude/hooks` is off `main` or behind `origin/main`, so merged hook fixes are not live on this machine. | | `auto-pr` | catstack itself changed: tell the agent to open a PR, no request needed. | | `cat-mode-default` | Every investigation or execution prompt, and every subagent prompt sent through the Agent tool: apply `cat-mode` without typing `/cat-mode`. Off unless `CATSTACK_CAT_MODE_DEFAULT=1` (env or `.env`; see `engine/hooks/cat-mode-default/README.md`). | diff --git a/corpus/skills/cat-mode/SKILL.md b/corpus/skills/cat-mode/SKILL.md index d554a7a..5e09943 100644 --- a/corpus/skills/cat-mode/SKILL.md +++ b/corpus/skills/cat-mode/SKILL.md @@ -250,9 +250,9 @@ agent switch, or resubmit is a fix, and none comes before the repro. **An interruption or stuck state gets instrument-level proof before a fix, and the fix goes to a subagent.** A poll loop not converging, a process not responding as expected, a restart that doesn't complete — treat this as its own investigation, not something to guess through inline. Gather real evidence first (the target's own logs, `ps -o stat,wchan`, a live query) before naming a cause, then delegate the actual fix to a subagent rather than hand-patching it in the main thread. A DO1 restart once looked hung on a stale PID; the owner's own log showed the real mechanism in two lines — `received SIGTERM, shutting down gracefully` followed 30s later by `process survived SIGTERM for 30000ms after worker stop; restarting worker` — a per-worker watchdog resurrecting mid-shutdown under real task load, not a hang. -**A factual or technical claim gets a real repro script, not a history search.** -Judging an old comment or a "probably confabulated" suspicion needs an actual attempt under the claimed conditions, not a `git log` sweep. No citation means -"never verified," not "false." A live repro proved a dismissed "yauzl hangs" comment was real on the pinned versions. +**UI testing must not disrupt the user's own session.** Prove a UI or surface change somewhere disposable — a test channel or workspace, a throwaway profile, a second display, a VM, a headless run. Driving the user's real keyboard, mouse, or screen is a last resort needing an explicit hands-off window first: state the acceptance test in one line, get the yes, `touch /tmp/.ui-input-window`, and remove it when the window closes; a PreToolUse hook (`engine/hooks/ui-input-guard/`) blocks synthetic input and screen recording while no window is open, the screen is locked, or the user is still typing. Stop at the first sign the session is theirs again (idle time drops, the frontmost app changes, the screen locks), and leave no residue: undo stray messages, pins, or reactions, or say what was left behind. + +**A factual or technical claim gets a real repro script, not a history search.** Judging an old comment or a "probably confabulated" suspicion needs an actual attempt under the claimed conditions, not a `git log` sweep. No citation means "never verified," not "false." **Unhedged root-cause or fix claims about live system behavior need instrument-level proof in the same message, or `UNVERIFIED:`.** The gate is the claim type ("this is why it's slow," "this is the bug"), not a diff --git a/docs/ecosystem.md b/docs/ecosystem.md index f074a8c..a06794f 100644 --- a/docs/ecosystem.md +++ b/docs/ecosystem.md @@ -82,6 +82,7 @@ again. | `new-file-callout` | hook | | `agent-relay-attribution` | hook (advisory) | | `scratchpad-collision` | hook | +| `ui-input-guard` | hook | | `hook-freshness` | hook (advisory) | | `engine/CLAUDE.core.md` | global hand-written Claude rules | | `scripts/`, `always-on/`, `cursor/rules/` (repo root), root `install.sh` | runtime (engine-owned entrypoints at root for CI) | diff --git a/engine/CLAUDE.core.md b/engine/CLAUDE.core.md index 525e170..dcb7362 100644 --- a/engine/CLAUDE.core.md +++ b/engine/CLAUDE.core.md @@ -26,6 +26,7 @@ These override brevity. If proof makes a message longer, the message gets longer - If a test was skipped, timed out, or I ran a subset, say exactly which and why — never let a partial run stand in for a full one. - If the user asks "did you verify X?", answer yes or no first, then show the evidence or admit there is none. Do not re-argue the original claim. - When I catch myself about to assert something I did not observe, stop and run the check instead of writing the sentence. +- **A check that could not run is not a pass.** When a guard, gate, scan, or query meets input it cannot read — a file past a size cap, an unresolved path, a field a projection omits, a probe that errored — it says so or refuses. It never returns clean. Give such a check three outcomes (hit, clean, unchecked), not two, and pin the third with a test; whether it then fails open or closed is a per-check decision that gets written down. Saltzer and Schroeder put the burden the same way in "Basic Principles of Information Protection" (1975): base access decisions on permission rather than exclusion, so the default is lack of access and the scheme names the conditions under which access is permitted (https://web.mit.edu/Saltzer/www/publications/protection/Basic.html). - The same rule applies to claims about the conversation itself, not just about code: "I ignored/missed/forgot X" is a claim that needs evidence too. Grep the actual transcript for the instruction before saying that. If nothing turns up, say "I don't have a record of that instruction in this session" — not self-blaming language for something that was never said. - A Grep or name hit is not a check. Do not cite a file, line, or "the bug is X" until this turn's Read or command output is in the same message. If two files could match, Read both. Prefix `UNVERIFIED:` until then. Saying "my earlier check was wrong" means the claim went out before the check — that is a process failure, not a polite recovery. @@ -37,6 +38,7 @@ These override brevity. If proof makes a message longer, the message gets longer - Before inviting me to test: run one full end-to-end machine-verified rehearsal of the exact flow I will perform. Pieces verified separately don't count as ready. Never say "go" on assembly alone. - Before the live test starts, restate the acceptance test in one sentence and get my yes ("the test is: you speak, and X happens"). I should never have to write it myself in caps. - Once I'm testing: freeze the demo surface. No edits, relaunches, or cosmetic changes to the thing I'm looking at unless I asked or the test is failing. Same session: an unrequested layout edit during the test window corrupted the demo page. Mechanically: when the live window opens, write the demo-surface paths (one absolute path, `dir/` prefix, or glob per line) to `/tmp/.demo-freeze`, and delete the file when the window ends — a PreToolUse hook (`engine/hooks/demo-freeze/`) blocks edits to matching paths while it exists (auto-expires after 2h). +- Prove UI work somewhere disposable (test channel, throwaway profile, second display, VM, headless run) instead of my live session. Driving my real keyboard, mouse, or screen needs an explicit hands-off window: state the acceptance test in one line, get my yes, then `touch /tmp/.ui-input-window` and delete it when the window ends — a PreToolUse hook (`engine/hooks/ui-input-guard/`) blocks synthetic input and screen recording while no window is open, the screen is locked, or I am still typing. Undo stray messages or reactions the run created, or say what was left. - Every message during a live window ends with exactly one action for me, or "nothing needed from you for ~N minutes". Never leave me waiting without a named next step. - If the deliverable is a same-day demo, plan the demo path first — the smallest end-to-end visible slice. Product-grade extras (settings UIs, multi-platform parity, test suites) come only after the demo runs. diff --git a/engine/hooks/scope-lock/tests/test_hooks.py b/engine/hooks/scope-lock/tests/test_hooks.py index c80d42e..5809fb2 100644 --- a/engine/hooks/scope-lock/tests/test_hooks.py +++ b/engine/hooks/scope-lock/tests/test_hooks.py @@ -64,6 +64,45 @@ def run_main(main, payload: dict) -> tuple[int, str, str]: return code, out.getvalue(), err.getvalue() +class TestUnreadableInput(unittest.TestCase): + """Pin what happens to input the detector cannot read. + + Two files feed a decision here, and they resolve in opposite directions + on purpose: a corrupt state file means no lock is in force, while an + unreadable transcript means no scope contract was recorded, so a lock + already in force is not released by a file that could not be read. + """ + + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.saved_state_dir = detect.STATE_DIR + detect.STATE_DIR = self.tmp.name + + def tearDown(self) -> None: + detect.STATE_DIR = self.saved_state_dir + self.tmp.cleanup() + + def test_fails_open_when_the_state_file_is_malformed(self): + payload = {"session_id": "session-corrupt"} + with open(detect.state_path(payload), "w", encoding="utf-8") as handle: + handle.write("{not json at all") + self.assertEqual(detect.load_state(payload), {}) + + def test_fails_open_when_the_state_file_holds_a_non_object(self): + payload = {"session_id": "session-list"} + with open(detect.state_path(payload), "w", encoding="utf-8") as handle: + handle.write("[1, 2, 3]") + self.assertEqual(detect.load_state(payload), {}) + + def test_unreadable_transcript_records_no_contract_so_a_lock_holds(self): + payload = {"session_id": "session-2", "transcript_path": "/nonexistent/session.jsonl"} + self.assertEqual(detect.recorded_contract(payload, 0), "") + + def test_missing_transcript_counts_zero_lines_rather_than_guessing(self): + payload = {"session_id": "session-3", "transcript_path": "/nonexistent/session.jsonl"} + self.assertEqual(detect._line_count(detect._transcript_path(payload)), 0) + + class ScopeLockCase(unittest.TestCase): def setUp(self) -> None: self.tmp = tempfile.TemporaryDirectory() diff --git a/engine/hooks/ui-input-guard/README.md b/engine/hooks/ui-input-guard/README.md new file mode 100644 index 0000000..f7a5cc2 --- /dev/null +++ b/engine/hooks/ui-input-guard/README.md @@ -0,0 +1,53 @@ +# ui-input-guard + +PreToolUse hook (Bash): never drive the user's own keyboard, mouse, or screen +uninvited. Synthetic input acts on the session the user is sitting in — typed +into the wrong window it sends real messages, trips real shortcuts, and lands +in the lock screen; a screen recording captures whatever they have open. + +Blocked mechanisms: AppleScript `System Events` with `keystroke`, `key code`, +or `click at`; `cliclick`; `xdotool`; `screencapture -V`; `ffmpeg` capturing +an `avfoundation` screen device. A command that runs a local script is scanned +through that script's contents, because the wrapper hides what it does, and +shell variables in the path are resolved first (`S=/tmp/run; $S/drive.sh`). + +Scripts are streamed in chunks rather than skipped for being large: a silent +skip is an unchecked file reported as clean. Past an 8 MB ceiling, or on a +read error, the command is refused with the path and the reason instead, +because a guard that cannot check does not assume safe. The escape is the +same hands-off marker, or splitting the input-driving part into a file that +can be read. + +Allowed when all three hold: + +1. A hands-off window is open — `touch /tmp/.ui-input-window` (override with + `UI_INPUT_WINDOW_FILE`), younger than 30 minutes. +2. The screen is not locked (macOS `CGSSessionScreenIsLocked`). +3. The user has been idle at least 10 seconds (macOS `HIDIdleTime`). + +Stays silent on the neighbours that only observe or author: `open` on a deep +link, an AppleScript geometry read, a still `screencapture`, `ffmpeg` +transcoding a file, a `cat > script < None: + try: + payload = json.load(sys.stdin) + except (json.JSONDecodeError, OSError): + return + try: + message = decide(payload if isinstance(payload, dict) else {}) + except Exception as exc: + sys.stderr.write(f"ui-input-guard: detector error, allowing this call: {exc!r}\n") + return + if not message: + return + sys.stderr.write(message + "\n") + sys.exit(2) + + +if __name__ == "__main__": + main() diff --git a/engine/hooks/ui-input-guard/detect.py b/engine/hooks/ui-input-guard/detect.py new file mode 100644 index 0000000..2e2bf10 --- /dev/null +++ b/engine/hooks/ui-input-guard/detect.py @@ -0,0 +1,361 @@ +"""ui-input-guard: never drive the user's own keyboard, mouse, or screen uninvited. + +Synthetic input (AppleScript `System Events` keystrokes and clicks, +`cliclick`, `xdotool`) and screen recording act on the session the user is +sitting in. Typed into the wrong window they send real messages, trip real +shortcuts, and land in the lock screen; a recording captures whatever the +user has open. + +A Bash command carrying one of those is blocked unless the user has granted +a hands-off window (a fresh marker file), the screen is unlocked, and the +user has been idle for a moment. Wrapper scripts count: a command that runs +a local file is scanned through that file's contents, because the tool text +alone hides what the script does. + +Blunt on purpose. Probe errors fail open; a missing marker does not. +""" +from __future__ import annotations + +import os +import re +import subprocess +import sys +import time + +MARKER = os.environ.get("UI_INPUT_WINDOW_FILE", "/tmp/.ui-input-window") +MAX_WINDOW_AGE_SECS = 30 * 60 +MIN_IDLE_SECS = 10 +SCAN_CHUNK_BYTES = 64 * 1024 +SCAN_OVERLAP_BYTES = 512 +MAX_SCRIPT_BYTES = 8 * 1024 * 1024 +PROBE_TIMEOUT_SECS = 3 + +SYNTHETIC_INPUT_RES = [ + ( + "AppleScript System Events input", + re.compile(r"System\s+Events", re.IGNORECASE | re.DOTALL), + re.compile(r"\bkeystroke\b|\bkey\s+code\b|\bclick\s+at\b", re.IGNORECASE | re.DOTALL), + ), +] +COMMAND_MECHANISMS = [ + ("cliclick", "cliclick", None), + ("xdotool", "xdotool", None), + ("screencapture video", "screencapture", re.compile(r"\s-{1,2}[Vv]\b")), + ( + "ffmpeg screen capture", + "ffmpeg", + re.compile(r"avfoundation[^\n]*Capture\s+screen", re.IGNORECASE), + ), +] + +READ_ONLY_COMMANDS = { + "ack", "ag", "awk", "cat", "cd", "cut", "echo", "find", "fgrep", "git", + "grep", "egrep", "head", "jq", "less", "ls", "printf", "pwd", "rg", "sed", + "sort", "tail", "test", "tr", "type", "uniq", "wc", "which", +} +SHELL_INTERPRETERS = {"applescript", "bash", "expect", "osascript", "sh", "zsh"} +DATA_INTERPRETERS = {"node", "perl", "python", "python3", "ruby"} +INLINE_CODE_RE = re.compile( + r"(?:^|[\s;|&(])(?:node|perl|python3?|ruby)\s+-\w*[ce]\w*\s+('(?:[^']|'\\'')*'|\"(?:[^\"\\]|\\.)*\")", + re.DOTALL, +) +ENV_ASSIGN_RE = re.compile(r"^\w+=") + +HEREDOC_RE = re.compile(r"^([^\n]*?)<<-?\s*['\"]?(\w+)['\"]?[^\n]*$", re.MULTILINE) +PATH_TOKEN_RE = re.compile(r"[\"']?((?:/|\./|\$\w+/|~/)[\w./$-]+\.(?:sh|bash|zsh|applescript|scpt|py|mjs|js))[\"']?") + +UNSCANNABLE_PREFIX = "unscannable:" + +UNSCANNABLE_MESSAGE = ( + "ui-input-guard: this command runs {path}, which could not be read to the " + "end ({why}), so whether it drives the user's live session is unknown. A " + "guard that cannot check does not assume safe. Either run the mechanism " + "inline where it can be seen, split the script so the input-driving part " + "is its own readable file, or grant a hands-off window with " + "`touch {marker}` if you already know what it does." +) + +MESSAGE = ( + "ui-input-guard: this command drives the user's live session ({reason}){where}. " + "{state}\n" + "Get an explicit hands-off window first, then `touch {marker}` (expires in 30 " + "minutes). Prefer a surface that is not the user's own: a test channel or " + "workspace, a throwaway profile, a second display, a VM, or a headless run. " + "Never while the screen is locked or the user is typing." +) + + +def strip_write_heredocs(command): + """Drop heredoc bodies that are data, keeping the ones an interpreter runs. + + `cat > script.sh <<'EOF'` authors text and `git commit -F -` carries a + message; `osascript <<'AS'` runs what follows. Only a heredoc introduced + by a shell or AppleScript interpreter is kept, so a Python or Node + heredoc that merely holds these words as data does not fire. + """ + text = command or "" + for match in list(HEREDOC_RE.finditer(text)): + introducer = match.group(1) + tag = match.group(2) + words = [w for w in introducer.split() if not ENV_ASSIGN_RE.match(w)] + heads = {os.path.basename(w.strip("\"'()")) for w in words} + if heads & SHELL_INTERPRETERS: + continue + body = re.compile( + r"(" + re.escape(match.group(0)) + r")\n.*?\n" + re.escape(tag) + r"\s*$", + re.DOTALL | re.MULTILINE, + ) + text = body.sub(r"\1\n", text, count=1) + return text + + +def split_segments(text): + """Split a command into stages on unquoted `;`, `|`, `&&`, and newlines. + + Quote-aware: a `|` inside a search pattern is part of that pattern, not a + pipe, so `git grep -E "a|b"` stays one read-only stage. + """ + parts = [] + current = [] + quote = "" + index = 0 + while index < len(text or ""): + char = text[index] + if char == "\\" and quote != "'": + current.append(text[index:index + 2]) + index += 2 + continue + if quote: + if char == quote: + quote = "" + current.append(char) + elif char in "'\"": + quote = char + current.append(char) + elif char in ";\n|&": + parts.append("".join(current)) + current = [] + while index + 1 < len(text) and text[index + 1] in "|&": + index += 1 + else: + current.append(char) + index += 1 + parts.append("".join(current)) + return parts + + +def segments(command): + """Pipeline stages, each with its leading command word resolved.""" + out = [] + for raw in split_segments(command or ""): + segment = raw.strip() + if not segment: + continue + words = [w for w in segment.split() if not ENV_ASSIGN_RE.match(w) and not w.startswith(("<", ">"))] + head = os.path.basename(words[0].strip("\"'()")) if words else "" + out.append((head, segment)) + return out + + +def is_read_only_pipeline(command): + """True when every stage only reads: a search pattern is not an action.""" + stages = segments(command) + if not stages: + return False + return all(head in READ_ONLY_COMMANDS for head, _ in stages if head) + + +def resolve_shell_vars(command): + """Substitute `VAR=value` assignments made earlier in the same command. + + `S=/tmp/run; $S/drive.sh` is the shape an agent writes when it stages a + helper script, so the path token has to be resolved before the file can + be scanned at all. + """ + text = command or "" + for name, value in re.findall(r"(?m)(?:^|[;&|]\s*|^\s*)(\w+)=([^\s;&|]+)", text): + if "$" in value: + continue + cleaned = value.strip("\"'") + text = text.replace(f"${{{name}}}", cleaned).replace(f"${name}", cleaned) + return text + + +def scan_file(path, opener=open, getsize=os.path.getsize): + """(mechanism, unscannable reason) for one file, read in bounded chunks. + + Streams the whole file rather than skipping a large one: a silent skip is + an unchecked file reported as clean. Past the hard ceiling, or on a read + error, the reason is returned so the caller can refuse instead of + guessing. + """ + try: + size = getsize(path) + except OSError as exc: + return None, f"stat failed: {exc.strerror or exc}" + if size > MAX_SCRIPT_BYTES: + return None, f"{size} bytes, over the {MAX_SCRIPT_BYTES}-byte scan ceiling" + try: + with opener(path, encoding="utf-8", errors="replace") as handle: + carry = "" + while True: + chunk = handle.read(SCAN_CHUNK_BYTES) + if not chunk: + return None, None + reason = synthetic_input_reason(carry + chunk) + if reason: + return reason, None + carry = chunk[-SCAN_OVERLAP_BYTES:] + except OSError as exc: + return None, f"read failed: {exc.strerror or exc}" + + +def script_paths(command, isfile=os.path.isfile): + """Local script paths this command runs, with shell variables resolved.""" + found = [] + for match in PATH_TOKEN_RE.finditer(resolve_shell_vars(command)): + path = os.path.expanduser(match.group(1)) + if "$" in path or path in found: + continue + if isfile(path): + found.append(path) + return found + + +def strip_inline_program_text(command): + """Drop code passed to a non-shell interpreter with -c or -e. + + `python3 -c "...keystroke..."` is a program that holds these words as + data, the same as a Python heredoc; `osascript -e` is not stripped, + because there the words are the mechanism. + """ + text = command or "" + for match in list(INLINE_CODE_RE.finditer(text)): + text = text.replace(match.group(1), "''", 1) + return text + + +def synthetic_input_reason(text): + """Name of the synthetic-input mechanism in this text, or None.""" + haystack = text or "" + for label, first_re, second_re in SYNTHETIC_INPUT_RES: + if first_re.search(haystack) and second_re.search(haystack): + return label + for label, tool, flag_re in COMMAND_MECHANISMS: + for head, segment in segments(haystack): + if head != tool: + continue + if flag_re is None or flag_re.search(segment): + return label + return None + + +def find_reason(command, **io): + """(reason, source) for a command, following wrapper scripts. ('', '') if clean. + + A third shape exists: `source` is the path of a script that could not be + read to the end, returned with an empty reason so the caller refuses + rather than treating an unchecked file as clean. + """ + visible = strip_inline_program_text(strip_write_heredocs(command)) + if is_read_only_pipeline(visible): + return "", "" + reason = synthetic_input_reason(visible) + if reason: + return reason, "this command" + unscannable = None + for path in script_paths(visible, **{k: v for k, v in io.items() if k == "isfile"}): + found, why = scan_file(path, **{k: v for k, v in io.items() if k in ("opener", "getsize")}) + if found: + return found, "the script it runs" + if why and unscannable is None: + unscannable = (path, why) + if unscannable: + return "", UNSCANNABLE_PREFIX + "\t".join(unscannable) + return "", "" + + +def _probe(args, run=None): + runner = run or subprocess.run + result = runner(args, capture_output=True, text=True, timeout=PROBE_TIMEOUT_SECS, check=False) + return result.stdout or "" + + +def screen_is_locked(run=None, platform=None): + """True when the macOS login session is locked; None when unknown.""" + if (platform or sys.platform) != "darwin": + return None + try: + out = _probe(["ioreg", "-n", "Root", "-d1", "-a"], run=run) + except (OSError, subprocess.SubprocessError): + return None + match = re.search(r"CGSSessionScreenIsLocked\s*<(true|false)/>", out) + if not match: + return False if "CGSSessionScreenIsLocked" not in out else None + return match.group(1) == "true" + + +def idle_seconds(run=None, platform=None): + """Seconds since the last real keyboard or mouse event; None when unknown.""" + if (platform or sys.platform) != "darwin": + return None + try: + out = _probe(["ioreg", "-c", "IOHIDSystem"], run=run) + except (OSError, subprocess.SubprocessError): + return None + match = re.search(r'"HIDIdleTime"\s*=\s*(\d+)', out) + if not match: + return None + return int(match.group(1)) / 1_000_000_000 + + +def window_age_seconds(marker=None, now=time.time, stat=os.stat): + """Age of the hands-off marker in seconds, or None when there is none.""" + path = marker or MARKER + try: + return now() - stat(path).st_mtime + except OSError: + return None + + +def blocking_state(age, locked, idle, marker=None): + """Why the live session is off limits right now, or '' when it is granted.""" + path = marker or MARKER + if age is None: + return f"No hands-off window is open ({path} is missing)." + if age > MAX_WINDOW_AGE_SECS: + return f"The hands-off window in {path} expired; ask for a new one." + if locked is True: + return "The screen is locked, so input would go to the lock screen." + if idle is not None and idle < MIN_IDLE_SECS: + return f"The user was active {idle:.0f}s ago; wait until the session is idle." + return "" + + +def decide(payload, marker=None, now=time.time, stat=os.stat, run=None, platform=None, **io): + """Blocking feedback for a PreToolUse Bash call, or None to allow it.""" + tool_input = payload.get("tool_input") or {} + command = tool_input.get("command") or "" + reason, source = find_reason(command, **io) + if not reason and source.startswith(UNSCANNABLE_PREFIX): + path, why = source[len(UNSCANNABLE_PREFIX):].split("\t", 1) + if blocking_state( + window_age_seconds(marker=marker, now=now, stat=stat), + screen_is_locked(run=run, platform=platform), + idle_seconds(run=run, platform=platform), + marker=marker, + ): + return UNSCANNABLE_MESSAGE.format(path=path, why=why, marker=marker or MARKER) + return None + if not reason: + return None + state = blocking_state( + window_age_seconds(marker=marker, now=now, stat=stat), + screen_is_locked(run=run, platform=platform), + idle_seconds(run=run, platform=platform), + marker=marker, + ) + if not state: + return None + where = "" if source == "this command" else f" via {source}" + return MESSAGE.format(reason=reason, where=where, state=state, marker=marker or MARKER) diff --git a/engine/hooks/ui-input-guard/install_claude_hook.py b/engine/hooks/ui-input-guard/install_claude_hook.py new file mode 100644 index 0000000..9d794b4 --- /dev/null +++ b/engine/hooks/ui-input-guard/install_claude_hook.py @@ -0,0 +1,46 @@ +#!/usr/bin/env python3 +"""Merge ui-input-guard into ~/.claude/settings.json PreToolUse hooks. Idempotent.""" +from __future__ import annotations + +import json +import os + +HERE = os.path.dirname(os.path.abspath(__file__)) +SETTINGS_PATH = os.path.expanduser("~/.claude/settings.json") +FRAGMENT_PATH = os.path.join(HERE, "claude.hook.json") +MARKER = "ui-input-guard/claude_pretooluse_check.py" +EVENT = "PreToolUse" + + +def _is_ours(entry: dict) -> bool: + return any(MARKER in h.get("command", "") for h in entry.get("hooks", [])) + + +def merge_hook(settings: dict, fragment: dict) -> bool: + entry_list = settings.setdefault("hooks", {}).setdefault(EVENT, []) + new_entries = fragment.get("hooks", {}).get(EVENT, []) + before = json.dumps(entry_list, sort_keys=True) + kept = [e for e in entry_list if not _is_ours(e)] + entry_list[:] = kept + new_entries + return json.dumps(entry_list, sort_keys=True) != before + + +def main() -> None: + settings: dict = {} + if os.path.exists(SETTINGS_PATH): + with open(SETTINGS_PATH) as handle: + settings = json.load(handle) + with open(FRAGMENT_PATH) as handle: + fragment = json.load(handle) + if not merge_hook(settings, fragment): + print("ok claude PreToolUse ui-input-guard already up to date") + return + os.makedirs(os.path.dirname(SETTINGS_PATH), exist_ok=True) + with open(SETTINGS_PATH, "w") as handle: + json.dump(settings, handle, indent=2) + handle.write("\n") + print("added claude PreToolUse ui-input-guard") + + +if __name__ == "__main__": + main() diff --git a/engine/hooks/ui-input-guard/tests/fixtures/commands_fire.json b/engine/hooks/ui-input-guard/tests/fixtures/commands_fire.json new file mode 100644 index 0000000..d6186b0 --- /dev/null +++ b/engine/hooks/ui-input-guard/tests/fixtures/commands_fire.json @@ -0,0 +1,41 @@ +[ + { + "label": "AppleScript keystroke into Slack, sanitized from a real session", + "command": "osascript <<'AS'\ntell application \"Slack\" to activate\ndelay 0.8\ntell application \"System Events\"\n tell process \"Slack\"\n set frontmost to true\n keystroke \"3\"\n key code 36\n end tell\nend tell\nAS", + "reason": "AppleScript System Events input" + }, + { + "label": "AppleScript click at a computed window coordinate", + "command": "osascript -e 'tell application \"System Events\" to tell process \"Slack\" to click at {753, 422}'", + "reason": "AppleScript System Events input" + }, + { + "label": "ffmpeg screen recording of the user's display", + "command": "ffmpeg -hide_banner -f avfoundation -framerate 10 -i \"Capture screen 0:none\" -c:v libx264 /tmp/before.mp4", + "reason": "ffmpeg screen capture" + }, + { + "label": "screencapture in video mode", + "command": "screencapture -V 5 -x /tmp/clip.mov", + "reason": "screencapture video" + }, + { + "label": "cliclick", + "command": "cliclick c:500,400", + "reason": "cliclick" + }, + { + "label": "xdotool on linux", + "command": "xdotool type --delay 20 'hello'", + "reason": "xdotool" + }, + { + "label": "wrapper script the command merely runs", + "command": "bash /tmp/ui-input-guard-fixture/slack-type.sh \"3\"", + "reason": "AppleScript System Events input", + "script": { + "path": "/tmp/ui-input-guard-fixture/slack-type.sh", + "body": "#!/usr/bin/env bash\nosascript -e 'tell application \"System Events\" to tell process \"Slack\" to keystroke \"3\"'\n" + } + } +] diff --git a/engine/hooks/ui-input-guard/tests/fixtures/commands_silent.json b/engine/hooks/ui-input-guard/tests/fixtures/commands_silent.json new file mode 100644 index 0000000..b5e0925 --- /dev/null +++ b/engine/hooks/ui-input-guard/tests/fixtures/commands_silent.json @@ -0,0 +1,50 @@ +[ + { + "label": "opening a deep link does not drive input", + "command": "open \"slack://channel?team=T0AJUNMP7T7&id=C0BKK4A2091\"" + }, + { + "label": "AppleScript that only reads window geometry", + "command": "osascript -e 'tell application \"System Events\" to tell process \"Slack\" to get {position, size} of window 1'" + }, + { + "label": "AppleScript notification, no input synthesis", + "command": "osascript -e 'display notification \"done\"'" + }, + { + "label": "a still screenshot, not a recording", + "command": "screencapture -x -t png /tmp/state.png" + }, + { + "label": "ffmpeg transcoding a file, not the screen", + "command": "ffmpeg -hide_banner -i /tmp/before-raw.mp4 -vf fps=1/20 /tmp/frames/f%02d.png" + }, + { + "label": "authoring a driver script is not running it", + "command": "cat > /tmp/slack-type.sh <<'EOF'\nosascript -e 'tell application \"System Events\" to keystroke \"3\"'\nEOF\nchmod +x /tmp/slack-type.sh" + }, + { + "label": "grepping for the pattern in a transcript", + "command": "rg -n 'System Events.*keystroke' ~/.claude/projects/*/*.jsonl | head" + }, + { + "label": "git grep whose pattern lists the mechanisms (real session, quote-aware split)", + "command": "C=/repo/catstack; cd $C; echo \"=== UI input guard\"; git grep -n -iE \"keystroke|HIDIdleTime|System Events|cliclick|synthetic input\" origin/main -- engine corpus 2>/dev/null | cut -c1-140 | head -6" + }, + { + "label": "python heredoc editing the detector holds the words as data (real session)", + "command": "cd /repo/hooks/ui-input-guard && python3 - <<'PY'\np='detect.py'; s=open(p).read()\ns = s.replace('keystroke', 'key code') # System Events\nopen(p,'w').write(s)\nPY" + }, + { + "label": "probing whether a clicker is installed is not clicking", + "command": "which cliclick 2>&1; brew list cliclick 2>&1 | head -3" + }, + { + "label": "git commit whose message body names the mechanisms", + "command": "git commit -q -F - <<'MSG'\nhook: block cliclick and System Events keystroke input\n\nBody mentioning keystroke and click at.\nMSG" + }, + { + "label": "python -c holding the mechanism words as literals while editing this hook (real session)", + "command": "cd /repo/hooks/ui-input-guard && python3 -c \"import sys; sys.path.insert(0,'.'); import detect; print(detect.find_reason('osascript -e tell application System Events to keystroke 3'))\"" + } +] diff --git a/engine/hooks/ui-input-guard/tests/test_hooks.py b/engine/hooks/ui-input-guard/tests/test_hooks.py new file mode 100644 index 0000000..36e7ddb --- /dev/null +++ b/engine/hooks/ui-input-guard/tests/test_hooks.py @@ -0,0 +1,268 @@ +#!/usr/bin/env python3 +"""Tests for the ui-input-guard PreToolUse hook. + +Run: python3 -m unittest discover -s engine/hooks/ui-input-guard/tests -v + +Fixtures are sanitized commands from a real recording session: AppleScript +keystrokes into Slack, an ffmpeg screen capture, and the wrapper script the +agent actually invoked. The silent set holds the neighbours that must stay +allowed -- a deep link, a geometry read, a still screenshot, and authoring a +driver script. +""" +from __future__ import annotations + +import io +import json +import os +import sys +import tempfile +import unittest +from contextlib import redirect_stderr +from unittest.mock import patch + +HOOK_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +FIXTURES = os.path.join(HOOK_DIR, "tests", "fixtures") +sys.path.insert(0, HOOK_DIR) + +import claude_pretooluse_check # noqa: E402 +import detect # noqa: E402 + +DRIVE_LINE = ( + "osascript -e 'tell application \"System Events\" to keystroke \"3\"'\n" +) + +IOREG_UNLOCKED = 'CGSSessionScreenIsLocked\t' +IOREG_LOCKED = 'CGSSessionScreenIsLocked\t' + + +def load(name): + with open(os.path.join(FIXTURES, name), encoding="utf-8") as handle: + return json.load(handle) + + +class FakeProc: + def __init__(self, stdout): + self.stdout = stdout + + +def fake_probe(locked=False, idle_secs=300): + def run(args, **kwargs): + if "Root" in args: + return FakeProc(IOREG_LOCKED if locked else IOREG_UNLOCKED) + return FakeProc(f' "HIDIdleTime" = {int(idle_secs * 1_000_000_000)}') + return run + + +def materialize(case): + """Write a fixture's wrapper script to disk; return its directory or None.""" + script = case.get("script") + if not script: + return None + path = script["path"] + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as handle: + handle.write(script["body"]) + return path + + +def payload(command): + return {"tool_name": "Bash", "tool_input": {"command": command}} + + +def run_hook(command, marker_exists=False): + err = io.StringIO() + with patch.object(sys, "stdin", io.StringIO(json.dumps(payload(command)))): + with redirect_stderr(err): + try: + claude_pretooluse_check.main() + except SystemExit as exc: + return exc.code, err.getvalue() + return 0, err.getvalue() + + +class TestBlocksLiveSessionInput(unittest.TestCase): + def test_hit_every_fire_fixture_is_detected(self): + for case in load("commands_fire.json"): + with self.subTest(label=case["label"]): + written = materialize(case) + try: + reason, _ = detect.find_reason(case["command"]) + finally: + if written and os.path.exists(written): + os.unlink(written) + self.assertEqual(reason, case["reason"]) + + def test_hit_blocks_when_no_window_is_open(self): + case = load("commands_fire.json")[0] + with tempfile.TemporaryDirectory() as tmp: + message = detect.decide( + payload(case["command"]), + marker=os.path.join(tmp, "absent"), + platform="darwin", run=fake_probe(), + ) + self.assertIsNotNone(message) + self.assertIn("No hands-off window", message) + + def test_hit_blocks_when_the_window_expired(self): + with tempfile.NamedTemporaryFile(suffix=".marker") as marker: + stale = lambda: 0 # noqa: E731 - now() far behind the marker's mtime + message = detect.decide( + payload(load("commands_fire.json")[1]["command"]), + marker=marker.name, + now=lambda: os.stat(marker.name).st_mtime + detect.MAX_WINDOW_AGE_SECS + 60, + platform="darwin", run=fake_probe(), + ) + self.assertIsNotNone(stale) + self.assertIsNotNone(message) + self.assertIn("expired", message) + + def test_hit_blocks_when_the_screen_is_locked(self): + with tempfile.NamedTemporaryFile(suffix=".marker") as marker: + message = detect.decide( + payload(load("commands_fire.json")[0]["command"]), + marker=marker.name, + platform="darwin", run=fake_probe(locked=True), + ) + self.assertIsNotNone(message) + self.assertIn("locked", message) + + def test_hit_blocks_while_the_user_is_active(self): + with tempfile.NamedTemporaryFile(suffix=".marker") as marker: + message = detect.decide( + payload(load("commands_fire.json")[0]["command"]), + marker=marker.name, + platform="darwin", run=fake_probe(idle_secs=1), + ) + self.assertIsNotNone(message) + self.assertIn("was active", message) + + def test_hit_exit_code_is_2_with_guidance(self): + code, err = run_hook(load("commands_fire.json")[2]["command"]) + self.assertEqual(code, 2) + self.assertIn("ui-input-guard", err) + self.assertIn("hands-off window", err) + + def test_hit_names_the_wrapper_script_as_the_source(self): + case = load("commands_fire.json")[-1] + written = materialize(case) + try: + with tempfile.TemporaryDirectory() as tmp: + message = detect.decide( + payload(case["command"]), + marker=os.path.join(tmp, "absent"), + platform="darwin", run=fake_probe(), + ) + finally: + if written and os.path.exists(written): + os.unlink(written) + self.assertIsNotNone(message) + self.assertIn("the script it runs", message) + + +class TestLargeAndIndirectScripts(unittest.TestCase): + def _staged(self, body): + directory = tempfile.mkdtemp() + path = os.path.join(directory, "drive.sh") + with open(path, "w", encoding="utf-8") as handle: + handle.write(body) + return path + + def test_hit_mechanism_past_the_old_size_cap_is_still_found(self): + padding = "# filler line to push the real call far into the file\n" * 4000 + path = self._staged("#!/usr/bin/env bash\n" + padding + DRIVE_LINE) + self.assertGreater(os.path.getsize(path), 64 * 1024) + reason, _ = detect.scan_file(path) + self.assertEqual(reason, "AppleScript System Events input") + + def test_hit_wrapper_reached_through_a_shell_variable(self): + path = self._staged("#!/usr/bin/env bash\n" + DRIVE_LINE) + directory = os.path.dirname(path) + command = f"S={directory}; $S/drive.sh \"3\"" + with tempfile.TemporaryDirectory() as tmp: + message = detect.decide(payload(command), marker=os.path.join(tmp, "absent"), platform="darwin", run=fake_probe()) + self.assertIsNotNone(message) + self.assertIn("the script it runs", message) + + def test_hit_unreadable_script_is_refused_not_assumed_clean(self): + path = self._staged("#!/usr/bin/env bash\necho hello\n") + with tempfile.TemporaryDirectory() as tmp: + message = detect.decide( + payload(f"bash {path}"), + marker=os.path.join(tmp, "absent"), + platform="darwin", run=fake_probe(), + getsize=lambda _p: detect.MAX_SCRIPT_BYTES + 1, + ) + self.assertIsNotNone(message) + self.assertIn("could not be read to the end", message) + self.assertIn("scan ceiling", message) + + def test_no_hit_large_script_with_no_mechanism(self): + path = self._staged("#!/usr/bin/env bash\n" + ("echo padding\n" * 20000)) + self.assertGreater(os.path.getsize(path), 64 * 1024) + self.assertEqual(detect.scan_file(path), (None, None)) + self.assertIsNone(detect.decide(payload(f"bash {path}"), marker="/nonexistent/marker")) + + def test_no_hit_unscannable_script_inside_a_granted_window(self): + path = self._staged("#!/usr/bin/env bash\necho hello\n") + with tempfile.NamedTemporaryFile(suffix=".marker") as marker: + message = detect.decide( + payload(f"bash {path}"), + marker=marker.name, + platform="darwin", run=fake_probe(idle_secs=120), + getsize=lambda _p: detect.MAX_SCRIPT_BYTES + 1, + ) + self.assertIsNone(message) + + +class TestAllowsEverythingElse(unittest.TestCase): + def test_no_hit_on_any_silent_fixture(self): + for case in load("commands_silent.json"): + with self.subTest(label=case["label"]): + self.assertIsNone(detect.decide(payload(case["command"]), marker="/nonexistent/marker")) + + def test_allows_input_inside_a_granted_window(self): + with tempfile.NamedTemporaryFile(suffix=".marker") as marker: + message = detect.decide( + payload(load("commands_fire.json")[0]["command"]), + marker=marker.name, + platform="darwin", run=fake_probe(locked=False, idle_secs=120), + ) + self.assertIsNone(message) + + def test_allows_a_non_bash_payload_without_a_command(self): + self.assertIsNone(detect.decide({"tool_name": "Read", "tool_input": {"file_path": "/x"}})) + + def test_fails_open_on_garbage_stdin(self): + err = io.StringIO() + with patch.object(sys, "stdin", io.StringIO("not json")): + with redirect_stderr(err): + claude_pretooluse_check.main() + self.assertEqual(err.getvalue(), "") + + def test_no_hit_probes_are_skipped_off_macos(self): + self.assertIsNone(detect.screen_is_locked(platform="linux")) + self.assertIsNone(detect.idle_seconds(platform="linux")) + + def test_fails_open_when_probes_raise(self): + def boom(*args, **kwargs): + raise OSError("no ioreg here") + + self.assertIsNone(detect.screen_is_locked(run=boom, platform="darwin")) + self.assertIsNone(detect.idle_seconds(run=boom, platform="darwin")) + + def test_no_hit_search_pattern_is_not_a_pipe(self): + command = 'git grep -n -iE "keystroke|System Events" origin/main | head -3' + self.assertTrue(detect.is_read_only_pipeline(command)) + self.assertEqual(detect.find_reason(command), ("", "")) + + def test_shell_heredoc_still_counts_as_execution(self): + command = "osascript <<'AS'\ntell application \"System Events\" to keystroke \"3\"\nAS" + self.assertEqual(detect.find_reason(command)[0], "AppleScript System Events input") + + def test_write_heredoc_body_is_not_treated_as_execution(self): + command = load("commands_silent.json")[-2]["command"] + self.assertNotIn("keystroke", detect.strip_write_heredocs(command)) + + +if __name__ == "__main__": + unittest.main() diff --git a/engine/hooks/wrong-check-reflect/detect.py b/engine/hooks/wrong-check-reflect/detect.py index 9644470..c41aa79 100644 --- a/engine/hooks/wrong-check-reflect/detect.py +++ b/engine/hooks/wrong-check-reflect/detect.py @@ -64,6 +64,17 @@ re.compile( r"(?i)\bi\s+misread\s+(it|that|this|my\s+own|the)\b" ), + re.compile( + r"(?i)\byour\s+(?:instinct|hunch|gut|suspicion|read)\s+(?:was|were)\s+right\b" + ), + re.compile( + r"(?i)^\s*[*_#\s>-]*(?:you'?re\s+right|you\s+are\s+right|good\s+catch)\b" + r".{0,200}?(?:verifying\s+(?:it\s+|that\s+)?now|checking\s+(?:it\s+|that\s+)?now|" + r"i\s+hadn'?t\b|i\s+had\s+not\b|i\s+didn'?t\b|i\s+did\s+not\b|" + r"i\s+should\s+have\b|instead\s+of\s+(?:labeling|labelling|assuming|guessing)|" + r"i\s+never\s+(?:ran|checked|read|verified))", + re.DOTALL, + ), ] # Hypothetical / product-blame shapes that must stay silent even if a diff --git a/engine/hooks/wrong-check-reflect/tests/test_hooks.py b/engine/hooks/wrong-check-reflect/tests/test_hooks.py index 3b05883..b27d876 100644 --- a/engine/hooks/wrong-check-reflect/tests/test_hooks.py +++ b/engine/hooks/wrong-check-reflect/tests/test_hooks.py @@ -92,6 +92,31 @@ def test_hit_i_misread_without_youre_right_prefix(self): detect.find_admission("I misread the front matter on that skill.") ) + def test_hit_youre_right_verifying_it_now(self): + self.assertIsNotNone(detect.find_admission( + "You're right. Verifying it now instead of labeling it." + )) + + def test_hit_good_catch_i_should_have_checked(self): + self.assertIsNotNone(detect.find_admission( + "Good catch on the hook. I should have run the two greps before sending that." + )) + + def test_no_hit_youre_right_agreeing_with_a_choice(self): + self.assertIsNone(detect.find_admission( + "You're right that the second option is cheaper, so I will build that one." + )) + + def test_hit_your_instinct_was_right_stands_alone(self): + self.assertIsNotNone(detect.find_admission( + "Your instinct was right — the size cap was silently skipping files." + )) + + def test_hit_your_hunch_was_right(self): + self.assertIsNotNone(detect.find_admission( + "Your hunch was right, the wrapper path was never resolved." + )) + def test_no_hit_product_test_was_wrong(self): self.assertIsNone(detect.find_admission("the test was wrong")) diff --git a/engine/skills/reflect/SKILL.md b/engine/skills/reflect/SKILL.md index 9658c4c..ddc33f0 100644 --- a/engine/skills/reflect/SKILL.md +++ b/engine/skills/reflect/SKILL.md @@ -52,6 +52,7 @@ Every invocation of this skill — single-transcript or multi-conversation mode - A non-trivial workflow emerged that isn't captured anywhere. - A session, or a corpus-scan bucket, shows heavy user involvement — many corrections, clarifying answers typed out by hand, repeated manual confirmations — over a short span. That is a **FAILURE**, not a preference ping: the user had to stay in the loop because the agent missed a named constraint. Route to `automate-me` (step 4). Do not write a one-off task-skill edit and call it done. - The user said "you fucked up", "you messed up", "I told you", "you're ignoring me", or equivalent agent-blame. Treat this reflect pass as FAIL. The class is *ignored named constraint*, not the swear word. Product-blame ("the UI is messed up") is not this class. +- The reply conceded the user's suspicion — "your instinct was right", "your hunch was right", "good catch" followed by a correction. Conceding means the user found what the agent's own checks did not, so the miss is the finding, not the concession. Treat as FAIL; the `wrong-check-reflect` detector fires on these shapes. - The user had to keep iterating, restate requirements, or change product direction because the agent missed something already named. FAIL, then `automate-me`. A genuine mind-change (user learned new facts, then redirected) is not failure. A forced restatement of an already-named constraint is. - The same *type* of complaint appears in 2+ turns or 2+ sessions (repro-then-fix, UI proof before done, e2e before claiming pass, obey the named verb). That class is a bug. **Must** invoke `automate-me` — not optional, do not wait for the user to say "automate me." `token_audit.py`'s `intervention-must-automate` flag is the mechanical catch; human-message only, never tool_result / skill-injection / `/loop` polls. - It's been a while since the corpus-wide pass (`top_sessions.py` + this skill's lenses across the worst offenders) last ran. No fixed cadence and no cron — just periodically worth doing by hand. diff --git a/install.sh b/install.sh index 8ace0c2..b136a8c 100755 --- a/install.sh +++ b/install.sh @@ -239,6 +239,7 @@ link_item "hedge-runs-prove-it" "$REPO_DIR/engine/hooks/hedge-runs-prove-it" "$H link_item "new-file-callout" "$REPO_DIR/engine/hooks/new-file-callout" "$HOME/.claude/hooks/new-file-callout" link_item "agent-relay-attribution" "$REPO_DIR/engine/hooks/agent-relay-attribution" "$HOME/.claude/hooks/agent-relay-attribution" link_item "scratchpad-collision" "$REPO_DIR/engine/hooks/scratchpad-collision" "$HOME/.claude/hooks/scratchpad-collision" +link_item "ui-input-guard" "$REPO_DIR/engine/hooks/ui-input-guard" "$HOME/.claude/hooks/ui-input-guard" link_item "hook-freshness" "$REPO_DIR/engine/hooks/hook-freshness" "$HOME/.claude/hooks/hook-freshness" echo "--- cursor hooks dir (\$HOME/.cursor/hooks) ---" @@ -251,6 +252,7 @@ link_item "pr-schema-gate" "$REPO_DIR/engine/hooks/pr-schema-gate" "$HOME/.curso link_item "wrong-check-reflect" "$REPO_DIR/engine/hooks/wrong-check-reflect" "$HOME/.cursor/hooks/wrong-check-reflect" link_item "build-the-lever" "$REPO_DIR/engine/hooks/build-the-lever" "$HOME/.cursor/hooks/build-the-lever" link_item "repeat-error-stop" "$REPO_DIR/engine/hooks/repeat-error-stop" "$HOME/.cursor/hooks/repeat-error-stop" +link_item "ui-input-guard" "$REPO_DIR/engine/hooks/ui-input-guard" "$HOME/.cursor/hooks/ui-input-guard" echo "--- codex hooks (\$HOME/.codex/hooks) ---" mkdir -p "$HOME/.codex/hooks" @@ -261,6 +263,7 @@ link_item "pr-schema-gate" "$REPO_DIR/engine/hooks/pr-schema-gate" "$HOME/.codex link_item "wrong-check-reflect" "$REPO_DIR/engine/hooks/wrong-check-reflect" "$HOME/.codex/hooks/wrong-check-reflect" link_item "build-the-lever" "$REPO_DIR/engine/hooks/build-the-lever" "$HOME/.codex/hooks/build-the-lever" link_item "repeat-error-stop" "$REPO_DIR/engine/hooks/repeat-error-stop" "$HOME/.codex/hooks/repeat-error-stop" +link_item "ui-input-guard" "$REPO_DIR/engine/hooks/ui-input-guard" "$HOME/.codex/hooks/ui-input-guard" # Deleted worktrees leave symlinks behind that point into this repo but at a # path that no longer exists (e.g. .worktrees//engine/hooks/). @@ -335,6 +338,7 @@ python3 "$REPO_DIR/engine/hooks/hedge-runs-prove-it/install_claude_hook.py" python3 "$REPO_DIR/engine/hooks/new-file-callout/install_claude_hook.py" python3 "$REPO_DIR/engine/hooks/agent-relay-attribution/install_claude_hook.py" python3 "$REPO_DIR/engine/hooks/scratchpad-collision/install_claude_hook.py" +python3 "$REPO_DIR/engine/hooks/ui-input-guard/install_claude_hook.py" python3 "$REPO_DIR/engine/hooks/hook-freshness/install_claude_hook.py" python3 "$REPO_DIR/scripts/prune_dead_hook_entries.py" diff --git a/scripts/check_hook_test_coverage.py b/scripts/check_hook_test_coverage.py index 22a58a5..ba89e6f 100755 --- a/scripts/check_hook_test_coverage.py +++ b/scripts/check_hook_test_coverage.py @@ -38,6 +38,31 @@ "clean", "empty", ) +UNCHECKED_RE = ( + "unscannable", + "unchecked", + "unreadable", + "too_large", + "over_ceiling", + "refus", + "cannot_check", + "could_not_read", + "fails_open", + "fail_open", + "missing", + "nonexistent", + "no_transcript", + "garbage", + "corrupt", + "malformed", +) +READS_INPUT_RE = ( + "open(", + "read(", + "getsize", + "st_size", + "isfile", +) POSITIVE_RE = ( "hit", "fire", @@ -79,6 +104,21 @@ def _test_names(tests_dir: str) -> list[str]: return names +def reads_external_input(detector_path: str) -> bool: + """True when a detector opens or sizes a file it did not receive inline. + + Such a detector has a third outcome besides hit and clean: input it could + not read. That outcome has to be pinned by a test, whichever way the hook + resolves it, so it cannot silently collapse into clean. + """ + try: + with open(detector_path, encoding="utf-8") as handle: + source = handle.read() + except OSError: + return False + return any(token in source for token in READS_INPUT_RE) + + def hooks_with_detector() -> list[str]: if not os.path.isdir(HOOKS_DIR): return [] @@ -109,6 +149,17 @@ def check_hook(hook_dir: str) -> list[str]: f"{name}: no negative test found (a test name matching {NEGATIVE_RE} " "that proves the detector stays silent on a clean case)" ) + detector = os.path.join(hook_dir, "detect.py") + if reads_external_input(detector) and not any( + pat in n.lower() for n in names for pat in UNCHECKED_RE + ): + problems.append( + f"{name}: detect.py reads files but no test pins its behavior on input it " + "could not read (name it for the unreadable case: unscannable, unreadable, " + "too_large, refuses, fails_open, missing, malformed). Fail open or fail " + "closed is the hook's own documented choice; leaving it untested, so an " + "unchecked file passes as clean, is not." + ) return problems diff --git a/tests/test_cat_mode.py b/tests/test_cat_mode.py index a3526cf..8f1cf20 100644 --- a/tests/test_cat_mode.py +++ b/tests/test_cat_mode.py @@ -100,6 +100,33 @@ def test_body_names_the_default_hook_and_flag(self): self.assertEqual(parse_frontmatter(read_skill_text())["disable-model-invocation"], "true") +class TestUiTestingRule(unittest.TestCase): + """The rule that keeps a UI proof run off the user's own session. + + Prose cannot be executed, but the pieces an agent has to act on -- a + disposable surface, a granted window, the marker path the guard hook + reads, and cleanup -- must all still be named, and the marker path must + match the hook that enforces it. + """ + + def test_names_a_disposable_surface_and_a_granted_window(self): + text = read_skill_text().lower() + self.assertIn("disposable", text) + self.assertIn("hands-off window", text) + + def test_names_the_marker_path_the_guard_hook_reads(self): + self.assertIn("/tmp/.ui-input-window", read_skill_text()) + + def test_marker_path_matches_the_hook_default(self): + hook = os.path.join(REPO_ROOT, "engine", "hooks", "ui-input-guard", "detect.py") + with open(hook, encoding="utf-8") as handle: + self.assertIn('"/tmp/.ui-input-window"', handle.read()) + + def test_requires_cleanup_of_what_the_run_left_behind(self): + text = read_skill_text().lower() + self.assertTrue("residue" in text or "undo stray" in text, "cleanup rule missing") + + class TestCatModeReferences(unittest.TestCase): def test_every_referenced_skill_still_exists(self): text = read_skill_text() diff --git a/tests/test_check_hook_test_coverage.py b/tests/test_check_hook_test_coverage.py new file mode 100644 index 0000000..13a05d9 --- /dev/null +++ b/tests/test_check_hook_test_coverage.py @@ -0,0 +1,99 @@ +#!/usr/bin/env python3 +"""Positive + negative tests for check_hook_test_coverage's unchecked-input rule. + +A detector that opens a file has a third outcome besides hit and clean: +input it could not read. This gate requires a test that pins that outcome, +whichever way the hook resolves it, so an unchecked file cannot pass as +clean by default. Fixtures here are synthetic hook directories. +""" +from __future__ import annotations + +import sys +import tempfile +import unittest +from pathlib import Path + +REPO = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO / "scripts")) +import check_hook_test_coverage as chtc # noqa: E402 + +FIRES_AND_SILENT = ( + "def test_hit_detects_the_bad_case():\n pass\n\n" + "def test_no_hit_stays_silent_on_a_clean_case():\n pass\n" +) +UNREADABLE_TEST = "\n\ndef test_fails_open_on_unreadable_input():\n pass\n" +INLINE_DETECTOR = "import re\nPATTERN = re.compile('x')\n\ndef decide(payload):\n return None\n" +FILE_READING_DETECTOR = ( + "import os\n\n" + "def decide(payload):\n" + " path = payload['tool_input']['path']\n" + " if os.path.getsize(path) > 1024:\n" + " return None\n" + " with open(path) as handle:\n" + " return handle.read()[:1]\n" +) + + +def hook(root: Path, name: str, detector: str, tests: str) -> Path: + hook_dir = root / name + (hook_dir / "tests").mkdir(parents=True) + (hook_dir / "detect.py").write_text(detector, encoding="utf-8") + (hook_dir / "tests" / "test_hooks.py").write_text(tests, encoding="utf-8") + return hook_dir + + +class TestUncheckedInputRule(unittest.TestCase): + def test_file_reading_detector_without_an_unreadable_test_fails(self): + with tempfile.TemporaryDirectory() as tmp: + hook_dir = hook(Path(tmp), "reader", FILE_READING_DETECTOR, FIRES_AND_SILENT) + problems = chtc.check_hook(str(hook_dir)) + self.assertEqual(len(problems), 1, problems) + self.assertIn("could not read", problems[0]) + + def test_file_reading_detector_with_an_unreadable_test_passes(self): + with tempfile.TemporaryDirectory() as tmp: + hook_dir = hook( + Path(tmp), "reader", FILE_READING_DETECTOR, FIRES_AND_SILENT + UNREADABLE_TEST + ) + self.assertEqual(chtc.check_hook(str(hook_dir)), []) + + def test_inline_only_detector_is_not_asked_for_an_unreadable_test(self): + with tempfile.TemporaryDirectory() as tmp: + hook_dir = hook(Path(tmp), "inline", INLINE_DETECTOR, FIRES_AND_SILENT) + self.assertEqual(chtc.check_hook(str(hook_dir)), []) + + def test_reads_external_input_detects_a_size_check(self): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "detect.py" + path.write_text(FILE_READING_DETECTOR, encoding="utf-8") + self.assertTrue(chtc.reads_external_input(str(path))) + + def test_reads_external_input_is_false_for_an_inline_detector(self): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "detect.py" + path.write_text(INLINE_DETECTOR, encoding="utf-8") + self.assertFalse(chtc.reads_external_input(str(path))) + + def test_missing_detector_file_fails_open_rather_than_erroring(self): + self.assertFalse(chtc.reads_external_input("/nonexistent/detect.py")) + + +class TestExistingRulesStillHold(unittest.TestCase): + def test_no_positive_test_still_fails(self): + with tempfile.TemporaryDirectory() as tmp: + hook_dir = hook( + Path(tmp), "silent-only", INLINE_DETECTOR, + "def test_no_hit_stays_silent():\n pass\n", + ) + problems = chtc.check_hook(str(hook_dir)) + self.assertTrue(any("no positive test" in p for p in problems), problems) + + def test_the_real_repo_passes_this_gate(self): + problems = [] + for hook_dir in chtc.hooks_with_detector(): + problems.extend(chtc.check_hook(hook_dir)) + self.assertEqual(problems, []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_install.py b/tests/test_install.py index 51dbcd2..c69eb90 100644 --- a/tests/test_install.py +++ b/tests/test_install.py @@ -189,6 +189,15 @@ def test_hedge_runs_prove_it_linked_and_stop_wired_for_claude(self): 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) + def test_ui_input_guard_linked_and_pretooluse_wired_for_claude(self): + target = os.path.join(self.fake_home, ".claude", "hooks", "ui-input-guard") + self.assertTrue(os.path.islink(target), target) + self.assertEqual(os.readlink(target), hook_src("ui-input-guard")) + commands = self._claude_hook_commands("PreToolUse") + self.assertTrue( + any("ui-input-guard/claude_pretooluse_check.py" in c for c in commands), commands + ) + def test_hook_freshness_linked_and_prompt_wired_for_claude(self): target = os.path.join(self.fake_home, ".claude", "hooks", "hook-freshness") self.assertTrue(os.path.islink(target), target)