diff --git a/engine/hooks/gh-write-verification/detect.py b/engine/hooks/gh-write-verification/detect.py index 47b73b81..7a287210 100644 --- a/engine/hooks/gh-write-verification/detect.py +++ b/engine/hooks/gh-write-verification/detect.py @@ -213,6 +213,7 @@ def self_match_message(hits: list[str]) -> str: r"|\bgit\s+branch\s+-r\s+--contains\b" r"|\bgit\s+branch\s+--contains\b[^\n]*\s-r\b" ) +LANDING_OK_RE = re.compile(r"(?m)^\s*OK:") VERIFY_SCRIPT_RELPATH = "gh-write-verification/verify_pr_landed_on_trunk.sh" UNVERIFIED_MERGE_MESSAGE = ( @@ -288,22 +289,33 @@ def silenced_mutations(raw_text: str) -> list[str]: return hits -def _proves_landing(command: str, number: str | None) -> bool: +CommandRecord = tuple[str, str | None] + + +def _proves_landing(command: str, number: str | None, result: str | None) -> bool: """True when this command checks where a merge commit actually landed. An invocation of the shipped verification script must name the PR it is - vouching for; a hand-rolled ancestry check is accepted as written, since - it takes a commit sha rather than a PR number. + vouching for, and the paired tool result must report the passing verdict. """ command = command or "" if not LANDING_PROOF_RE.search(command): return False if "verify_pr_landed_on_trunk" in command and number is not None: - return number in command - return True + if number not in command: + return False + return bool(result and LANDING_OK_RE.search(result)) + + +def _command_text(record: str | CommandRecord) -> str: + return record[0] if isinstance(record, tuple) else record + + +def _command_result(record: str | CommandRecord) -> str | None: + return record[1] if isinstance(record, tuple) else None -def merges_missing_landing_proof(commands: list[str]) -> list[str]: +def merges_missing_landing_proof(commands: list[str | CommandRecord]) -> list[str]: """PR subjects merged in this turn with no landing check run afterwards. Returns the merged subjects (a PR number, or "the current branch's PR" @@ -312,13 +324,17 @@ def merges_missing_landing_proof(commands: list[str]) -> list[str]: proof ran after the merge. """ subjects: list[str] = [] - for index, command in enumerate(commands): + for index, record in enumerate(commands): + command = _command_text(record) match = GH_PR_MERGE_RE.search(command or "") if not match: continue number = match.group("number") subject = f"PR #{number}" if number else "the current branch's PR" - if any(_proves_landing(later, number) for later in commands[index + 1:]): + if any( + _proves_landing(_command_text(later), number, _command_result(later)) + for later in commands[index + 1:] + ): continue if subject not in subjects: subjects.append(subject) @@ -359,6 +375,17 @@ def _text_content(data: dict) -> str: return "" +def _tool_result_text(block: dict) -> str: + content = block.get("content") + if isinstance(content, str): + return content + if isinstance(content, list): + return "\n".join( + b.get("text", "") for b in content if isinstance(b, dict) and b.get("type") == "text" + ) + return "" + + def _is_human_user_line(data: dict) -> bool: if data.get("type") != "user": return False @@ -366,7 +393,7 @@ def _is_human_user_line(data: dict) -> bool: return bool(text.strip()) and not text.lstrip().startswith("<") -def bash_commands_this_turn(raw_lines) -> list[str]: +def bash_commands_this_turn(raw_lines) -> list[CommandRecord]: """Bash tool commands issued since the last authored user message.""" parsed: list[dict] = [] for raw in raw_lines: @@ -380,8 +407,21 @@ def bash_commands_this_turn(raw_lines) -> list[str]: for index, data in enumerate(parsed): if _is_human_user_line(data): turn_start = index - commands: list[str] = [] - for data in parsed[turn_start:]: + records = parsed[turn_start:] + results: dict[str, str] = {} + for data in records: + message = data.get("message") + content = message.get("content") if isinstance(message, dict) else None + if not isinstance(content, list): + continue + for block in content: + if not isinstance(block, dict) or block.get("type") != "tool_result": + continue + tool_id = block.get("tool_use_id") + if tool_id: + results[str(tool_id)] = _tool_result_text(block) + commands: list[CommandRecord] = [] + for data in records: if data.get("type") != "assistant": continue message = data.get("message") @@ -395,7 +435,9 @@ def bash_commands_this_turn(raw_lines) -> list[str]: continue tool_input = block.get("input") if isinstance(tool_input, dict): - commands.append(str(tool_input.get("command") or "")) + tool_id = block.get("id") or block.get("tool_use_id") + result = results.get(str(tool_id)) if tool_id else None + commands.append((str(tool_input.get("command") or ""), result)) return commands diff --git a/engine/hooks/gh-write-verification/tests/test_hooks.py b/engine/hooks/gh-write-verification/tests/test_hooks.py index c329b15a..3e6a7f6e 100644 --- a/engine/hooks/gh-write-verification/tests/test_hooks.py +++ b/engine/hooks/gh-write-verification/tests/test_hooks.py @@ -57,15 +57,24 @@ def bash_payload(command: str) -> dict: return {"tool_name": "Bash", "cwd": HOOK_DIR, "tool_input": {"command": command}} -def transcript(commands: list[str]) -> str: +def transcript(commands: list[str | tuple[str, str | None]]) -> str: lines = [json.dumps({"type": "user", "message": {"role": "user", "content": "land the stack"}})] - for command in commands: + for index, item in enumerate(commands): + command, result = item if isinstance(item, tuple) else (item, None) + tool_id = f"bash-{index}" lines.append(json.dumps({ "type": "assistant", "message": {"content": [ - {"type": "tool_use", "name": "Bash", "input": {"command": command}} + {"type": "tool_use", "id": tool_id, "name": "Bash", "input": {"command": command}} ]}, })) + if result is not None: + lines.append(json.dumps({ + "type": "user", + "message": {"content": [ + {"type": "tool_result", "tool_use_id": tool_id, "content": result} + ]}, + })) handle = tempfile.NamedTemporaryFile("w", suffix=".jsonl", delete=False) handle.write("\n".join(lines) + "\n") handle.close() @@ -228,6 +237,9 @@ def test_entrypoint_denies_the_incident_wait_loop(self): class TestUnverifiedLanding(unittest.TestCase): + def proven(self, command: str) -> tuple[str, str]: + return (command, "OK: abc123 is an ancestor of origin/main\n") + def test_a_merge_with_no_landing_check_is_flagged(self): self.assertEqual( merges_missing_landing_proof(["gh pr merge 291 --squash --admin", "gh pr view 291"]), @@ -238,7 +250,7 @@ def test_every_unproven_merge_in_the_turn_is_flagged(self): commands = [ "gh pr merge 291 --squash", "gh pr merge 292 --squash", - "bash verify_pr_landed_on_trunk.sh 292", + self.proven("bash verify_pr_landed_on_trunk.sh 292"), ] self.assertEqual(merges_missing_landing_proof(commands), ["PR #291"]) @@ -251,9 +263,18 @@ def test_a_verified_landing_stays_silent(self): "git branch -r --contains 314f0447", ): self.assertEqual( - merges_missing_landing_proof(["gh pr merge 291 --squash", proof]), [], proof + merges_missing_landing_proof(["gh pr merge 291 --squash", self.proven(proof)]), [], proof ) + def test_a_landing_check_without_a_result_is_flagged(self): + self.assertEqual( + merges_missing_landing_proof([ + "gh pr merge 291 --squash", + "bash verify_pr_landed_on_trunk.sh 291", + ]), + ["PR #291"], + ) + def test_a_turn_with_no_merge_stays_silent(self): self.assertEqual(merges_missing_landing_proof(["git status", "gh pr view 291"]), []) @@ -271,7 +292,11 @@ def test_stop_entrypoint_denies_an_unproven_merge(self): def test_stop_entrypoint_allows_a_proven_merge(self): path = transcript([ "gh pr merge 291 --squash --admin", - 'bash "$HOME/.claude/hooks/gh-write-verification/verify_pr_landed_on_trunk.sh" 291', + ( + 'bash "$HOME/.claude/hooks/gh-write-verification/verify_pr_landed_on_trunk.sh" 291', + "pr=#291 repo=acme/widgets merged=true base=main merge_commit=abc123\n" + "OK: abc123 is an ancestor of origin/main\n", + ), ]) try: result = run_entrypoint(STOP_CHECK, {"transcript_path": path}) @@ -279,6 +304,37 @@ def test_stop_entrypoint_allows_a_proven_merge(self): finally: os.unlink(path) + def test_stop_entrypoint_denies_a_failed_landing_check(self): + path = transcript([ + "gh pr merge 291 --squash --admin", + ( + 'bash "$HOME/.claude/hooks/gh-write-verification/verify_pr_landed_on_trunk.sh" 291', + "pr=#291 repo=acme/widgets merged=true base=stack merge_commit=abc123\n" + "FAIL: PR #291 in acme/widgets reports MERGED but abc123 is not on origin/main\n", + ), + ]) + try: + result = run_entrypoint(STOP_CHECK, {"transcript_path": path}) + self.assertEqual(result.returncode, 2, result.stderr) + self.assertIn("PR #291", result.stderr) + finally: + os.unlink(path) + + def test_stop_entrypoint_denies_an_unchecked_landing_check(self): + path = transcript([ + "gh pr merge 291 --squash --admin", + ( + 'bash "$HOME/.claude/hooks/gh-write-verification/verify_pr_landed_on_trunk.sh" 291', + "UNCHECKED: gh cannot resolve a repository here\n", + ), + ]) + try: + result = run_entrypoint(STOP_CHECK, {"transcript_path": path}) + self.assertEqual(result.returncode, 2, result.stderr) + self.assertIn("PR #291", result.stderr) + finally: + os.unlink(path) + def test_a_missing_transcript_fails_open(self): self.assertIsNone(decide_stop({"transcript_path": "/nonexistent/transcript.jsonl"})) self.assertIsNone(decide_stop({}))