tests/test_baton.py

#!/usr/bin/env python3 """Prove the baton path does what it claims - and that each check can fail. 1. Tiers are fractions of the window, so they mean the same on a small plan and a large one. 2. The Stop nudge speaks at the right moments, stays quiet when covered, and never loops (stop_hook_active) or speaks in seed mode. 3. A baton goes only to its owner: /clear and compaction of the same process take it, a new process takes it only when the writer is gone, and two waiting are never guessed between. 4. A taken baton is consumed once, and the same session's machine handoff is retired with it, so a later session never opens with a stale copy. 5. A shrug is refused: a handoff has to say something. Each claim is re-run against a deliberately broken copy (MUTANTS at the bottom); a check that passes against broken code is not counted as a check. """ from __future__ import annotations import io import json import os import shutil import subprocess import sys import tempfile from contextlib import redirect_stdout from pathlib import Path BIN = Path(__file__).resolve().parent.parent / "bin" FAILED: list[str] = [] def check(name: str, ok: bool, detail: str = "") -> bool: print(f" {'ok ' if ok else 'FAIL'} {name}" + (f" {detail}" if detail else "")) if not ok: FAILED.append(name) return ok def transcript(path: Path, tok: int) -> None: rows = [{"type": "user", "message": {"role": "user", "content": "keep the logs under /srv/app/logs, never delete them"}}, {"type": "assistant", "message": {"role": "assistant", "content": [{"type": "text", "text": "ok"}], "usage": {"input_tokens": 100, "cache_read_input_tokens": tok - 100, "cache_creation_input_tokens": 0, "output_tokens": 5}}}] path.write_text("\n".join(json.dumps(r) for r in rows) + "\n", encoding="utf-8") BATON = ("# BATON\n## WHAT I AM DOING\nMoving the log rotation to the new volume.\n" "## IN FLIGHT\nrotate.conf edited, not reloaded. Next: reload and watch one rotation.\n" "## EXACT NAMES\n/srv/app/logs, rotate.conf, unit app-rotate.timer\n") def run(bin_dir: Path) -> list[str]: FAILED.clear() tmp = Path(tempfile.mkdtemp(prefix="logbook-baton-test-")) os.environ.update({"LOGBOOK_HOME": str(tmp / "store"), "LOGBOOK_PROJECT": "proj", "LOGBOOK_WINDOW": "200000", "HOME": str(tmp / "home")}) for k in ("LOGBOOK_HANDOFF_MODE", "LOGBOOK_AGENT_PID", "CLAUDE_SESSION_ID", "CLAUDE_TRANSCRIPT_PATH"): os.environ.pop(k, None) for m in ("baton", "handoff", "paths"): sys.modules.pop(m, None) sys.path.insert(0, str(bin_dir)) try: import baton, handoff # noqa: E401 W = 200_000 # 1. tiers as fractions check("under 55% is quiet", baton.tier(100_000, W) == "") check("55% asks", baton.tier(112_000, W) == "ask") check("70% pushes", baton.tier(141_000, W) == "push") check("82% is hard", baton.tier(165_000, W) == "hard") check("same fraction, bigger window, same tier", baton.tier(560_000, 1_000_000) == "ask") check("window grows past 200k when the session visibly does", (os.environ.pop("LOGBOOK_WINDOW") and baton.window(250_000)) == 1_000_000) os.environ["LOGBOOK_WINDOW"] = "200000" # 2. the Stop nudge tr = tmp / "t.jsonl" transcript(tr, 165_000) def stop(payload): buf = io.StringIO() with redirect_stdout(buf): baton.do_stop(payload) return buf.getvalue().strip() p = {"session_id": "aaaa1111", "transcript_path": str(tr)} out = stop(p) check("hard tier with no baton blocks the stop", '"decision": "block"' in out, out[:80]) check("never blocks twice in one stop (stop_hook_active)", stop({**p, "stop_hook_active": True}) == "") handoff.mode_file().write_text("seed", encoding="utf-8") check("seed mode never nudges", stop(p) == "") handoff.mode_file().unlink() check("baton is the default mode", handoff.get_mode() == "baton") transcript(tr, 112_000) st = {} first = baton.nag(112_000, W, {}, st) again = baton.nag(115_000, W, {}, st) later = baton.nag(123_000, W, {}, st) check("ask speaks once, then waits 5% more", bool(first) and again == "" and bool(later)) # 5. a shrug is refused; 3. ownership env = {**os.environ, "CLAUDE_SESSION_ID": "aaaa1111", "CLAUDE_TRANSCRIPT_PATH": str(tr), "LOGBOOK_AGENT_PID": "4242"} r = subprocess.run([sys.executable, str(bin_dir / "baton.py"), "write"], input="ok", text=True, capture_output=True, env=env) check("a two-letter baton is refused", r.returncode == 2) transcript(tr, 150_000) # written at 150k: covers until 150k + 15% of 200k = 180k r = subprocess.run([sys.executable, str(bin_dir / "baton.py"), "write"], input=BATON * 2, text=True, capture_output=True, env=env) check("a real baton is written", r.returncode == 0, r.stderr[-200:]) b = baton.pending("proj") check("exactly one baton on disk", len(b) == 1) body = b[0].read_text(encoding="utf-8") if b else "" check("the machine appendix rides underneath", "ATTACHED BY LOGBOOK" in body and "/srv/app/logs" in body) transcript(tr, 166_000) baton._state_path("aaaa1111").unlink(missing_ok=True) # forget the earlier nudge out = stop(p) check("a written baton covers the session: no hard block, told to hand off", "ready for /clear" in out and "cannot end" not in out, out[:90]) transcript(tr, 185_000) check("past its cover the hard block comes back", "cannot end" in stop(p)) (baton._dir() / "proj--aaaa1111.md").write_text("machine copy", encoding="utf-8") os.environ["LOGBOOK_AGENT_PID"] = "9999" check("/clear in a DIFFERENT process does not take it", baton.take("clear", "bbbb2222") == "") check("resume never takes one", baton.take("resume", "bbbb2222") == "") os.environ["LOGBOOK_AGENT_PID"] = "4242" got = baton.take("clear", "bbbb2222") check("/clear in the SAME process takes it", "Moving the log rotation" in got) check("consumed once", baton.take("clear", "bbbb2222") == "") check("its machine handoff was retired with it", not (baton._dir() / "proj--aaaa1111.md").exists()) # compaction keeps the session id; startup needs the writer gone; two are refused for s in ("cccc3333", "dddd4444"): os.environ["LOGBOOK_AGENT_PID"] = "1" # pid 1 is never the agent: "writer gone" subprocess.run([sys.executable, str(bin_dir / "baton.py"), "write", "--session", s], input=BATON * 2, text=True, capture_output=True, env={**os.environ, "CLAUDE_TRANSCRIPT_PATH": str(tr)}) os.environ["LOGBOOK_AGENT_PID"] = "7777" two = baton.take("startup", "eeee5555") check("two waiting are never guessed between", "NOT DELIVERED" in two and len(baton.pending("proj")) == 2) baton.pending("proj")[0].unlink() check("compaction of the same session takes its own", "Moving" in baton.take("compact", "dddd4444")) finally: sys.path.remove(str(bin_dir)) shutil.rmtree(tmp, ignore_errors=True) return list(FAILED) MUTANTS = [ # (what is broken, file, old, new) - every one must make the suite FAIL ("ask tier ignored", "baton.py", 'else "ask" if f >= ASK else ""', 'else ""'), ("loops on stop_hook_active", "baton.py", 'if p.get("stop_hook_active") or p.get("agent_id"):', 'if p.get("agent_id"):'), ("any process takes a /clear baton", "baton.py", 'pick = [c for c in cands if (pid and _meta(c).get("pid") == str(pid))', 'pick = [c for c in cands if (True)'), ("not consumed", "baton.py", 'p.replace(p.with_name(p.name + ".taken"))', 'pass'), ("shrugs accepted", "baton.py", "if len(text.strip()) < MIN_CHARS:", "if False:"), ("seed is the default", "handoff.py", 'DEFAULT_MODE = "baton"', 'DEFAULT_MODE = "seed"'), ("guesses between two", "baton.py", "if len(pick) != 1:", "if len(pick) == 0:"), ("a baton never covers", "baton.py", "return m if m and tok - int(m.get(\"tokens\", 0) or 0) < COVER * win else {}", "return {}"), ("a baton covers forever", "baton.py", "return m if m and tok - int(m.get(\"tokens\", 0) or 0) < COVER * win else {}", "return m"), ] def main() -> int: print("baton: the real code") bad = run(BIN) print("\nbaton: mutants (each must be caught)") survived = [] for name, f, old, new in MUTANTS: d = Path(tempfile.mkdtemp(prefix="logbook-baton-mut-")) try: shutil.copytree(BIN, d / "bin") p = d / "bin" / f s = p.read_text(encoding="utf-8") if old not in s: survived.append(f"{name} (anchor missing)") continue p.write_text(s.replace(old, new, 1), encoding="utf-8") with redirect_stdout(io.StringIO()): caught = bool(run(d / "bin")) print(f" {'KILLED ' if caught else 'SURVIVED'} {name}") if not caught: survived.append(name) finally: shutil.rmtree(d, ignore_errors=True) if bad or survived: print(f"\nbaton test FAILED: {len(bad)} check(s) failed, {len(survived)} mutant(s) survived") return 1 print(f"\nall baton checks passed; {len(MUTANTS)}/{len(MUTANTS)} mutants caught") return 0 if __name__ == "__main__": raise SystemExit(main())