From 3199d8afcc1876ff44c0e85ba2852b2f7b422a09 Mon Sep 17 00:00:00 2001 From: Edbert Chan Date: Sat, 10 Oct 2026 23:05:42 -0700 Subject: [PATCH] Block a test-runner pass that stands in for the real command. The proof gate checked ship claims, not whether the program that showed the failure was the program that showed the fix. Co-authored-by: Cursor Change-Id: I161a67e5cce7fea332ca43c4eddfb5ba0ce308c8 --- engine/hooks/prove-it-ship-gate/README.md | 10 + .../hooks/prove-it-ship-gate/cursor.hook.json | 10 + .../prove-it-ship-gate/cursor_stop_check.py | 20 ++ engine/hooks/prove-it-ship-gate/detect.py | 254 ++++++++++++++++-- .../prove-it-ship-gate/tests/test_hooks.py | 128 +++++++++ 5 files changed, 407 insertions(+), 15 deletions(-) create mode 100644 engine/hooks/prove-it-ship-gate/cursor.hook.json create mode 100644 engine/hooks/prove-it-ship-gate/cursor_stop_check.py diff --git a/engine/hooks/prove-it-ship-gate/README.md b/engine/hooks/prove-it-ship-gate/README.md index e69b4a358..5a980ab25 100644 --- a/engine/hooks/prove-it-ship-gate/README.md +++ b/engine/hooks/prove-it-ship-gate/README.md @@ -48,6 +48,16 @@ Fixture tests and UI registration do not count. That is the whole point: Invoker PRs #10553-#10558 shipped cross-repo-research after unit + fixture + UI only, and the user had to force a live Linear e2e. +## Same command + +A product command (a hyphenated program such as `invoker-cli doctor`) that +already ran in the session, including a subagent transcript beside it, has +to run again after the last file edit. A later `vitest` / `pnpm test` / +`pytest` run does not replace it. The hook names the missing program and +subcommand and blocks. Unix pipelines, `node -e` scripts, and a session that +never ran a product command stay silent. A quoted program-and-subcommand pair inside a script counts as that command. +The transcript read fails open. + Mechanical half of `corpus/skills/prove-it-ship-gate`. Judgment (is this work really live-side-effect work) stays with the model; the hook only matches shapes. diff --git a/engine/hooks/prove-it-ship-gate/cursor.hook.json b/engine/hooks/prove-it-ship-gate/cursor.hook.json new file mode 100644 index 000000000..9f8bbc2b2 --- /dev/null +++ b/engine/hooks/prove-it-ship-gate/cursor.hook.json @@ -0,0 +1,10 @@ +{ + "hooks": { + "stop": [ + { + "command": "python3 $HOME/.cursor/hooks/prove-it-ship-gate/cursor_stop_check.py", + "timeout": 10 + } + ] + } +} diff --git a/engine/hooks/prove-it-ship-gate/cursor_stop_check.py b/engine/hooks/prove-it-ship-gate/cursor_stop_check.py new file mode 100644 index 000000000..a2f36b577 --- /dev/null +++ b/engine/hooks/prove-it-ship-gate/cursor_stop_check.py @@ -0,0 +1,20 @@ +#!/usr/bin/env python3 +"""Cursor Stop hook entrypoint for prove-it-ship-gate.""" +from __future__ import annotations + +import os +import sys + +sys.path.insert(0, os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_sdk")) + +from detect import detect # noqa: E402 +from runtime import run_hook # noqa: E402 + + +def main() -> None: + run_hook("prove-it-ship-gate", "cursor", detect) + + +if __name__ == "__main__": + main() diff --git a/engine/hooks/prove-it-ship-gate/detect.py b/engine/hooks/prove-it-ship-gate/detect.py index dfd255d63..eec0fdded 100644 --- a/engine/hooks/prove-it-ship-gate/detect.py +++ b/engine/hooks/prove-it-ship-gate/detect.py @@ -31,6 +31,7 @@ import json import re import os +import shlex import sys import hashlib @@ -43,6 +44,31 @@ from finding import Finding # noqa: E402 RULE_UNPROVEN_LIVE_CLAIM = "prove-it-ship-gate.unproven-live-claim" +RULE_SAME_REPRO = "prove-it-ship-gate.same-repro-not-rerun" + +SHELL_TOOL_NAMES = frozenset({"Bash", "Shell", "bash", "shell"}) +EDIT_TOOL_NAMES = frozenset({"Write", "StrReplace", "Edit", "EditNotebook", "ApplyPatch"}) +COMMAND_PREFIXES = frozenset({"sudo", "env", "time", "command", "nohup", "nice"}) +GENERIC_PROGRAMS = frozenset({ + "git", "node", "python", "python3", "bash", "zsh", "sh", "ls", "cat", "rg", + "grep", "find", "mkdir", "rm", "cp", "mv", "ln", "sed", "awk", "jq", "gh", + "curl", "ssh", "pnpm", "npm", "npx", "vitest", "pytest", "cargo", "go", + "make", "docker", "which", "readlink", "file", "echo", "head", "tail", "wc", + "sort", "uniq", "tee", "xargs", "tar", "diff", "patch", "perl", "ruby", + "open", "ps", "kill", "pwd", "date", "brew", "touch", "chmod", "mktemp", + "cut", "tr", "printf", "cd", "export", "true", "false", +}) +CLI_PROGRAM_RE = re.compile(r"^[a-z][a-z0-9]*(-[a-z][a-z0-9]*)+$") +EMBEDDED_CLI_RE = re.compile( + r"""['\"]([a-z][a-z0-9]*(?:-[a-z][a-z0-9]*)+)['\"]\s*,\s*['\"]([a-z][a-z0-9-]*|--help|--version)['\"]""" +) +CLI_SUBCOMMAND_RE = re.compile(r"^[a-z][a-z0-9-]*$") +TEST_RUNNER_RE = re.compile( + r"\bvitest\b|\bpytest\b|\bunittest\b|\bjest\b|" + r"\bpnpm\b[^\n]{0,120}\b(?:run\s+)?test\b|\bnpm\s+test\b|" + r"\bcargo\s+test\b|\bgo\s+test\b", + re.IGNORECASE, +) # A claim is a status assertion about the work, not any mention of the word. @@ -181,33 +207,231 @@ def bash_commands_this_turn(transcript_path: str) -> list[str] | None: data = json.loads(raw) except json.JSONDecodeError: continue - if not isinstance(data, dict) or data.get("type") != "assistant": - continue - message = data.get("message") - content = message.get("content") if isinstance(message, dict) else None - if not isinstance(content, list): + if not isinstance(data, dict): continue - for block in content: - if isinstance(block, dict) and block.get("type") == "tool_use" and block.get("name") == "Bash": + for block in _tool_uses(data): + if block.get("name") in SHELL_TOOL_NAMES: command = (block.get("input") or {}).get("command") if isinstance(command, str): commands.append(command) return commands +def _tool_uses(data: dict) -> list[dict]: + message = data.get("message") if isinstance(data.get("message"), dict) else {} + role = data.get("role") or message.get("role") or data.get("type") + if role != "assistant": + return [] + content = message.get("content") if message else data.get("content") + if not isinstance(content, list): + return [] + return [ + block for block in content + if isinstance(block, dict) and block.get("type") == "tool_use" + ] + + +def _unquoted_segments(command: str) -> list[str]: + parts: list[str] = [] + buf: list[str] = [] + quote = "" + i = 0 + while i < len(command): + ch = command[i] + if quote: + buf.append(ch) + if ch == quote and command[i - 1] != "\\": + quote = "" + i += 1 + continue + if ch in ("'", '"'): + quote = ch + buf.append(ch) + i += 1 + continue + if command.startswith("&&", i) or command.startswith("||", i) or command.startswith("<<", i): + if command.startswith("<<", i): + break + parts.append("".join(buf)) + buf = [] + i += 2 + continue + if ch in ";&|\n": + parts.append("".join(buf)) + buf = [] + i += 1 + continue + buf.append(ch) + i += 1 + parts.append("".join(buf)) + return [part.strip() for part in parts if part.strip()] + + +def surface_keys(command: str) -> list[str]: + """Program plus subcommand for a real CLI invocation. Empty for scripts, + pipelines of unix tools, and test runners.""" + keys: list[str] = [] + for segment in _unquoted_segments(command): + try: + words = shlex.split(segment, posix=True) + except ValueError: + continue + while words and words[0] in COMMAND_PREFIXES: + words = words[1:] + if not words: + continue + program = os.path.basename(words[0]) + if program in GENERIC_PROGRAMS or not CLI_PROGRAM_RE.fullmatch(program): + continue + subcommand = "" + for word in words[1:]: + if word in ("--help", "--version"): + subcommand = word + break + if word.startswith("-"): + continue + if CLI_SUBCOMMAND_RE.fullmatch(word): + subcommand = word + break + keys.append(f"{program} {subcommand}".strip()) + for program, subcommand in EMBEDDED_CLI_RE.findall(command): + if CLI_SUBCOMMAND_RE.fullmatch(subcommand) or subcommand in ("--help", "--version"): + keys.append(f"{program} {subcommand}") + return keys + + +def _is_test_runner(command: str) -> bool: + return bool(TEST_RUNNER_RE.search(command)) + + +def _read_events(path: str) -> list[tuple[str, str]] | None: + events: list[tuple[str, str]] = [] + try: + with open(path, encoding="utf-8") as handle: + lines = handle.readlines() + except OSError: + return None + for raw in lines: + try: + data = json.loads(raw) + except json.JSONDecodeError: + continue + if not isinstance(data, dict): + continue + for block in _tool_uses(data): + name = block.get("name") + if name in EDIT_TOOL_NAMES: + events.append(("edit", "")) + elif name in SHELL_TOOL_NAMES: + command = (block.get("input") or {}).get("command") + if isinstance(command, str): + events.append(("cmd", command)) + return events + + +def _session_paths(transcript_path: str) -> list[str]: + paths = [transcript_path] + parent = os.path.dirname(transcript_path) + if os.path.basename(parent) == "subagents": + session_dir = os.path.dirname(parent) + for name in os.listdir(session_dir): + candidate = os.path.join(session_dir, name) + if name.endswith(".jsonl") and os.path.isfile(candidate): + paths.append(candidate) + paths.extend( + os.path.join(parent, name) + for name in os.listdir(parent) + if name.endswith(".jsonl") + ) + else: + subagents = os.path.join(parent, "subagents") + if os.path.isdir(subagents): + paths.extend( + os.path.join(subagents, name) + for name in sorted(os.listdir(subagents)) + if name.endswith(".jsonl") + ) + seen: list[str] = [] + for path in paths: + if path not in seen: + seen.append(path) + return seen + + +def same_repro_gap(transcript_path: str) -> list[str] | None: + """Commands that showed the real program and were not run again after the + last edit. None when the transcript cannot be read (fail open).""" + if not transcript_path: + return [] + primary = _read_events(transcript_path) + if primary is None: + return None + last_edit = None + for index, (kind, _command) in enumerate(primary): + if kind == "edit": + last_edit = index + if last_edit is None: + return [] + if not any(kind == "cmd" and _is_test_runner(command) for index, (kind, command) in enumerate(primary) if index > last_edit): + return [] + seen: set[str] = set() + for path in _session_paths(transcript_path): + events = primary if path == transcript_path else _read_events(path) + if events is None: + continue + for kind, command in events: + if kind == "cmd": + seen.update(surface_keys(command)) + rerun: set[str] = set() + for index, (kind, command) in enumerate(primary): + if index > last_edit and kind == "cmd": + rerun.update(surface_keys(command)) + return sorted(seen - rerun) + + +def _same_repro_feedback(payload: dict[str, object]) -> str | None: + transcript_path = ( + payload.get("agent_transcript_path") + or payload.get("transcript_path") + or payload.get("transcriptPath") + or "" + ) + if not isinstance(transcript_path, str) or not transcript_path: + return None + missing = same_repro_gap(transcript_path) + if not missing: + return None + listed = ", ".join(f"`{command}`" for command in missing) + return ( + "prove-it-ship-gate: the real command that showed the behavior was " + f"{listed}. A later edit was checked with a test runner instead of that " + "command. Rerun the same program and subcommand and paste its output. " + "A vitest, pnpm test, or pytest run does not replace it." + ) + + def detect(event: dict[str, object]) -> list[Finding]: """Return findings for unproven live ship claims.""" + findings: list[Finding] = [] message = _blocking_feedback(event) - if not message: - return [] reply = event.get("last_assistant_message") or "" reply_text = reply if isinstance(reply, str) else "" - return [Finding( - rule_id=RULE_UNPROVEN_LIVE_CLAIM, - subject="reply:" + hashlib.sha256(reply_text.encode("utf-8")).hexdigest(), - message=message, - evidence=reply_text[:500], - )] + if message: + findings.append(Finding( + rule_id=RULE_UNPROVEN_LIVE_CLAIM, + subject="reply:" + hashlib.sha256(reply_text.encode("utf-8")).hexdigest(), + message=message, + evidence=reply_text[:500], + )) + same = _same_repro_feedback(event) + if same: + findings.append(Finding( + rule_id=RULE_SAME_REPRO, + subject="same-repro:" + hashlib.sha256(same.encode("utf-8")).hexdigest(), + message=same, + evidence=same[:500], + )) + return findings def decide(payload: dict) -> str | None: diff --git a/engine/hooks/prove-it-ship-gate/tests/test_hooks.py b/engine/hooks/prove-it-ship-gate/tests/test_hooks.py index 2c748d077..8beee6f3e 100644 --- a/engine/hooks/prove-it-ship-gate/tests/test_hooks.py +++ b/engine/hooks/prove-it-ship-gate/tests/test_hooks.py @@ -478,3 +478,131 @@ def test_silent_when_the_subagent_itself_ran_the_live_command(self): code, err = run_claude(self.payload(["npm test"], ["ssh do1 'systemctl status slack-manager'"])) self.assertEqual(code, 0) self.assertEqual(err, "") + + +def write_cursor_transcript(directory, name, events): + lines = [] + for kind, payload in events: + if kind == "shell": + block = {"type": "tool_use", "name": "Shell", "input": {"command": payload}} + elif kind == "edit": + block = {"type": "tool_use", "name": "StrReplace", "input": {"path": "packages/cli/src/onboarding.ts"}} + else: + raise AssertionError(kind) + lines.append(json.dumps({ + "role": "assistant", + "message": {"role": "assistant", "content": [block]}, + })) + path = os.path.join(directory, name) + with open(path, "w", encoding="utf-8") as handle: + handle.write("\n".join(lines) + "\n") + return path + + +class TestSameRepro(unittest.TestCase): + def test_surface_keys_keep_the_real_command_and_drop_scripts(self): + self.assertEqual(detect.surface_keys("invoker-cli --help"), ["invoker-cli --help"]) + self.assertEqual(detect.surface_keys("invoker-cli doctor"), ["invoker-cli doctor"]) + self.assertEqual( + detect.surface_keys( + 'python3 - << \'PY\'\n' + 'p = subprocess.Popen(["invoker-cli", "doctor"])\n' + 'text.encode("utf-8", "replace")\n' + ), + ["invoker-cli doctor"], + ) + self.assertEqual( + detect.surface_keys('which invoker-cli; invoker-cli --version'), + ["invoker-cli --version"], + ) + self.assertEqual( + detect.surface_keys("ps aux | rg -i 'invoker|owner-serve|electron'"), + [], + ) + self.assertEqual( + detect.surface_keys( + 'node --input-type=module -e \'console.log("cli-runtime")\'' + ), + [], + ) + self.assertEqual( + detect.surface_keys( + "pnpm --filter @invoker/cli exec vitest run src/__tests__/cli.test.ts" + ), + [], + ) + + def test_fires_when_a_test_runner_replaces_the_real_command(self): + with tempfile.TemporaryDirectory() as directory: + path = write_cursor_transcript(directory, "session.jsonl", [ + ("shell", "invoker-cli --help"), + ("shell", "invoker-cli doctor"), + ("edit", ""), + ("shell", "pnpm --filter @invoker/cli exec vitest run src/__tests__/cli.test.ts"), + ]) + feedback = detect.decide({ + "last_assistant_message": "vitest passed", + "transcript_path": path, + }) + self.assertIsNotNone(feedback) + self.assertIn("invoker-cli --help", feedback) + self.assertIn("invoker-cli doctor", feedback) + + def test_silent_when_the_same_command_is_rerun_after_the_edit(self): + with tempfile.TemporaryDirectory() as directory: + path = write_cursor_transcript(directory, "session.jsonl", [ + ("shell", "invoker-cli doctor"), + ("edit", ""), + ("shell", "pnpm --filter @invoker/cli exec vitest run src/__tests__/cli.test.ts"), + ("shell", "invoker-cli doctor"), + ]) + self.assertIsNone(detect.decide({ + "last_assistant_message": "doctor now prints the agent", + "transcript_path": path, + })) + + def test_silent_on_a_test_only_edit(self): + with tempfile.TemporaryDirectory() as directory: + path = write_cursor_transcript(directory, "session.jsonl", [ + ("edit", ""), + ("shell", "pnpm test"), + ]) + self.assertIsNone(detect.decide({ + "last_assistant_message": "pnpm test passed", + "transcript_path": path, + })) + + def test_silent_when_the_real_command_was_not_followed_by_a_test(self): + with tempfile.TemporaryDirectory() as directory: + path = write_cursor_transcript(directory, "session.jsonl", [ + ("shell", "invoker-cli doctor"), + ("edit", ""), + ]) + self.assertIsNone(detect.decide({ + "last_assistant_message": "edited the doctor check", + "transcript_path": path, + })) + + def test_fires_when_the_real_command_ran_in_a_subagent(self): + with tempfile.TemporaryDirectory() as directory: + os.makedirs(os.path.join(directory, "subagents")) + path = write_cursor_transcript(directory, "session.jsonl", [ + ("edit", ""), + ("shell", "pnpm --filter @invoker/cli exec vitest run src/__tests__/cli.test.ts"), + ]) + write_cursor_transcript(os.path.join(directory, "subagents"), "agent.jsonl", [ + ("shell", "invoker-cli --help"), + ]) + feedback = detect.decide({ + "last_assistant_message": "the test passed", + "transcript_path": path, + }) + self.assertIsNotNone(feedback) + self.assertIn("invoker-cli --help", feedback) + + def test_fails_open_when_the_same_repro_transcript_is_missing(self): + self.assertIsNone(detect.same_repro_gap("/nonexistent/same-repro.jsonl")) + self.assertIsNone(detect.decide({ + "last_assistant_message": "vitest passed", + "transcript_path": "/nonexistent/same-repro.jsonl", + }))