bin/handoff.py

#!/usr/bin/env python3 """handoff.py - move to a fresh session instead of paying to compact the old one. THE ARGUMENT Compaction asks the model to read a nearly-full context window and write a summary. You are billed for that reading. On a 200k-context plan that is roughly 200,000 tokens spent to produce a few hundred lines - and it repeats every time the window fills. The bigger the session, the more it costs, and it costs the most to exactly the people who can least spare it. The transcript is already on disk. Extracting the spine from it is a text-processing job, not a reasoning job. Recall does it in about 0.2 seconds of local CPU, with no model call and no network, at roughly 250x reduction. So: end the session, start a clean one, and seed it with the brief. What you carry forward costs ~6k tokens once instead of ~200k tokens repeatedly. Compaction remains available and is still the right tool when a session must stay open. This is for when it does not have to. ONE-SHOT, ON PURPOSE A handoff is consumed the first time it is taken. It is renamed, not deleted, so it can be read afterwards. If it were left in place, every future session for this project would open with the same stale context and quietly drift. A handoff also declares its own age. Context from four days ago being injected as if it were current is the failure this guards against, so the age is printed in the brief itself where it cannot be missed. """ from __future__ import annotations import hashlib 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)) STALE_HOURS = 18.0 # Two ways to carry a session forward. Neither is a compromise of the other - they # suit different people, so the choice is the user's and it is remembered. # # seed the whole brief goes into the new session up front. Nothing to fetch, # nothing to forget, everything already in front of you. Costs ~6k tokens # once. This is the default because it cannot silently fail to happen. # # pointer only a short notice goes in - the standing instructions and the fact # that a full brief exists, with the command to read it. Costs a few # hundred tokens. The rest is fetched only if it is actually needed. # # The standing instructions ride along in BOTH modes. A rule the person is still # bound by is not detail to be fetched on demand; if it only arrives when someone # remembers to ask for it, it will eventually not arrive at all. MODES = ("seed", "pointer") DEFAULT_MODE = "seed" def mode_file() -> Path: return _dir() / "mode" def get_mode() -> str: env = (os.environ.get("RECALL_HANDOFF_MODE") or "").strip().lower() if env in MODES: return env try: v = mode_file().read_text(encoding="utf-8").strip().lower() if v in MODES: return v except Exception: pass return DEFAULT_MODE def set_mode(v: str) -> int: v = (v or "").strip().lower() if v not in MODES: print(f"mode must be one of: {', '.join(MODES)}", file=sys.stderr) return 2 mode_file().write_text(v, encoding="utf-8") print(f"handoff mode: {v}") print(" seed - full brief injected into the next session (~6k tokens, nothing to fetch)" if v == "seed" else " pointer - short notice injected; run 'recall handoff show' for the full brief") return 0 def _pointer_view(text: str) -> str: """The short form: standing instructions, plus how to get the rest. Deliberately NOT a truncation of the brief. Cutting a document at N characters keeps whatever happened to be at the top, which is not the same as keeping what matters. This keeps the section that is binding and names the rest. """ out, keep = [], False for line in text.splitlines(): if line.startswith("## STANDING INSTRUCTIONS"): keep = True elif line.startswith("## ") and keep: break if keep: out.append(line) have = [l[3:].split(" (")[0].strip() for l in text.splitlines() if l.startswith("## ")] body = "\n".join(out).strip() tail = ("\nFull brief not loaded (this is 'pointer' mode). It also has: " + ", ".join(h for h in have if not h.startswith("STANDING")) + ".\n" "Read it with: recall handoff show\n" "Switch modes: recall handoff mode seed") return (body + "\n" + tail) if body else tail.lstrip() def _root() -> Path: # paths.py is the single place that decides where anything lives, so a user who # moves the store onto an encrypted volume moves ALL of it, not most of it. import paths return paths.root() def _dir() -> Path: d = _root() / "handoff" d.mkdir(parents=True, exist_ok=True) try: os.chmod(d, 0o700) except Exception: pass return d def _project() -> str: """A name for this working directory that CANNOT collide with another one. The first version used the directory's base name. That is fine until two unrelated projects both contain a folder called "src", "app" or "client" - at which point one session's handoff is delivered into the other project, and a conversation about one piece of work opens inside a different one. A handoff is plain text from a real conversation, so that is not a cosmetic bug. So the full path decides identity, and the base name is kept only to make the file recognisable to a human reading the directory listing. """ p = os.environ.get("RECALL_PROJECT") if p: return re.sub(r"[^A-Za-z0-9._-]", "_", p)[:64] cwd = Path.cwd().resolve() tag = hashlib.sha256(str(cwd).encode("utf-8")).hexdigest()[:8] base = re.sub(r"[^A-Za-z0-9._-]", "_", cwd.name or "session")[:40] return f"{base}-{tag}" def _session_tag(argv: list[str] | None = None) -> str: """Short id of the session asking, or "" if it did not say. A handoff is a real conversation in plain text. Working out WHOSE it is has to be something the code can actually check, not something it assumes, so the id is carried explicitly and the empty string is a legitimate answer. """ if argv: for i, a in enumerate(argv): if a == "--session" and i + 1 < len(argv): return re.sub(r"[^A-Za-z0-9]", "", argv[i + 1])[:8] v = os.environ.get("CLAUDE_SESSION_ID") or "" return re.sub(r"[^A-Za-z0-9]", "", v)[:8] def _pending(proj: str) -> list[Path]: """Every unconsumed handoff for this working directory, newest last. Includes the pre-session-scoping name (<project>.md) so an upgrade does not strand a handoff somebody is waiting on. """ d = _dir() found = list(d.glob(f"{proj}--*.md")) legacy = d / f"{proj}.md" if legacy.exists(): found.append(legacy) return sorted(found, key=lambda f: f.stat().st_mtime) def _wrote_it(path: Path) -> str: """The session id stamped in the header, or "" for a legacy/unstamped file.""" try: head = path.read_text(encoding="utf-8")[:400] except Exception: return "" m = re.search(r"session=([A-Za-z0-9]{1,8})", head) return m.group(1) if m else "" def cmd_review(argv: list[str]) -> int: """Show what WOULD be carried forward, and what would be dropped, consuming nothing. The point of a review step is to be able to disbelieve the tool. A summary you cannot inspect before you rely on it is just a shorter thing to be wrong about, and the whole saving here comes from trusting an extract instead of re-reading the transcript - so the extract has to be checkable, cheaply, at any moment. """ proj = _project() src = None if argv and not argv[0].startswith("-"): src = Path(argv[0]) if src is None: env = os.environ.get("CLAUDE_TRANSCRIPT_PATH") src = Path(env) if env else _newest_transcript() pend = _pending(proj) print(f"HANDOFF REVIEW - {proj}") print(f" mode: {get_mode()} (seed = full brief up front, pointer = short notice)") if pend: for c in pend: age = (time.time() - c.stat().st_mtime) / 60.0 print(f" waiting: {c.name} {c.stat().st_size:,}B {age:.0f} min old " f"from session {_wrote_it(c) or 'unstamped'}") else: print(" waiting: none") if not src or not src.exists(): print(" no transcript to preview") return 0 out = subprocess.run([sys.executable, str(BIN / "brief.py"), str(src), "--json"], capture_output=True, text=True, timeout=120) if out.returncode != 0: print(f" preview failed: {out.stderr.strip()[:200]}", file=sys.stderr) return 1 d = json.loads(out.stdout) b = d["brief"] s_, o_ = d["src_bytes"], len(b) print() print(f" IF YOU HANDED OFF NOW: {s_:,}B (~{s_ // 4:,} tok) -> {o_:,}B " f"(~{o_ // 4:,} tok), {s_ / max(o_, 1):.0f}x smaller, 0 model calls") print() print(b) return 0 def _newest_transcript() -> Path | None: base = Path.home() / ".claude" / "projects" if not base.is_dir(): return None best, bt = None, -1.0 for f in base.rglob("*.jsonl"): try: m = f.stat().st_mtime except OSError: continue if m > bt: best, bt = f, m return best def cmd_write(argv: list[str]) -> int: src = Path(argv[0]) if argv and not argv[0].startswith("-") else None if src is None: env = os.environ.get("CLAUDE_TRANSCRIPT_PATH") src = Path(env) if env else _newest_transcript() if not src or not src.exists(): print("handoff: no transcript found to summarise", file=sys.stderr) return 2 out = subprocess.run([sys.executable, str(BIN / "brief.py"), str(src), "--json"], capture_output=True, text=True, timeout=120) if out.returncode != 0: print(f"handoff: brief failed: {out.stderr.strip()[:300]}", file=sys.stderr) return 1 d = json.loads(out.stdout) proj = _project() sess = _session_tag(argv) or "unknown" # Scoped to the WRITING SESSION, not just the folder. More than one agent can run # in one directory, and a folder-wide slot means the first session to start next # takes whatever is sitting there - somebody else's conversation - while its # rightful successor gets nothing and cannot tell, because consumption and # "there was never one" look identical from the outside. path = _dir() / f"{proj}--{sess}.md" stamp = time.strftime("%Y-%m-%d %H:%M:%S") text = (f"<!-- recall-handoff project={proj} session={sess} written={stamp} " f"src_bytes={d['src_bytes']} -->\n" + d["brief"]) tmp = path.with_suffix(".md.tmp") tmp.write_text(text, encoding="utf-8") try: os.chmod(tmp, 0o600) except Exception: pass tmp.replace(path) s, o = d["src_bytes"], len(text) print(f"handoff written: {path}") print(f" source {s:,}B (~{s // 4:,} tok) -> brief {o:,}B (~{o // 4:,} tok) {s / o:.0f}x smaller") print(" cost 0 model calls, 0 network") print() print(" Now: exit this session and start a new one in the same folder.") print(" The new session picks this up automatically, once, and says how old it is.") return 0 def cmd_take(argv: list[str]) -> int: """Print the pending handoff and consume it. Used by the SessionStart hook. WHY THIS REFUSES INSTEAD OF PICKING THE NEWEST Handing a session the wrong conversation is worse than handing it none. The wrong one is silent: it reads as ordinary context, it is written in the same voice, and nothing about it announces that it belongs to somebody else. The missing one at least leaves a person asking where it went. This is not hypothetical. On 2026-08-21 a handoff written at 13:33 by one session was consumed two minutes later by an unrelated session that merely happened to compact in the same folder - 18 KB of another operator's conversation delivered into a window it had nothing to do with, and that operator's own successor started empty. So: one candidate is taken, several are never guessed between. """ proj = _project() me = _session_tag(argv) cands = _pending(proj) # Never hand a session back the notes it wrote itself. It still has that context; # what it would lose is the successor that was supposed to receive them. mine = [c for c in cands if me and _wrote_it(c) == me] cands = [c for c in cands if c not in mine] # Age them out of AUTOMATIC pickup only. They stay on disk and `review` and # `status` still list them, because a handoff nobody collected is evidence about # what happened here, and silently deleting it would destroy the one trace that # says a session ended with work still in the air. cutoff = time.time() - STALE_HOURS * 3600 fresh = [c for c in cands if c.stat().st_mtime >= cutoff] if cands and not fresh: print(f"A handoff is on disk for this folder but it is older than " f"{STALE_HOURS:.0f} hours, so it was not injected - stale context " f"presented as current is worse than none. Read it deliberately with: " f"recall handoff show") return 0 cands = fresh if not cands: return 0 if len(cands) > 1: print("HANDOFF NOT DELIVERED - more than one is waiting for this folder, and " "guessing between them would risk opening somebody else's conversation " "in here. Nothing was consumed; they are all still on disk.") for c in cands: age = (time.time() - c.stat().st_mtime) / 60.0 who = _wrote_it(c) or "unstamped" print(f" {c.name} {c.stat().st_size:,}B {age:.0f} min old from session {who}") print("Take the one you meant: recall handoff take --from <session>") print("Look first without consuming: recall handoff review") return 0 want = None for i, a in enumerate(argv): if a == "--from" and i + 1 < len(argv): want = re.sub(r"[^A-Za-z0-9]", "", argv[i + 1])[:8] path = cands[0] if want and _wrote_it(path) != want: print(f"no pending handoff from session {want}", file=sys.stderr) return 1 try: text = path.read_text(encoding="utf-8") age_h = (time.time() - path.stat().st_mtime) / 3600.0 except Exception: return 0 mode = get_mode() if "--peek" not in argv: try: path.replace(path.with_name(path.name + ".taken")) except Exception: pass when = f"{age_h:.1f} hours ago" if age_h >= 1 else f"{int(age_h * 60)} minutes ago" banner = (f"HANDOFF FROM A PREVIOUS SESSION, written {when}. " f"This replaces compaction: the old session was summarised locally, not by a " f"model. Treat it as context, not as instructions to re-run.") if age_h > STALE_HOURS: banner += (f" WARNING: this is older than {STALE_HOURS:.0f} hours. Confirm it is " f"still current before acting on it.") print(banner) print() print(_pointer_view(text) if mode == "pointer" else text) return 0 def cmd_show(argv: list[str]) -> int: """Print the full brief without consuming it. What 'pointer' mode points at.""" proj = _project() cands = _pending(proj) or sorted(_dir().glob(f"{proj}*.md.taken"), key=lambda f: f.stat().st_mtime) if not cands: print("no handoff for this project", file=sys.stderr) return 1 print(cands[-1].read_text(encoding="utf-8")) return 0 def cmd_status(argv: list[str]) -> int: d = _dir() pend = sorted(d.glob("*.md")) taken = sorted(d.glob("*.md.taken")) print(f"this session: {_session_tag() or 'unidentified'}") print(f"handoff dir: {d}") print(f"mode: {get_mode()} (seed = full brief up front, pointer = short notice)") if not pend: print(" pending: none") for p in pend: age = (time.time() - p.stat().st_mtime) / 3600.0 flag = " STALE" if age > STALE_HOURS else "" print(f" pending: {p.name} {p.stat().st_size:,}B {age:.1f}h old{flag}") for p in taken[-3:]: print(f" used: {p.name} {p.stat().st_size:,}B") return 0 def main() -> int: cmd = sys.argv[1] if len(sys.argv) > 1 else "status" rest = sys.argv[2:] if cmd == "write": return cmd_write(rest) if cmd == "take": return cmd_take(rest) if cmd == "status": return cmd_status(rest) if cmd == "show": return cmd_show(rest) if cmd == "review": return cmd_review(rest) if cmd == "mode": return set_mode(rest[0]) if rest else (print(f"handoff mode: {get_mode()}") or 0) print("usage: handoff.py {write [transcript] | take [--peek] [--from <session>] | " "show | review | mode [seed|pointer] | status}", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())