Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions engine/hooks/prove-it-ship-gate/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -50,9 +50,13 @@ 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
The stop is wired for Claude, Cursor, and Codex. Each harness gets that
harness's own block: Claude exits 2, Cursor returns a followup message on
its stop event, and Codex returns a block decision. 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` /
to run again after the last file edit. Codex `exec_command` rows count as
that command, and `apply_patch` counts as the 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.
Expand Down
16 changes: 16 additions & 0 deletions engine/hooks/prove-it-ship-gate/codex.hook.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,16 @@
{
"hooks": {
"Stop": [
{
"matcher": ".*",
"hooks": [
{
"type": "command",
"command": "python3 $HOME/.codex/hooks/prove-it-ship-gate/codex_stop_check.py",
"timeout": 10
}
]
}
]
}
}
20 changes: 20 additions & 0 deletions engine/hooks/prove-it-ship-gate/codex_stop_check.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
#!/usr/bin/env python3
"""Codex 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", "codex", detect, "Stop", json_error_stderr=False)


if __name__ == "__main__":
main()
2 changes: 1 addition & 1 deletion engine/hooks/prove-it-ship-gate/cursor_stop_check.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@


def main() -> None:
run_hook("prove-it-ship-gate", "cursor", detect)
run_hook("prove-it-ship-gate", "cursor", detect, "stop")


if __name__ == "__main__":
Expand Down
60 changes: 45 additions & 15 deletions engine/hooks/prove-it-ship-gate/detect.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,8 +46,14 @@
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"})
SHELL_TOOL_NAMES = frozenset({
"Bash", "bash", "Shell", "shell", "Exec", "exec", "exec_command",
"functions.exec_command", "local_shell", "shell_call",
})
EDIT_TOOL_NAMES = frozenset({
"Write", "StrReplace", "Edit", "EditNotebook", "ApplyPatch", "apply_patch",
"MultiEdit", "NotebookEdit", "edit_file",
})
COMMAND_PREFIXES = frozenset({"sudo", "env", "time", "command", "nohup", "nice"})
GENERIC_PROGRAMS = frozenset({
"git", "node", "python", "python3", "bash", "zsh", "sh", "ls", "cat", "rg",
Expand Down Expand Up @@ -211,24 +217,48 @@ def bash_commands_this_turn(transcript_path: str) -> list[str] | None:
continue
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):
command = _shell_command(block)
if command is not None:
commands.append(command)
return commands


def _shell_command(block: dict) -> str | None:
raw = block.get("input")
if not isinstance(raw, dict):
return None
for key in ("command", "cmd"):
value = raw.get(key)
if isinstance(value, str):
return value
return None


def _tool_uses(data: dict) -> list[dict]:
found: 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"
]
if role == "assistant":
content = message.get("content") if message else data.get("content")
if isinstance(content, list):
found.extend(
block for block in content
if isinstance(block, dict) and block.get("type") == "tool_use"
)
payload = data.get("payload") if isinstance(data.get("payload"), dict) else data
if isinstance(payload, dict) and payload.get("type") in {"function_call", "custom_tool_call"}:
name = payload.get("name")
raw = payload.get("arguments")
if raw is None:
raw = payload.get("input")
if isinstance(raw, str):
try:
raw = json.loads(raw)
except json.JSONDecodeError:
raw = {}
if isinstance(name, str) and isinstance(raw, dict):
found.append({"type": "tool_use", "name": name, "input": raw})
return found


def _unquoted_segments(command: str) -> list[str]:
Expand Down Expand Up @@ -323,8 +353,8 @@ def _read_events(path: str) -> list[tuple[str, str]] | None:
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):
command = _shell_command(block)
if command is not None:
events.append(("cmd", command))
return events

Expand Down
89 changes: 89 additions & 0 deletions engine/hooks/prove-it-ship-gate/tests/test_hooks.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
import io
import json
import os
import subprocess
import sys
import tempfile
import unittest
Expand Down Expand Up @@ -606,3 +607,91 @@ def test_fails_open_when_the_same_repro_transcript_is_missing(self):
"last_assistant_message": "vitest passed",
"transcript_path": "/nonexistent/same-repro.jsonl",
}))

def test_fires_on_a_codex_exec_and_patch(self):
with tempfile.TemporaryDirectory() as directory:
path = os.path.join(directory, "session.jsonl")
rows = [
{
"type": "function_call",
"name": "exec_command",
"arguments": json.dumps({"cmd": "invoker-cli doctor"}),
},
{
"type": "function_call",
"name": "apply_patch",
"arguments": json.dumps({"command": ["apply_patch", "*** Begin Patch"]}),
},
{
"type": "function_call",
"name": "exec_command",
"arguments": json.dumps({"cmd": "pnpm test"}),
},
]
with open(path, "w", encoding="utf-8") as handle:
handle.write("\n".join(json.dumps(row) for row in rows) + "\n")
feedback = detect.decide({
"last_assistant_message": "the test passed",
"transcript_path": path,
})
self.assertIsNotNone(feedback)
self.assertIn("invoker-cli doctor", feedback)

def test_the_same_finding_blocks_claude_cursor_and_codex(self):
hook_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
with tempfile.TemporaryDirectory() as directory:
path = os.path.join(directory, "session.jsonl")
rows = [
{
"type": "function_call",
"name": "exec_command",
"arguments": json.dumps({"cmd": "invoker-cli doctor"}),
},
{
"type": "function_call",
"name": "apply_patch",
"arguments": "{}",
},
{
"type": "function_call",
"name": "exec_command",
"arguments": json.dumps({"cmd": "pnpm test"}),
},
]
with open(path, "w", encoding="utf-8") as handle:
handle.write("\n".join(json.dumps(row) for row in rows) + "\n")
env = os.environ.copy()
env["CATSTACK_HOOK_TEST_MODEL"] = "gpt-5.6-sol"
env["CATSTACK_HOOK_METRICS_DIR"] = os.path.join(directory, "metrics")
cases = [
("claude_stop_check.py", {"hook_event_name": "Stop"}, "claude"),
("cursor_stop_check.py", {"hook_event_name": "stop"}, "cursor"),
("codex_stop_check.py", {"hook_event_name": "Stop"}, "codex"),
]
for script, extra, harness in cases:
payload = {
"last_assistant_message": "the test passed",
"transcript_path": path,
**extra,
}
result = subprocess.run(
[sys.executable, os.path.join(hook_dir, script)],
input=json.dumps(payload),
capture_output=True,
text=True,
env=env,
check=False,
)
with self.subTest(harness=harness):
if harness == "claude":
self.assertEqual(2, result.returncode, result.stderr)
self.assertIn("invoker-cli doctor", result.stderr)
elif harness == "cursor":
self.assertEqual(0, result.returncode, result.stderr)
body = json.loads(result.stdout)
self.assertIn("invoker-cli doctor", body["followup_message"])
else:
self.assertEqual(0, result.returncode, result.stderr)
body = json.loads(result.stdout)
self.assertEqual("block", body["decision"])
self.assertIn("invoker-cli doctor", body["reason"])
Loading