Gabriel Vidal
d116e6571f
The CLI prefixes a Stop hook's reason with "Stop hook feedback: " itself, so the relay reason started with it twice; the parser tolerates the already-recorded doubles. /api/resume now answers delivered: "hold" when the sidecar left the message for the guard instead of the inbox socket. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
379 lines
14 KiB
Python
379 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Claude Code ``Stop`` hook for headless runs: hold the turn while a plain
|
|
background shell is still running.
|
|
|
|
Why this exists. A ``claude -p`` process ends when the model ends its turn.
|
|
Before exiting it waits for *some* background work — subagents and Monitors,
|
|
up to ``CLAUDE_CODE_PRINT_BG_WAIT_CEILING_MS`` — but a plain ``Bash
|
|
run_in_background`` task is explicitly excluded from that wait (the CLI's
|
|
``local_bash`` task type), so the moment the model says "I'll pick this up
|
|
when the build reports back" the process exits and the build is killed. The
|
|
next ``--resume`` then opens with a ``didn't finish before the previous
|
|
session ended`` notification and the conversation looks stuck in between.
|
|
|
|
What it does. The hook gets the session's ``background_tasks`` on stdin. If any
|
|
``running`` task of type ``shell`` is a plain background shell (not a Monitor
|
|
— those the CLI waits for itself), it polls the task's output file until the
|
|
CLI appends its ``[exited with code N]`` marker, then answers
|
|
``{"decision": "block", "reason": …}`` telling the model its task finished and
|
|
where its output is. While the hook holds the stop the process stays alive and
|
|
the model makes no API call. A task still running at ``MAX_WAIT_S`` is
|
|
reported as such and the model is asked to either stop it (TaskStop) or end
|
|
its turn again, which re-arms the guard.
|
|
|
|
The sidecar injects it into every run through ``--settings`` (see
|
|
``_claude_settings`` in sidecar.py); stdlib only, so it runs under whatever
|
|
Python the sidecar runs under, host or Mac worker.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import sys
|
|
import time
|
|
|
|
# How long one hook invocation holds the stop before handing the question back
|
|
# to the model. Under the hook's own timeout (HOOK_TIMEOUT_S in sidecar.py).
|
|
MAX_WAIT_S = float(os.environ.get("STOP_GUARD_MAX_WAIT_S", "3300"))
|
|
POLL_S = float(os.environ.get("STOP_GUARD_POLL_S", "2"))
|
|
|
|
# The sidecar's state dir for this run (its logs/ dir, exported into the run's
|
|
# env as STOP_GUARD_STATE_DIR). While the guard holds a turn it keeps
|
|
# ``<sid>.hold`` there, and the sidecar's /message — which can't reach a
|
|
# session that is inside a Stop hook through the inbox socket (the socket
|
|
# accepts and the CLI drops it) — leaves the user's text in ``<sid>.inbox/``
|
|
# instead; the guard relays it in its block reason. The guard's own lines go
|
|
# to ``<sid>.guard.log`` next to the run log (the CLI keeps a hook's stderr).
|
|
STATE_DIR = os.environ.get("STOP_GUARD_STATE_DIR") or None
|
|
SESSION_ID: str | None = None
|
|
LOG_FILE: pathlib.Path | None = None
|
|
|
|
|
|
def _state_path(suffix: str) -> pathlib.Path | None:
|
|
if not STATE_DIR or not SESSION_ID:
|
|
return None
|
|
return pathlib.Path(STATE_DIR) / f"{SESSION_ID}{suffix}"
|
|
|
|
|
|
def mark_hold(tasks: list[dict]) -> None:
|
|
p = _state_path(".hold")
|
|
if p is None:
|
|
return
|
|
try:
|
|
p.write_text(json.dumps({"since": time.time(),
|
|
"tasks": [t.get("id") for t in tasks]}))
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def clear_hold() -> None:
|
|
p = _state_path(".hold")
|
|
if p is not None:
|
|
p.unlink(missing_ok=True)
|
|
|
|
|
|
def take_mail() -> list[str]:
|
|
"""The user's messages the sidecar left for this hold, oldest first —
|
|
consumed (deleted) as they are read."""
|
|
d = _state_path(".inbox")
|
|
if d is None or not d.is_dir():
|
|
return []
|
|
out = []
|
|
for f in sorted(d.iterdir()):
|
|
try:
|
|
o = json.loads(f.read_text())
|
|
text = (o.get("prompt") or "").strip() if isinstance(o, dict) else ""
|
|
except (OSError, ValueError):
|
|
text = ""
|
|
f.unlink(missing_ok=True)
|
|
if text:
|
|
out.append(text)
|
|
return out
|
|
|
|
|
|
def _log(msg: str) -> None:
|
|
line = f"stop-guard: {msg}"
|
|
print(line, file=sys.stderr, flush=True)
|
|
if LOG_FILE is not None:
|
|
try:
|
|
with open(LOG_FILE, "a", encoding="utf-8") as f:
|
|
f.write(time.strftime("%Y-%m-%dT%H:%M:%S ") + line + "\n")
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
# The CLI appends this to a background shell's output file when it exits.
|
|
EXIT_MARK = re.compile(rb"\[exited with code -?\d+\]\s*$")
|
|
# A Monitor's launch result; its task id is what background_tasks lists.
|
|
MONITOR_STARTED = re.compile(r"Monitor started \(task ([A-Za-z0-9_-]+)")
|
|
# A background shell's launch result ("Command running in background with ID:
|
|
# b123. Output is being written to: /tmp/…/tasks/b123.output" — or the
|
|
# "moved to the background (ID: b123)" variant of a foreground timeout).
|
|
OUTPUT_PATH = re.compile(r"Output is being written to: (\S+?\.output)")
|
|
|
|
|
|
def running_shells(tasks: list[dict]) -> list[dict]:
|
|
"""The ``background_tasks`` entries that are shells still running."""
|
|
return [t for t in tasks or []
|
|
if t.get("type") == "shell" and t.get("status") == "running"
|
|
and t.get("id")]
|
|
|
|
|
|
def _result_texts(line: str):
|
|
"""Every tool_result text in one transcript record (cheap pre-filter: the
|
|
caller only passes lines that contain a marker substring)."""
|
|
try:
|
|
o = json.loads(line)
|
|
except ValueError:
|
|
return
|
|
c = (o.get("message") or {}).get("content")
|
|
if not isinstance(c, list):
|
|
return
|
|
for b in c:
|
|
if not isinstance(b, dict) or b.get("type") != "tool_result":
|
|
continue
|
|
inner = b.get("content")
|
|
if isinstance(inner, str):
|
|
yield inner
|
|
elif isinstance(inner, list):
|
|
for x in inner:
|
|
if isinstance(x, dict) and x.get("type") == "text":
|
|
yield x.get("text") or ""
|
|
|
|
|
|
def scan_transcripts(transcript_path: str | None) -> tuple[set[str], dict[str, str]]:
|
|
"""Read the session's transcript (and its subagents' — a subagent's
|
|
background tasks show up in the parent's ``background_tasks`` too) for
|
|
the ids of Monitors and the output paths of background shells.
|
|
|
|
Returns ``(monitor_ids, {shell_id: output_path})``."""
|
|
monitors: set[str] = set()
|
|
outputs: dict[str, str] = {}
|
|
if not transcript_path:
|
|
return monitors, outputs
|
|
main = pathlib.Path(transcript_path)
|
|
files = [main]
|
|
sub = main.with_suffix("") / "subagents"
|
|
if sub.is_dir():
|
|
files += sorted(sub.glob("agent-*.jsonl"))
|
|
for f in files:
|
|
try:
|
|
with open(f, encoding="utf-8", errors="replace") as fh:
|
|
for line in fh:
|
|
if "Monitor started (task" not in line \
|
|
and "Output is being written to:" not in line:
|
|
continue
|
|
for text in _result_texts(line):
|
|
for m in MONITOR_STARTED.finditer(text):
|
|
monitors.add(m.group(1))
|
|
for m in OUTPUT_PATH.finditer(text):
|
|
path = m.group(1)
|
|
outputs[pathlib.Path(path).stem] = path
|
|
except OSError:
|
|
continue
|
|
return monitors, outputs
|
|
|
|
|
|
def tasks_dir(hook_input: dict, outputs: dict[str, str]) -> pathlib.Path | None:
|
|
"""Where this session's ``<task id>.output`` files live: next to the
|
|
scratchpad the hook input names, else the parent of any output path the
|
|
transcript mentioned."""
|
|
sp = hook_input.get("scratchpad_dir")
|
|
if sp:
|
|
d = pathlib.Path(sp).parent / "tasks"
|
|
if d.is_dir():
|
|
return d
|
|
for p in outputs.values():
|
|
d = pathlib.Path(p).parent
|
|
if d.is_dir():
|
|
return d
|
|
return None
|
|
|
|
|
|
def output_path(task_id: str, tdir: pathlib.Path | None,
|
|
outputs: dict[str, str]) -> pathlib.Path | None:
|
|
"""The task's output file: the path its launch result named when that
|
|
still exists, else ``<tasks dir>/<id>.output``."""
|
|
named = pathlib.Path(outputs[task_id]) if task_id in outputs else None
|
|
if named is not None and named.exists():
|
|
return named
|
|
if tdir is not None:
|
|
return tdir / f"{task_id}.output"
|
|
return named
|
|
|
|
|
|
def finished(path: pathlib.Path | None) -> bool | None:
|
|
"""True once the CLI wrote the exit marker; False while the file exists
|
|
without it; None when there is no file to watch (nothing we can wait on)."""
|
|
if path is None:
|
|
return None
|
|
try:
|
|
with open(path, "rb") as f:
|
|
f.seek(0, os.SEEK_END)
|
|
size = f.tell()
|
|
f.seek(max(0, size - 64))
|
|
tail = f.read()
|
|
except FileNotFoundError:
|
|
return None
|
|
except OSError:
|
|
return False
|
|
return bool(EXIT_MARK.search(tail))
|
|
|
|
|
|
def pending_shells(hook_input: dict) -> tuple[list[dict], dict[str, pathlib.Path | None]]:
|
|
"""The running plain background shells (Monitors excluded) and the output
|
|
file to watch for each."""
|
|
shells = running_shells(hook_input.get("background_tasks") or [])
|
|
if not shells:
|
|
return [], {}
|
|
monitors, outputs = scan_transcripts(hook_input.get("transcript_path"))
|
|
shells = [t for t in shells if t["id"] not in monitors]
|
|
tdir = tasks_dir(hook_input, outputs)
|
|
paths = {t["id"]: output_path(t["id"], tdir, outputs) for t in shells}
|
|
# A task with no output file at all is not something we can wait on
|
|
# (and waiting MAX_WAIT_S on a wrong path every stop would be worse than
|
|
# the bug) — drop it, loudly.
|
|
keep = []
|
|
for t in shells:
|
|
if finished(paths[t["id"]]) is None:
|
|
_log(f"no output file for task {t['id']} ({paths[t['id']]}); "
|
|
"not waiting on it")
|
|
continue
|
|
keep.append(t)
|
|
return keep, paths
|
|
|
|
|
|
def _label(t: dict) -> str:
|
|
desc = (t.get("description") or "").strip()
|
|
cmd = (t.get("command") or "").strip().replace("\n", " ")
|
|
if len(cmd) > 80:
|
|
cmd = cmd[:77] + "…"
|
|
bits = [t["id"]]
|
|
if desc:
|
|
bits.append(f"“{desc}”")
|
|
if cmd:
|
|
bits.append(f"`{cmd}`")
|
|
return " ".join(bits)
|
|
|
|
|
|
def reason_done(done: list[dict], paths: dict, held_s: float) -> str:
|
|
lines = [f"- {_label(t)} → output: {paths.get(t['id'])}" for t in done]
|
|
return (
|
|
"Stop guard (headless run): this process exits when your turn ends, "
|
|
"which would have killed the background shell task(s) you were waiting "
|
|
f"on. The turn was held {int(held_s)} s until they finished:\n"
|
|
+ "\n".join(lines)
|
|
+ "\nRead each output file now, act on the result, and only then finish "
|
|
"your turn (its completion notification may also follow)."
|
|
)
|
|
|
|
|
|
def reason_still_running(left: list[dict], paths: dict, held_s: float) -> str:
|
|
lines = [f"- {_label(t)} → output so far: {paths.get(t['id'])}" for t in left]
|
|
return (
|
|
"Stop guard (headless run): this process exits when your turn ends, "
|
|
"which would kill these background shell task(s), still running after "
|
|
f"the turn was held {int(held_s // 60)} min:\n"
|
|
+ "\n".join(lines)
|
|
+ "\nIf you still need them, check their output, then end your turn "
|
|
"again — the guard holds it for another stretch. If not, stop them "
|
|
"with TaskStop first."
|
|
)
|
|
|
|
|
|
RELAY_OPEN = "<<<USER MESSAGE>>>"
|
|
RELAY_CLOSE = "<<<END USER MESSAGE>>>"
|
|
|
|
|
|
def reason_mail(mail: list[str], left: list[dict], shells: list[dict],
|
|
paths: dict, held_s: float) -> str:
|
|
"""A message from the user arrived while the turn was held: hand it over
|
|
verbatim (fenced so the viewer can show it as the user's turn), say what
|
|
is still running, and ask for the hold to be re-armed afterwards."""
|
|
body = "\n\n".join(mail)
|
|
done = [t for t in shells if t not in left]
|
|
out = [
|
|
# (the CLI itself prefixes the reason with "Stop hook feedback: ")
|
|
"Stop guard (headless run): your user sent a message while the turn "
|
|
f"was held ({int(held_s)} s):",
|
|
RELAY_OPEN, body, RELAY_CLOSE,
|
|
]
|
|
if left:
|
|
out.append("Still running in the background (ending your turn would kill "
|
|
"them): " + "; ".join(_label(t) for t in left)
|
|
+ ". Answer the user, then end your turn again — the guard "
|
|
"holds it again for them.")
|
|
if done:
|
|
out.append("Finished meanwhile: "
|
|
+ "; ".join(f"{_label(t)} → output: {paths.get(t['id'])}"
|
|
for t in done) + ".")
|
|
return "\n".join(out)
|
|
|
|
|
|
def wait(hook_input: dict, *, max_wait_s: float = MAX_WAIT_S,
|
|
poll_s: float = POLL_S, sleep=time.sleep, clock=time.monotonic) -> dict | None:
|
|
"""Hold until every pending background shell has exited (or the budget is
|
|
spent). Returns the hook's JSON decision, or None to let the stop through."""
|
|
shells, paths = pending_shells(hook_input)
|
|
if not shells:
|
|
return None
|
|
start = clock()
|
|
ids = ", ".join(t["id"] for t in shells)
|
|
_log(f"holding the turn for background shell(s) {ids}")
|
|
mark_hold(shells)
|
|
left = list(shells)
|
|
mail: list[str] = []
|
|
try:
|
|
while left and clock() - start < max_wait_s:
|
|
sleep(poll_s)
|
|
left = [t for t in left if finished(paths[t["id"]]) is False]
|
|
mail = take_mail()
|
|
if mail:
|
|
break
|
|
finally:
|
|
clear_hold()
|
|
held = clock() - start
|
|
if mail:
|
|
_log(f"relaying {len(mail)} user message(s) after {int(held)} s; "
|
|
f"still running: {', '.join(t['id'] for t in left) or '-'}")
|
|
return {"decision": "block",
|
|
"reason": reason_mail(mail, left, shells, paths, held)}
|
|
if left:
|
|
_log(f"still running after {int(held)} s: "
|
|
f"{', '.join(t['id'] for t in left)}")
|
|
return {"decision": "block",
|
|
"reason": reason_still_running(left, paths, held)}
|
|
_log(f"released after {int(held)} s ({ids} finished)")
|
|
return {"decision": "block", "reason": reason_done(shells, paths, held)}
|
|
|
|
|
|
def main() -> int:
|
|
try:
|
|
hook_input = json.load(sys.stdin)
|
|
except ValueError:
|
|
return 0
|
|
if hook_input.get("hook_event_name", "Stop") != "Stop":
|
|
return 0
|
|
global LOG_FILE, SESSION_ID
|
|
SESSION_ID = hook_input.get("session_id") or None
|
|
LOG_FILE = _state_path(".guard.log")
|
|
if LOG_FILE is None:
|
|
sp = hook_input.get("scratchpad_dir")
|
|
if sp:
|
|
LOG_FILE = pathlib.Path(sp).parent / "stop-guard.log"
|
|
try:
|
|
out = wait(hook_input)
|
|
except Exception as e: # never turn a guard bug into a stuck turn
|
|
_log(f"error, letting the stop through: {e!r}")
|
|
return 0
|
|
if out:
|
|
print(json.dumps(out))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|