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
1 change: 1 addition & 0 deletions docs/ecosystem.md
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,7 @@ again.
| `agent-relay-attribution` | hook (advisory) |
| `scratchpad-collision` | hook |
| `ui-input-guard` | hook |
| `handoff-needs-smoke-test` | hook |
| `hook-freshness` | hook (advisory) |
| `llm-judge` | hook (shared background model judge; its inbox delivers finished verdicts on the next turn: Claude `UserPromptSubmit`, Cursor `stop`, Codex `notify`) |
| `engine/CLAUDE.core.md` | global hand-written Claude rules |
Expand Down
37 changes: 37 additions & 0 deletions engine/hooks/handoff-needs-smoke-test/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,37 @@
# handoff-needs-smoke-test

Stop hook: a script handed to the user is a claim that it runs.

Fires when the outgoing reply asks the user to execute a script — `! bash
<path>` on its own line, or the same form inside inline code — and this
session's transcript shows no Bash call that ran that path through an
interpreter. Writing the file, `chmod`, `scp`, and `cat` do not count;
`bash <path>`, `sh`, `zsh`, `python3`, `node`, and `source` do.

Silent on: a handoff whose script this session already ran; a reply that
names why the run cannot happen here ("cannot run it here: the sign-in
needs your browser", "only you can approve it"); a command shown for
reference without the `!` handoff form; a one-off `gh` or `git` command
that is not a script path.

Block message names the unrun script and the two ways out: run it end to
end, or run the same transport with a harmless payload first. The escape is
naming the blocker, not omitting it.

## Fail direction

Fails open on every read it cannot complete: no `transcript_path`, an
unreadable or malformed transcript, or any detector error. A Stop hook that
cannot see the transcript cannot tell a tested handoff from an untested
one, and blocking every reply on an unreadable file would wedge the
session. `stop_hook_active` also returns early so the rewritten turn can
finish.

## Files

- `detect.py` — handoff shapes, the blocker vocabulary, transcript scan, `decide()`.
- `claude_stop_check.py` — Claude Stop entrypoint.
- `claude.hook.json` / `install_claude_hook.py` — settings.json merge (idempotent).
- `tests/fixtures/handoffs_{fires,silent}.json` — the verbatim reply that
motivated this, plus its near-neighbours.
- `tests/test_hooks.py`
16 changes: 16 additions & 0 deletions engine/hooks/handoff-needs-smoke-test/claude.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/.claude/hooks/handoff-needs-smoke-test/claude_stop_check.py",
"timeout": 10
}
]
}
]
}
}
31 changes: 31 additions & 0 deletions engine/hooks/handoff-needs-smoke-test/claude_stop_check.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
#!/usr/bin/env python3
"""Claude Code Stop hook: block a reply that hands the user a script this
session never ran, unless the reply names why the run cannot happen here.
Fails open on read or parse errors; `stop_hook_active` skips.
"""
from __future__ import annotations

import json
import sys

from detect import decide


def main() -> 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"handoff-needs-smoke-test: detector error, allowing this reply: {exc!r}\n")
return
if not message:
return
sys.stderr.write(message + "\n")
sys.exit(2)


if __name__ == "__main__":
main()
132 changes: 132 additions & 0 deletions engine/hooks/handoff-needs-smoke-test/detect.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,132 @@
"""handoff-needs-smoke-test: a script handed to the user is a claim it runs.

A reply that ends with `! bash <path>` is asking the user to execute
something on their own machine. If this session never executed that path,
nobody has: a syntax check of the wrapper does not run the payload, and a
script assembled from nested quoting can collapse into a single line that
parses locally and breaks remotely.

The escape is the honest one: say the run cannot happen here and why. An
interactive browser login is a real reason; not having tried is not.
"""
from __future__ import annotations

import json
import os
import re

HANDOFF_RES = [
re.compile(r"(?m)^\s*!\s*(?:bash|sh|zsh|python3?|node)\s+(\S+)"),
re.compile(r"`\s*!\s*(?:bash|sh|zsh|python3?|node)\s+(\S+)\s*`"),
]
SCRIPT_SUFFIXES = (".sh", ".bash", ".zsh", ".py", ".mjs", ".js")

CANNOT_RUN_RE = re.compile(
r"\b(?:cannot|can'?t|could not|couldn'?t|unable to|no way to)\b[^.\n]{0,80}"
r"\b(?:run|execute|test|try|verify|reach|reproduce)\b"
r"|\brequires? (?:your|a human|physical|interactive|browser)\b"
r"|\bonly you can\b|\bneeds your browser\b|\binteractive (?:login|consent|approval)\b",
re.IGNORECASE,
)

RUN_PREFIX_RE = re.compile(
r"(?:^|[\s;|&(])(?:bash|sh|zsh|python3?|node|source|\.)\s+(\S+)"
)
WRITE_ONLY_RE = re.compile(
r"(?:^|[\s;|&(])(?:cat|tee|chmod|cp|mv|scp|rsync|ls|stat|rm|touch|head|tail|wc|grep|rg)\b"
)

VERIFY_TOOLS = {"Bash"}

MESSAGE = (
"handoff-needs-smoke-test: this reply hands over {targets} with `!`, and this "
"session never executed {that}. A local syntax check does not run a remote "
"payload, and nested quoting can collapse a multi-line script into one line "
"that parses here and breaks there. Run it end to end, or run the same "
"transport with a harmless payload, before handing it over. If the run "
"genuinely cannot happen here -- an interactive browser login, a credential "
"only the user holds -- say so in the reply and name the blocker."
)


def handoff_paths(message):
"""Script paths the reply asks the user to run."""
found = []
for pattern in HANDOFF_RES:
for match in pattern.finditer(message or ""):
path = match.group(1).strip("`'\"")
if path.endswith(SCRIPT_SUFFIXES) and path not in found:
found.append(path)
return found


def names_a_blocker(message):
return bool(CANNOT_RUN_RE.search(message or ""))


def _executed_paths(lines):
"""Paths this session actually ran through an interpreter."""
ran = set()
for data in lines:
if data.get("type") != "assistant":
continue
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_use":
continue
if block.get("name") not in VERIFY_TOOLS:
continue
command = (block.get("input") or {}).get("command") or ""
for segment in re.split(r"\|\||&&|[|;\n]", command):
segment = segment.strip()
if not segment or WRITE_ONLY_RE.match(segment):
continue
match = RUN_PREFIX_RE.search(segment)
if match:
ran.add(os.path.basename(match.group(1).strip("`'\"")))
return ran


def parse_lines(raw_lines):
parsed = []
for raw in raw_lines:
try:
data = json.loads(raw)
except (json.JSONDecodeError, TypeError):
continue
if isinstance(data, dict):
parsed.append(data)
return parsed


def decide_from_lines(message, lines):
targets = handoff_paths(message)
if not targets or names_a_blocker(message):
return None
ran = _executed_paths(lines)
unrun = [t for t in targets if os.path.basename(t) not in ran]
if not unrun:
return None
names = ", ".join(f"`{os.path.basename(t)}`" for t in unrun)
return MESSAGE.format(targets=names, that="it" if len(unrun) == 1 else "them")


def decide(payload):
"""Blocking feedback for the Stop event, or None to let the turn finish."""
if payload.get("stop_hook_active"):
return None
message = payload.get("last_assistant_message") or ""
if not handoff_paths(message) or names_a_blocker(message):
return None
path = payload.get("transcript_path") or payload.get("transcriptPath") or ""
if not path:
return None
try:
with open(path, encoding="utf-8") as handle:
lines = parse_lines(handle)
except OSError:
return None
return decide_from_lines(message, lines)
46 changes: 46 additions & 0 deletions engine/hooks/handoff-needs-smoke-test/install_claude_hook.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
#!/usr/bin/env python3
"""Merge handoff-needs-smoke-test into ~/.claude/settings.json Stop 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 = "handoff-needs-smoke-test/claude_stop_check.py"
EVENT = "Stop"


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 Stop handoff-needs-smoke-test 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 Stop handoff-needs-smoke-test")


if __name__ == "__main__":
main()
Original file line number Diff line number Diff line change
@@ -0,0 +1,24 @@
[
{
"label": "the real reply that shipped a broken login script (verbatim)",
"reply": "Run this one line, and I will verify the result afterwards:\n\n```\n! bash /private/tmp/claude-501/scratchpad/demo-login.sh\n```\n\nOne SSH session as the `demo` user, two sign-ins. Codex uses a device code.",
"ran": []
},
{
"label": "inline handoff in prose",
"reply": "Everything is staged. Run this: `! bash scripts/provision-worker.sh` and paste what it prints.",
"ran": []
},
{
"label": "two scripts handed over, only one of them run",
"reply": "Two steps:\n\n! bash /tmp/stage-one.sh\n\nthen\n\n! bash /tmp/stage-two.sh\n",
"ran": [
"stage-one.sh"
]
},
{
"label": "python handoff",
"reply": "! python3 /tmp/repair-queue.py\n\nIt drains the queue and prints a count.",
"ran": []
}
]
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
[
{
"label": "the same handoff after the script was actually run this session",
"reply": "Run this one line:\n\n```\n! bash /private/tmp/claude-501/scratchpad/demo-login.sh\n```\n",
"ran": [
"demo-login.sh"
]
},
{
"label": "handoff whose blocker is named (interactive browser login)",
"reply": "! bash /tmp/demo-login.sh\n\nI cannot run it here: the sign-in needs your browser, so I ran the same transport with a harmless payload instead.",
"ran": []
},
{
"label": "a command shown for reference, not handed over",
"reply": "CI runs `bash scripts/ci-entry.sh` on every push, which is where the failure comes from.",
"ran": []
},
{
"label": "a one-off gh command, not a script",
"reply": "! gh pr merge 12030 --squash",
"ran": []
},
{
"label": "only you can do it, stated plainly",
"reply": "! bash /tmp/rotate-keys.sh\n\nOnly you can approve the rotation in the console, so this has to run from your side.",
"ran": []
}
]
Loading
Loading