diff --git a/corpus/skills/cat-mode/SKILL.md b/corpus/skills/cat-mode/SKILL.md index d554a7ab..5e09943e 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/engine/CLAUDE.core.md b/engine/CLAUDE.core.md index 525e1708..488047d4 100644 --- a/engine/CLAUDE.core.md +++ b/engine/CLAUDE.core.md @@ -37,6 +37,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/hook-freshness/README.md b/engine/hooks/hook-freshness/README.md new file mode 100644 index 00000000..8a2d4cb7 --- /dev/null +++ b/engine/hooks/hook-freshness/README.md @@ -0,0 +1,33 @@ +# hook-freshness + +UserPromptSubmit hook: the installed hooks are only as new as the checkout +behind them. `install.sh` symlinks `~/.claude/hooks/` at a catstack +checkout, so a hook or skill fix merged on `origin/main` does nothing on this +machine while that checkout sits on a feature branch or behind the remote. + +Resolves the checkout from the `~/.claude/hooks/diu-stop` symlink (override +with `CATSTACK_HOOKS_REPO`), reads `git branch --show-current` and +`git rev-list --count HEAD..origin/main`, and adds one advisory line to the +turn's context when the checkout is off `main` or behind it. Once per +session, keyed by transcript path. + +Advisory only — never blocks. No network by default; set +`CATSTACK_HOOK_FRESHNESS_FETCH=1` to allow a 3-second `git fetch` first, so +the count is not itself stale. `CATSTACK_HOOK_FRESHNESS=0` silences it. +Fails open on every error: no symlink, no git, a detached HEAD, a timeout. + +## Files + +- `detect.py` — checkout resolution, `repo_state()`, `advisory()`, `decide()`. +- `claude_prompt_submit.py` — Claude UserPromptSubmit entrypoint. +- `claude.prompt.hook.json` / `install_claude_hook.py` — settings.json merge (idempotent). +- `tests/test_hooks.py` — stale branch, behind count, clean checkout, fail-open. + +## Env + +| Var | Effect | +|-----|--------| +| `CATSTACK_HOOKS_REPO` | Use this checkout instead of resolving the symlink. | +| `CATSTACK_HOOK_FRESHNESS_FETCH=1` | Allow a short `git fetch origin main` first. | +| `CATSTACK_HOOK_FRESHNESS=0` | Silence the advisory. | +| `HOOK_FRESHNESS_STATE_DIR` | Once-per-session marker directory. | diff --git a/engine/hooks/hook-freshness/claude.prompt.hook.json b/engine/hooks/hook-freshness/claude.prompt.hook.json new file mode 100644 index 00000000..167939b9 --- /dev/null +++ b/engine/hooks/hook-freshness/claude.prompt.hook.json @@ -0,0 +1,15 @@ +{ + "hooks": { + "UserPromptSubmit": [ + { + "hooks": [ + { + "type": "command", + "command": "python3 $HOME/.claude/hooks/hook-freshness/claude_prompt_submit.py", + "timeout": 10 + } + ] + } + ] + } +} diff --git a/engine/hooks/hook-freshness/claude_prompt_submit.py b/engine/hooks/hook-freshness/claude_prompt_submit.py new file mode 100644 index 00000000..3cc5d026 --- /dev/null +++ b/engine/hooks/hook-freshness/claude_prompt_submit.py @@ -0,0 +1,28 @@ +#!/usr/bin/env python3 +"""Claude Code UserPromptSubmit: warn once per session when the catstack +checkout behind ~/.claude/hooks is off main or behind origin/main, so merged +hook fixes that are not live here get noticed. Advisory, fail-open, no block. +""" +from __future__ import annotations + +import json +import sys + +from detect import decide_json + + +def main() -> None: + try: + payload = json.load(sys.stdin) + except (json.JSONDecodeError, OSError): + return + try: + out = decide_json(payload if isinstance(payload, dict) else {}) + except Exception: + return + if out: + print(out) + + +if __name__ == "__main__": + main() diff --git a/engine/hooks/hook-freshness/detect.py b/engine/hooks/hook-freshness/detect.py new file mode 100644 index 00000000..6cb7dfa2 --- /dev/null +++ b/engine/hooks/hook-freshness/detect.py @@ -0,0 +1,145 @@ +"""hook-freshness: the installed hooks are only as new as the checkout behind them. + +`install.sh` symlinks `~/.claude/hooks/` at a catstack checkout, so a +hook fix that is merged on `origin/main` does nothing on this machine while +that checkout sits on a feature branch or behind the remote. This resolves +the checkout from the `diu-stop` symlink, reads its branch and its distance +from `origin/main`, and returns one advisory line for the turn. + +Advisory only: no block, no LLM, no network unless +CATSTACK_HOOK_FRESHNESS_FETCH=1. Fails open on every error. +""" +from __future__ import annotations + +import hashlib +import json +import os +import subprocess + +STATE_DIR = os.environ.get( + "HOOK_FRESHNESS_STATE_DIR", + os.path.join(os.path.expanduser("~"), ".cache", "catstack-hook-freshness"), +) + +ANCHOR_LINK = os.path.join(os.path.expanduser("~"), ".claude", "hooks", "diu-stop") +TRUNK = "origin/main" +FETCH_TIMEOUT_SECS = 3 +GIT_TIMEOUT_SECS = 5 + +MESSAGE = ( + "catstack hooks are stale: the checkout behind ~/.claude/hooks is {detail}. " + "Merged hook and skill fixes are not live here until it is updated. Run " + "`git -C {repo} pull --ff-only` (or merge {trunk} into the branch), then " + "`{repo}/install.sh`, and restart the harness." +) + + +def _run_git(args, cwd, timeout=GIT_TIMEOUT_SECS): + result = subprocess.run( + ["git", *args], + cwd=cwd, + capture_output=True, + text=True, + timeout=timeout, + check=False, + ) + if result.returncode != 0: + return None + return result.stdout.strip() + + +def resolve_repo(env=None, realpath=os.path.realpath, isdir=os.path.isdir): + """The catstack checkout the live hooks point at, or None.""" + env = env if env is not None else os.environ + override = env.get("CATSTACK_HOOKS_REPO") + if override: + return override if isdir(os.path.join(override, ".git")) else None + try: + target = realpath(ANCHOR_LINK) + except OSError: + return None + repo = os.path.dirname(os.path.dirname(os.path.dirname(target))) + return repo if isdir(os.path.join(repo, ".git")) else None + + +def repo_state(repo, env=None, run=_run_git): + """(branch, commits behind trunk) for the checkout, or (None, None).""" + env = env if env is not None else os.environ + try: + if env.get("CATSTACK_HOOK_FRESHNESS_FETCH") == "1": + run(["fetch", "--quiet", "origin", "main"], repo, FETCH_TIMEOUT_SECS) + branch = run(["branch", "--show-current"], repo) + behind_raw = run(["rev-list", "--count", f"HEAD..{TRUNK}"], repo) + except (OSError, subprocess.SubprocessError): + return None, None + if behind_raw is None: + return branch, None + try: + return branch, int(behind_raw) + except ValueError: + return branch, None + + +def advisory(repo, branch, behind): + """One line when the checkout is off trunk or behind it, else None.""" + if not repo or behind is None: + return None + off_trunk = bool(branch) and branch != "main" + if not off_trunk and behind <= 0: + return None + parts = [] + if off_trunk: + parts.append(f"on branch `{branch}`") + if behind > 0: + commit_word = "commit" if behind == 1 else "commits" + parts.append(f"{behind} {commit_word} behind {TRUNK}") + return MESSAGE.format(detail=" and ".join(parts), repo=repo, trunk=TRUNK) + + +def _state_file(key): + digest = hashlib.sha256((key or "no-transcript").encode("utf-8")).hexdigest()[:16] + return os.path.join(STATE_DIR, f"{digest}.advised") + + +def already_advised(key): + return os.path.isfile(_state_file(key)) + + +def mark_advised(key): + path = _state_file(key) + try: + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as handle: + handle.write((key or "") + "\n") + except OSError: + pass + + +def decide(payload, env=None, run=_run_git, state=True): + """Advisory context for this prompt, or None. Once per session.""" + env = env if env is not None else os.environ + if env.get("CATSTACK_HOOK_FRESHNESS") == "0": + return None + key = payload.get("transcript_path") or payload.get("transcriptPath") or "" + if state and already_advised(key): + return None + repo = resolve_repo(env=env) + if not repo: + return None + branch, behind = repo_state(repo, env=env, run=run) + line = advisory(repo, branch, behind) + if line and state: + mark_advised(key) + return line + + +def decide_json(payload, env=None, run=_run_git, state=True): + line = decide(payload, env=env, run=run, state=state) + if not line: + return None + return json.dumps({ + "hookSpecificOutput": { + "hookEventName": "UserPromptSubmit", + "additionalContext": line, + } + }) diff --git a/engine/hooks/hook-freshness/install_claude_hook.py b/engine/hooks/hook-freshness/install_claude_hook.py new file mode 100644 index 00000000..63e610b5 --- /dev/null +++ b/engine/hooks/hook-freshness/install_claude_hook.py @@ -0,0 +1,46 @@ +#!/usr/bin/env python3 +"""Merge hook-freshness into ~/.claude/settings.json UserPromptSubmit 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.prompt.hook.json") +MARKER = "hook-freshness/claude_prompt_submit.py" +EVENT = "UserPromptSubmit" + + +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 UserPromptSubmit hook-freshness 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 UserPromptSubmit hook-freshness") + + +if __name__ == "__main__": + main() diff --git a/engine/hooks/hook-freshness/tests/test_hooks.py b/engine/hooks/hook-freshness/tests/test_hooks.py new file mode 100644 index 00000000..38f91a92 --- /dev/null +++ b/engine/hooks/hook-freshness/tests/test_hooks.py @@ -0,0 +1,144 @@ +#!/usr/bin/env python3 +"""Tests for the hook-freshness UserPromptSubmit hook. + +Run: python3 -m unittest discover -s engine/hooks/hook-freshness/tests -v + +Git is injected as a fake runner, so no test touches a real repo or the +network. The stale case mirrors the real failure: the checkout behind the +live symlinks on a feature branch, dozens of commits behind origin/main, +with merged hook fixes therefore not installed. +""" +from __future__ import annotations + +import io +import json +import os +import sys +import tempfile +import unittest +from contextlib import redirect_stdout +from unittest.mock import patch + +HOOK_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, HOOK_DIR) + +import claude_prompt_submit # noqa: E402 +import detect # noqa: E402 + + +def fake_git(branch="main", behind="0", fail=(), record=None): + def run(args, cwd, timeout=None): + if record is not None: + record.append(args) + key = args[0] + if key in fail: + return None + if key == "branch": + return branch + if key == "rev-list": + return behind + if key == "fetch": + return "" + return "" + return run + + +class TestAdvisoryFires(unittest.TestCase): + def test_hit_feature_branch_far_behind_trunk(self): + line = detect.advisory("/repo/catstack", "feat/skill-usage-log", 40) + self.assertIsNotNone(line) + self.assertIn("feat/skill-usage-log", line) + self.assertIn("40 commits behind origin/main", line) + self.assertIn("install.sh", line) + + def test_hit_on_main_but_one_commit_behind(self): + line = detect.advisory("/repo/catstack", "main", 1) + self.assertIsNotNone(line) + self.assertIn("1 commit behind", line) + + def test_hit_decide_returns_context_json(self): + with tempfile.TemporaryDirectory() as tmp: + repo = os.path.join(tmp, "catstack") + os.makedirs(os.path.join(repo, ".git")) + env = {"CATSTACK_HOOKS_REPO": repo, "HOOK_FRESHNESS_STATE_DIR": tmp} + with patch.object(detect, "STATE_DIR", tmp): + out = detect.decide_json( + {"transcript_path": os.path.join(tmp, "t.jsonl")}, + env=env, + run=fake_git(branch="feat/x", behind="3"), + ) + payload = json.loads(out) + self.assertEqual(payload["hookSpecificOutput"]["hookEventName"], "UserPromptSubmit") + self.assertIn("3 commits behind", payload["hookSpecificOutput"]["additionalContext"]) + + def test_hit_prints_once_per_session(self): + with tempfile.TemporaryDirectory() as tmp: + repo = os.path.join(tmp, "catstack") + os.makedirs(os.path.join(repo, ".git")) + env = {"CATSTACK_HOOKS_REPO": repo} + payload = {"transcript_path": os.path.join(tmp, "t.jsonl")} + with patch.object(detect, "STATE_DIR", tmp): + first = detect.decide(payload, env=env, run=fake_git(branch="feat/x", behind="3")) + second = detect.decide(payload, env=env, run=fake_git(branch="feat/x", behind="3")) + self.assertIsNotNone(first) + self.assertIsNone(second) + + +class TestAdvisorySilent(unittest.TestCase): + def test_no_hit_on_main_up_to_date(self): + self.assertIsNone(detect.advisory("/repo/catstack", "main", 0)) + + def test_no_hit_when_behind_count_unavailable(self): + self.assertIsNone(detect.advisory("/repo/catstack", "main", None)) + + def test_no_hit_when_repo_unresolvable(self): + self.assertIsNone(detect.decide({}, env={"CATSTACK_HOOKS_REPO": "/nope/not/a/repo"})) + + def test_no_hit_when_disabled_by_env(self): + self.assertIsNone(detect.decide({}, env={"CATSTACK_HOOK_FRESHNESS": "0"})) + + def test_fails_open_when_git_errors(self): + with tempfile.TemporaryDirectory() as tmp: + repo = os.path.join(tmp, "catstack") + os.makedirs(os.path.join(repo, ".git")) + branch, behind = detect.repo_state(repo, env={}, run=fake_git(fail=("rev-list",))) + self.assertIsNone(behind) + + def test_no_fetch_unless_opted_in(self): + calls = [] + with tempfile.TemporaryDirectory() as tmp: + repo = os.path.join(tmp, "catstack") + os.makedirs(os.path.join(repo, ".git")) + detect.repo_state(repo, env={}, run=fake_git(record=calls)) + self.assertNotIn("fetch", [c[0] for c in calls]) + + def test_fetch_when_opted_in(self): + calls = [] + with tempfile.TemporaryDirectory() as tmp: + repo = os.path.join(tmp, "catstack") + os.makedirs(os.path.join(repo, ".git")) + detect.repo_state(repo, env={"CATSTACK_HOOK_FRESHNESS_FETCH": "1"}, run=fake_git(record=calls)) + self.assertIn("fetch", [c[0] for c in calls]) + + def test_fails_open_on_garbage_stdin(self): + out = io.StringIO() + with patch.object(sys, "stdin", io.StringIO("not json")): + with redirect_stdout(out): + claude_prompt_submit.main() + self.assertEqual(out.getvalue(), "") + + +class TestRepoResolution(unittest.TestCase): + def test_hit_resolves_repo_from_symlink_target(self): + with tempfile.TemporaryDirectory() as tmp: + repo = os.path.join(tmp, "catstack") + target = os.path.join(repo, "engine", "hooks", "diu-stop") + os.makedirs(target) + os.makedirs(os.path.join(repo, ".git")) + with patch.object(detect, "ANCHOR_LINK", os.path.join(tmp, "link")): + os.symlink(target, os.path.join(tmp, "link")) + self.assertEqual(detect.resolve_repo(env={}), os.path.realpath(repo)) + + +if __name__ == "__main__": + unittest.main() diff --git a/engine/hooks/ui-input-guard/README.md b/engine/hooks/ui-input-guard/README.md new file mode 100644 index 00000000..f7a5cc27 --- /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 00000000..f9e4e82e --- /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): + """True when the macOS login session is locked; None when unknown.""" + if 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): + """Seconds since the last real keyboard or mouse event; None when unknown.""" + if 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, **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), + idle_seconds(run=run), + 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), + idle_seconds(run=run), + 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 00000000..9d794b46 --- /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 00000000..d6186b08 --- /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 00000000..b5e09258 --- /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 00000000..2095c0f4 --- /dev/null +++ b/engine/hooks/ui-input-guard/tests/test_hooks.py @@ -0,0 +1,264 @@ +#!/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"), + 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, + 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, + 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, + 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"), + 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"), 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"), + 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, + 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, + 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_fails_open_when_probes_raise(self): + def boom(*args, **kwargs): + raise OSError("no ioreg here") + + self.assertIsNone(detect.screen_is_locked(run=boom)) + self.assertIsNone(detect.idle_seconds(run=boom)) + + 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 96444704..c41aa795 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 3b058834..b27d876a 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 9658c4cf..ddc33f0d 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/tests/test_cat_mode.py b/tests/test_cat_mode.py index a3526cfc..8f1cf20d 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()