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>
196 lines
7.3 KiB
Python
196 lines
7.3 KiB
Python
"""stop_guard: which background tasks hold the turn, and for how long."""
|
|
|
|
import json
|
|
|
|
import stop_guard as sg
|
|
|
|
|
|
def _hook_input(tmp_path, tasks, transcript_lines=()):
|
|
sid = "11111111-2222-3333-4444-555555555555"
|
|
tasks_dir = tmp_path / "run" / sid / "tasks"
|
|
tasks_dir.mkdir(parents=True)
|
|
scratch = tmp_path / "run" / sid / "scratchpad"
|
|
scratch.mkdir()
|
|
transcript = tmp_path / "proj" / f"{sid}.jsonl"
|
|
transcript.parent.mkdir()
|
|
transcript.write_text("".join(json.dumps(l) + "\n" for l in transcript_lines))
|
|
return {"session_id": sid, "hook_event_name": "Stop",
|
|
"transcript_path": str(transcript), "scratchpad_dir": str(scratch),
|
|
"background_tasks": tasks}, tasks_dir, transcript
|
|
|
|
|
|
def _result(text):
|
|
return {"type": "user", "message": {"role": "user", "content": [
|
|
{"type": "tool_result", "tool_use_id": "t", "content": text}]}}
|
|
|
|
|
|
def _shell(id_, **kw):
|
|
return {"id": id_, "type": "shell", "status": "running",
|
|
"description": kw.get("desc", "build"), "command": kw.get("cmd", "make")}
|
|
|
|
|
|
def test_no_running_shell_lets_the_stop_through(tmp_path):
|
|
inp, _, _ = _hook_input(tmp_path, [
|
|
{"id": "a1", "type": "subagent", "status": "running"},
|
|
{"id": "b1", "type": "shell", "status": "completed"}])
|
|
assert sg.wait(inp, sleep=lambda s: None) is None
|
|
|
|
|
|
def test_monitors_are_not_waited_for(tmp_path):
|
|
inp, tdir, _ = _hook_input(tmp_path, [_shell("bmon")], [
|
|
_result("Monitor started (task bmon, expires in 5m unless the source ends first)")])
|
|
(tdir / "bmon.output").write_bytes(b"")
|
|
assert sg.wait(inp, sleep=lambda s: None) is None
|
|
|
|
|
|
def test_task_without_an_output_file_is_not_waited_for(tmp_path, capsys):
|
|
inp, _, _ = _hook_input(tmp_path, [_shell("bnofile")])
|
|
assert sg.wait(inp, sleep=lambda s: None) is None
|
|
assert "no output file" in capsys.readouterr().err
|
|
|
|
|
|
def test_holds_until_the_exit_marker_then_blocks(tmp_path):
|
|
inp, tdir, _ = _hook_input(tmp_path, [_shell("b75", desc="npm build")], [
|
|
_result("Command running in background with ID: b75. Output is being "
|
|
f"written to: {tmp_path}/run/x/tasks/b75.output. You will be notified")])
|
|
# the transcript names a path that no longer exists (a wiped tmp) — the
|
|
# tasks dir next to the scratchpad is the fallback
|
|
out = tdir / "b75.output"
|
|
out.write_bytes(b"compiling...\n")
|
|
polls = {"n": 0}
|
|
|
|
def sleep(_s):
|
|
polls["n"] += 1
|
|
if polls["n"] == 3:
|
|
out.write_bytes(b"compiling...\ndone\n\n[exited with code 0]\n")
|
|
|
|
res = sg.wait(inp, sleep=sleep, max_wait_s=999)
|
|
assert polls["n"] == 3
|
|
assert res["decision"] == "block"
|
|
assert "b75" in res["reason"] and "npm build" in res["reason"]
|
|
assert "until they finished" in res["reason"]
|
|
assert str(out) in res["reason"]
|
|
|
|
|
|
def test_output_path_from_the_transcript_wins_over_the_tasks_dir(tmp_path):
|
|
other = tmp_path / "elsewhere"
|
|
other.mkdir()
|
|
(other / "bx.output").write_bytes(b"[exited with code 1]\n")
|
|
inp, tdir, _ = _hook_input(tmp_path, [_shell("bx")], [
|
|
_result(f"Command running in background with ID: bx. Output is being "
|
|
f"written to: {other}/bx.output. You will be notified")])
|
|
res = sg.wait(inp, sleep=lambda s: None)
|
|
assert res["decision"] == "block" and str(other / "bx.output") in res["reason"]
|
|
|
|
|
|
def test_budget_spent_reports_still_running(tmp_path):
|
|
inp, tdir, _ = _hook_input(tmp_path, [_shell("bslow", cmd="sleep 1d")])
|
|
(tdir / "bslow.output").write_bytes(b"...")
|
|
t = {"now": 0.0}
|
|
|
|
def clock():
|
|
return t["now"]
|
|
|
|
def sleep(s):
|
|
t["now"] += s
|
|
|
|
res = sg.wait(inp, sleep=sleep, clock=clock, max_wait_s=10, poll_s=4)
|
|
assert res["decision"] == "block"
|
|
assert "still running" in res["reason"] and "TaskStop" in res["reason"]
|
|
assert "bslow" in res["reason"]
|
|
|
|
|
|
def test_subagent_transcripts_are_scanned_for_monitors(tmp_path):
|
|
inp, tdir, transcript = _hook_input(tmp_path, [_shell("bsub")])
|
|
sub = transcript.with_suffix("") / "subagents"
|
|
sub.mkdir(parents=True)
|
|
(sub / "agent-abc.jsonl").write_text(
|
|
json.dumps(_result("Monitor started (task bsub, expires in 5m)")) + "\n")
|
|
(tdir / "bsub.output").write_bytes(b"")
|
|
assert sg.wait(inp, sleep=lambda s: None) is None
|
|
|
|
|
|
def test_finished_marker_detection(tmp_path):
|
|
p = tmp_path / "o.output"
|
|
assert sg.finished(p) is None
|
|
p.write_bytes(b"x" * 10_000 + b"\n[exited with code 137]\n")
|
|
assert sg.finished(p) is True
|
|
p.write_bytes(b"[exited with code 0] but then more output")
|
|
assert sg.finished(p) is False
|
|
assert sg.finished(None) is None
|
|
|
|
|
|
def test_main_is_silent_without_pending_work(monkeypatch, capsys):
|
|
import io
|
|
monkeypatch.setattr("sys.stdin", io.StringIO(json.dumps(
|
|
{"hook_event_name": "Stop", "background_tasks": []})))
|
|
assert sg.main() == 0
|
|
assert capsys.readouterr().out == ""
|
|
|
|
|
|
def test_main_never_blocks_on_garbage(monkeypatch, capsys):
|
|
import io
|
|
monkeypatch.setattr("sys.stdin", io.StringIO("not json"))
|
|
assert sg.main() == 0
|
|
assert capsys.readouterr().out == ""
|
|
|
|
|
|
def _state(monkeypatch, tmp_path, sid="11111111-2222-3333-4444-555555555555"):
|
|
st = tmp_path / "state"
|
|
st.mkdir()
|
|
monkeypatch.setattr(sg, "STATE_DIR", str(st))
|
|
monkeypatch.setattr(sg, "SESSION_ID", sid)
|
|
return st
|
|
|
|
|
|
def test_hold_marker_lives_for_the_hold(tmp_path, monkeypatch):
|
|
st = _state(monkeypatch, tmp_path)
|
|
inp, tdir, _ = _hook_input(tmp_path, [_shell("bh")])
|
|
out = tdir / "bh.output"
|
|
out.write_bytes(b"...")
|
|
seen = {}
|
|
|
|
def sleep(_s):
|
|
hold = st / f"{sg.SESSION_ID}.hold"
|
|
seen["during"] = hold.exists() and "bh" in json.loads(hold.read_text())["tasks"]
|
|
out.write_bytes(b"[exited with code 0]\n")
|
|
|
|
res = sg.wait(inp, sleep=sleep)
|
|
assert seen["during"] is True
|
|
assert not (st / f"{sg.SESSION_ID}.hold").exists()
|
|
assert res["decision"] == "block"
|
|
assert (st / f"{sg.SESSION_ID}.guard.log").exists() is False # main() sets LOG_FILE
|
|
|
|
|
|
def test_a_message_left_in_the_mailbox_ends_the_hold_and_is_relayed(tmp_path, monkeypatch):
|
|
st = _state(monkeypatch, tmp_path)
|
|
inp, tdir, _ = _hook_input(tmp_path, [_shell("bm", desc="apk build")])
|
|
(tdir / "bm.output").write_bytes(b"...")
|
|
box = st / f"{sg.SESSION_ID}.inbox"
|
|
polls = {"n": 0}
|
|
|
|
def sleep(_s):
|
|
polls["n"] += 1
|
|
if polls["n"] == 2:
|
|
box.mkdir()
|
|
(box / "1.json").write_text(json.dumps({"prompt": "Up? how far is it"}))
|
|
(box / "2.json").write_text(json.dumps({"prompt": "and ping me after"}))
|
|
|
|
res = sg.wait(inp, sleep=sleep, max_wait_s=999)
|
|
assert polls["n"] == 2
|
|
assert res["decision"] == "block"
|
|
r = res["reason"]
|
|
assert r.startswith("Stop guard (headless run): your user sent")
|
|
assert f"{sg.RELAY_OPEN}\nUp? how far is it\n\nand ping me after\n{sg.RELAY_CLOSE}" in r
|
|
assert "Still running" in r and "apk build" in r and "end your turn again" in r
|
|
assert list(box.iterdir()) == [] # consumed
|
|
assert not (st / f"{sg.SESSION_ID}.hold").exists()
|
|
|
|
|
|
def test_without_a_state_dir_nothing_is_written(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(sg, "STATE_DIR", None)
|
|
inp, tdir, _ = _hook_input(tmp_path, [_shell("bn")])
|
|
(tdir / "bn.output").write_bytes(b"[exited with code 0]\n")
|
|
assert sg.wait(inp, sleep=lambda s: None)["decision"] == "block"
|
|
assert sg.take_mail() == []
|