tankslab.com/logbook/source
bin/baton.py
482 lines, exactly as they ship in the download.
#!/usr/bin/env python3
"""baton.py - the agent writes its own handoff, while it still has the whole picture.
WHY THIS IS THE DEFAULT
Compaction picks what survives for you, and it loses the things that matter most:
exact paths, names, what is actually running versus only prepared. The machine
handoff (handoff.py) is better - local, no model call - but it is still a program
choosing what to keep. A baton is the agent's own account of where the work
stands, written before the window fills, and it carries the things only the agent
knows are important.
The machine extract still rides along underneath it: the instructions that are
still binding and what was asked, exactly as it was typed. The baton says what
the agent chose to carry; the appendix makes sure nothing you said was dropped
on the way.
HOW IT RUNS - ONE HOOK, AND IT SEES NOTHING NEW
Logbook refuses the hooks that would read your prompts or tool output (see the
README). So the nudge lives in the Stop hook, which already reads the transcript
file to capture it. At the end of a turn it reads the size of the live context
from the last reply's own usage numbers (exact, not guessed), compares it with the
window, and acts on three tiers - fractions of the window, so they mean the same
thing on a 200k plan and a 1M one:
ASK 55% once, then every further 5%: write a baton at this natural break
PUSH 70% every further 2%: write it now
HARD 82% the turn may not end until a baton is on file
Each is a fraction of the window and can be changed (LOGBOOK_BATON_ASK/PUSH/HARD).
The window is LOGBOOK_WINDOW if set, else autoCompactWindow from the agent's
settings, else 200,000 - and 1,000,000 once a session has visibly grown past 200k.
A baton "covers" the session until the context grows another 15% of the window.
Once covered, the agent is told to say so: the person types /clear and the next
context opens with it.
WHO GETS IT
A baton belongs to the process that wrote it. /clear and compaction keep the same
process, so the next context of that same agent takes it - and only that one. A
brand-new process takes it only if the writer is gone, and nothing is guessed when
more than one is waiting. Wrong context delivered silently is worse than none.
A hook must never break the session: every path exits 0, and the Stop block is
returned at most once per stop (stop_hook_active).
baton.py write [--session SID] the agent's handoff, from stdin
baton.py status [--transcript T] tier, size, baton on file
baton.py take --source S --session SID used by SessionStart (prints it, consumes it)
baton.py stop the Stop hook (payload on stdin)
"""
from __future__ import annotations
import json
import os
import re
import subprocess
import sys
import time
from pathlib import Path
BIN = Path(__file__).resolve().parent
sys.path.insert(0, str(BIN))
ME = BIN / "baton.py"
def _frac(name: str, default: float) -> float:
try:
v = float(os.environ.get(name, ""))
return v if 0.05 <= v <= 0.99 else default
except ValueError:
return default
ASK = _frac("LOGBOOK_BATON_ASK", 0.55)
PUSH = _frac("LOGBOOK_BATON_PUSH", 0.70)
HARD = _frac("LOGBOOK_BATON_HARD", 0.82)
COVER = 0.15 # a baton covers the session until it grows this much more
STALE_HOURS = 18.0 # same age limit as a machine handoff
MIN_CHARS = 200 # a baton shorter than this is not a handoff, it is a shrug
TEMPLATE = """Write it with: python3 "{me}" write <<'EOF'
# BATON - <date/time>
## WHAT I AM DOING the task, in a sentence or two, and who asked for it
## DONE AND VERIFIED what is finished, and how you know (tests run, output seen)
## IN FLIGHT exactly where you stopped: files, commands, the next step
## WAITING ON SOMEONE questions asked and not yet answered
## LEARNED traps hit, so the next context does not hit them again
## EXACT NAMES paths, ids, versions, anything that must be typed exactly
EOF
Write it for a stranger who must continue the work. The instructions still in force and
what was asked, exactly as typed are attached for you - do not copy them in."""
# ------------------------------------------------------------------ where it lives
def _dir() -> Path:
import paths
d = paths.root() / "handoff"
d.mkdir(parents=True, exist_ok=True)
try:
os.chmod(d, 0o700)
except Exception:
pass
return d
def _project() -> str:
import handoff
return handoff._project()
def _sid8(v: str) -> str:
return re.sub(r"[^A-Za-z0-9]", "", v or "")[:8]
def _log(msg: str) -> None:
try:
import paths
with open(paths.root() / "baton.log", "a", encoding="utf-8") as f:
f.write(f"{time.strftime('%Y-%m-%d %H:%M:%S')} {msg}\n")
except Exception:
pass
# ------------------------------------------------------------------ whose process
def _parent(pid: int) -> tuple[int, str]:
"""(parent pid, name) of a process. /proc where it exists, `ps` elsewhere."""
try:
stat = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8")
name = Path(f"/proc/{pid}/comm").read_text(encoding="utf-8").strip()
return int(stat.rsplit(")", 1)[1].split()[1]), name
except Exception:
pass
try:
out = subprocess.run(["ps", "-o", "ppid=", "-o", "comm=", "-p", str(pid)],
capture_output=True, text=True, timeout=3).stdout.strip()
pp, _, name = out.partition(" ")
return int(pp), os.path.basename(name.strip())
except Exception:
return 0, ""
def agent_pid() -> int:
"""The agent process this runs under: /clear and compaction keep it, a new session
gets a new one. 0 when it cannot be found (then nothing is matched by process)."""
v = os.environ.get("LOGBOOK_AGENT_PID", "")
if v.isdigit():
return int(v)
p = os.getppid()
for _ in range(12):
if p <= 1:
break
pp, name = _parent(p)
if name == "claude":
return p
p = pp
return 0
def _alive(pid: int) -> bool:
if pid <= 0:
return False
return _parent(pid)[1] == "claude"
# ------------------------------------------------------------------ how full
def _tail(path: Path, n: int = 400) -> list[str]:
try:
with open(path, "rb") as f:
f.seek(0, os.SEEK_END)
size, data = f.tell(), b""
while size > 0 and data.count(b"\n") <= n:
step = min(262144, size)
size -= step
f.seek(size)
data = f.read(step) + data
return data.decode("utf-8", "replace").splitlines()[-n:]
except Exception:
return []
def tokens(transcript) -> int:
"""Live context = the prompt side of the last reply's usage. 0 if unknown."""
for line in reversed(_tail(Path(transcript))):
if '"usage"' not in line:
continue
try:
u = (json.loads(line).get("message") or {}).get("usage") or {}
except Exception:
continue
t = (int(u.get("input_tokens", 0) or 0) + int(u.get("cache_read_input_tokens", 0) or 0)
+ int(u.get("cache_creation_input_tokens", 0) or 0))
if t > 0:
return t
return 0
def window(tok: int = 0) -> int:
v = os.environ.get("LOGBOOK_WINDOW", "")
if v.isdigit() and int(v) >= 10000:
return int(v)
try:
s = json.loads((Path.home() / ".claude" / "settings.json").read_text(encoding="utf-8"))
w = int(s.get("autoCompactWindow") or 0)
if w >= 10000:
return w
except Exception:
pass
return 1_000_000 if tok > 200_000 else 200_000
def tier(tok: int, win: int) -> str:
f = tok / win if win else 0
return "hard" if f >= HARD else "push" if f >= PUSH else "ask" if f >= ASK else ""
# ------------------------------------------------------------------ batons on disk
def _path(proj: str, sid: str) -> Path:
return _dir() / f"baton--{proj}--{sid}.md"
def _meta(p: Path) -> dict:
try:
head = p.read_text(encoding="utf-8")[:400]
except Exception:
return {}
m = re.search(r"<!-- logbook-baton (.*?) -->", head)
if not m:
return {}
return dict(kv.split("=", 1) for kv in m.group(1).split() if "=" in kv)
def pending(proj: str) -> list[Path]:
return sorted(_dir().glob(f"baton--{proj}--*.md"), key=lambda f: f.stat().st_mtime)
def mine(proj: str, sid: str) -> dict:
p = _path(proj, _sid8(sid))
return _meta(p) if p.exists() else {}
def covered(proj: str, sid: str, tok: int, win: int) -> dict:
m = mine(proj, sid)
return m if m and tok - int(m.get("tokens", 0) or 0) < COVER * win else {}
def _appendix(transcript: str) -> str:
"""What the machine attaches under the agent's own words: the instructions still in
force and what was asked, exactly as typed. Local text extraction, no model call."""
if not transcript or not Path(transcript).exists():
return ""
try:
out = subprocess.run([sys.executable, str(BIN / "brief.py"), transcript, "--budget", "6000"],
capture_output=True, text=True, timeout=60).stdout
except Exception:
return ""
keep, cur = [], None
for line in out.splitlines():
if line.startswith("## "):
cur = line.startswith("## STANDING INSTRUCTIONS") or line.startswith("## WHAT WAS ASKED")
if cur:
keep.append(line)
return "\n".join(keep).strip()
def cmd_write(argv: list[str]) -> int:
text = sys.stdin.read()
if len(text.strip()) < MIN_CHARS:
print(f"baton: refused - {len(text.strip())} characters is not a handoff. Write what the "
f"next context needs (at least {MIN_CHARS} characters).", file=sys.stderr)
return 2
sid = ""
for i, a in enumerate(argv):
if a == "--session" and i + 1 < len(argv):
sid = argv[i + 1]
sid = _sid8(sid or os.environ.get("CLAUDE_SESSION_ID", "")) or "unknown"
tr = os.environ.get("CLAUDE_TRANSCRIPT_PATH", "")
if not tr:
import handoff
t = handoff._newest_transcript()
tr = str(t) if t else ""
tok = tokens(tr) if tr else 0
proj = _project()
stamp = time.strftime("%Y-%m-%d %H:%M:%S")
head = (f"<!-- logbook-baton project={proj} session={sid} pid={agent_pid()} tokens={tok} "
f"ts={int(time.time())} written={stamp} -->\n")
app = _appendix(tr)
body = head + text.strip() + "\n"
if app:
body += ("\n---\n## ATTACHED BY LOGBOOK (machine-extracted from the transcript, not the "
"agent's words)\n" + app + "\n")
p = _path(proj, sid)
tmp = p.with_suffix(".md.tmp")
tmp.write_text(body, encoding="utf-8")
try:
os.chmod(tmp, 0o600)
except Exception:
pass
tmp.replace(p)
_log(f"write sid={sid} tokens={tok} bytes={len(body)}")
print(f"baton written: {p} ({len(text.strip()):,} characters of yours"
f"{', plus the attached instructions and asks' if app else ''})")
print(" Tell the person you are working with: type /clear when ready - the next context opens with it.")
return 0
def take(source: str, sid: str) -> str:
"""The baton this context is owed, consumed - or "" if none. Never a guess.
clear / compact : the same agent process continuing - take ITS baton (matched by
process, or by session for compact), never anybody else's.
startup : a new process - take one only if its writer is gone, and only
if exactly one is waiting.
resume / fork : already hold their own history; nothing.
"""
proj, me, pid = _project(), _sid8(sid), agent_pid()
cutoff = time.time() - STALE_HOURS * 3600
cands = [c for c in pending(proj) if c.stat().st_mtime >= cutoff]
if source in ("clear", "compact"):
pick = [c for c in cands if (pid and _meta(c).get("pid") == str(pid))
or (source == "compact" and me and _meta(c).get("session") == me)]
elif source in ("startup", ""):
pick = [c for c in cands if not _alive(int(_meta(c).get("pid", "0") or 0))]
else:
return ""
if len(pick) != 1:
if len(pick) > 1:
_log(f"take REFUSED source={source} sid={me}: {len(pick)} candidates")
return ("BATON NOT DELIVERED - more than one is waiting for this folder. Nothing was "
"consumed. See: python3 \"" + str(ME) + "\" status")
return ""
p = pick[0]
meta = _meta(p)
try:
text = p.read_text(encoding="utf-8")
age = (time.time() - p.stat().st_mtime) / 60
p.replace(p.with_name(p.name + ".taken"))
except Exception:
return ""
# The same session's machine handoff is now superseded: retire it too, or a later new
# session would open with the stale copy of a conversation that was already carried.
try:
sib = _dir() / f"{proj}--{meta.get('session', '')}.md"
if sib.exists():
sib.replace(sib.with_name(sib.name + ".taken"))
except Exception:
pass
_log(f"take source={source} sid={me} from={meta.get('session')} age={age:.0f}m")
when = f"{age / 60:.1f} hours ago" if age >= 60 else f"{age:.0f} minutes ago"
return (f"YOUR BATON, written {when} by the previous context of this work, in its own words "
f"(the attached part is machine-extracted). Pick up where it says you stopped; verify "
f"anything it calls running before relying on it.\n\n" + re.sub(r"^<!--.*?-->\n", "", text))
def cmd_take(argv: list[str]) -> int:
src = sid = ""
for i, a in enumerate(argv):
if a == "--source" and i + 1 < len(argv):
src = argv[i + 1]
if a == "--session" and i + 1 < len(argv):
sid = argv[i + 1]
out = take(src, sid)
if out:
print(out)
return 0
def nag(tok: int, win: int, cov: dict, st: dict) -> str:
"""The Stop-hook text for this moment, or "" to stay quiet. Updates st in place."""
t, k, last = tier(tok, win), tok // 1000, int(st.get("last_tok", 0) or 0)
pct = round(100 * tok / win) if win else 0
if cov:
if t in ("", "ask") or (last and tok - last < 0.05 * win):
return ""
st["last_tok"] = tok
return (f"[logbook] Context {k}k ({pct}% of the window). Your baton is on file. Tell the "
f"person you are ready for /clear - the next context opens with it. If a lot has "
f"changed since you wrote it, rewrite it first.")
if t == "ask":
if last and tok - last < 0.05 * win:
return ""
st["last_tok"] = tok
return (f"[logbook] Context {k}k ({pct}% of the window). Write your baton now - your own "
f"handoff, so the next context starts where you are and YOU choose what carries.\n"
+ TEMPLATE.format(me=ME))
if t == "push":
if last and tok - last < 0.02 * win:
return ""
st["last_tok"] = tok
return (f"[logbook] Context {k}k ({pct}%) and no baton on file. Write it before any other "
f"work, and say so to the person.\n" + TEMPLATE.format(me=ME))
if t == "hard":
st["last_tok"] = tok
return (f"[logbook] Context {k}k ({pct}%), no baton, and compaction is close - it will "
f"choose what you keep, not you. This turn cannot end until the baton is "
f"written.\n" + TEMPLATE.format(me=ME))
return ""
def _state_path(sid: str) -> Path:
return _dir() / f".baton-state--{_sid8(sid) or 'unknown'}.json"
def do_stop(p: dict) -> int:
"""The Stop hook. Blocks at most once per stop; never raises; always exits 0."""
try:
if p.get("stop_hook_active") or p.get("agent_id"):
return 0
import handoff
if handoff.get_mode() != "baton":
return 0
sid, tr = str(p.get("session_id") or ""), str(p.get("transcript_path") or "")
if not sid or not tr:
return 0
tok = tokens(tr)
if not tok:
return 0
win = window(tok)
sp = _state_path(sid)
try:
st = json.loads(sp.read_text(encoding="utf-8"))
except Exception:
st = {}
msg = nag(tok, win, covered(_project(), sid, tok, win), st)
try:
sp.write_text(json.dumps(st), encoding="utf-8")
except Exception:
pass
if msg:
_log(f"stop nudge sid={_sid8(sid)} tokens={tok} window={win} tier={tier(tok, win)}")
print(json.dumps({"decision": "block", "reason": msg}))
except Exception as e:
_log(f"stop FAILED (the session continues): {e}")
return 0
def cmd_status(argv: list[str]) -> int:
tr = ""
for i, a in enumerate(argv):
if a == "--transcript" and i + 1 < len(argv):
tr = argv[i + 1]
if not tr:
import handoff
t = handoff._newest_transcript()
tr = str(t) if t else ""
tok = tokens(tr) if tr else 0
win = window(tok)
import handoff
print(f"mode: {handoff.get_mode()} (baton = the agent writes its own handoff)")
print(f"context: {tok:,} of {win:,} ({round(100 * tok / win) if win else 0}%) tier: {tier(tok, win) or 'under'}")
print(f"tiers: ask {ASK:.0%} push {PUSH:.0%} hard {HARD:.0%} of the window")
pend = pending(_project())
if not pend:
print("batons: none waiting for this folder")
for b in pend:
m = _meta(b)
age = (time.time() - b.stat().st_mtime) / 60
print(f"baton: {b.name} {b.stat().st_size:,}B {age:.0f} min old "
f"session {m.get('session', '?')} at {int(m.get('tokens', 0) or 0) // 1000}k")
return 0
def main() -> int:
cmd = sys.argv[1] if len(sys.argv) > 1 else "status"
argv = sys.argv[2:]
if cmd == "write":
return cmd_write(argv)
if cmd == "take":
return cmd_take(argv)
if cmd == "stop":
try:
p = json.loads(sys.stdin.read() or "{}")
except Exception:
p = {}
return do_stop(p)
if cmd == "status":
return cmd_status(argv)
print(__doc__)
return 2
if __name__ == "__main__":
sys.exit(main())