bin/hook.py

#!/usr/bin/env python3 """ hook.py - the only thing Claude Code calls. Four events, all cheap. SessionStart : print the recent-context block, plus any pending handoff PreCompact : hand the summariser a precomputed brief so it reads less SessionEnd : leave a handoff behind so the next session can skip compaction Stop : ingest the finished transcript WHAT THIS DELIBERATELY DOES NOT HOOK ------------------------------------ PostToolUse "*" - the firehose the predecessor used; every tool result crossed it PreToolUse Read - every file you open UserPromptSubmit - a subprocess spawned on every keystroke-submit Those three are why the old system cost ~1.9GB resident and put 9 chat-bot tokens, a provider API key and 13 URL credentials into cleartext storage. Recall reads the transcript Claude Code already wrote, at Stop, once. A hook must never break the session. Every path exits 0; real failures are recorded and surfaced by `recall health`, which can actually fail. """ from __future__ import annotations import json import os import subprocess import sys import time from pathlib import Path BIN = Path(__file__).resolve().parent from paths import log_file LOG = log_file() TIMEOUT = 45 def _log(msg: str) -> None: try: with LOG.open("a", encoding="utf-8") as f: f.write(f"{time.strftime('%Y-%m-%dT%H:%M:%S')} {msg}\n") except Exception: pass _PAYLOAD_READ = False def _payload() -> dict: """Read the hook payload from stdin. ONCE. stdin is a stream, not a file you can re-open. main() reads it, and a handler that calls this a second time gets EOF, parses "" to {}, finds no transcript path, and returns quietly - a hook that looks wired, exits 0, logs nothing and does nothing. That is exactly what PreCompact did on its first run here. So the second read is now loud instead of empty. """ global _PAYLOAD_READ if _PAYLOAD_READ: _log("BUG: _payload() called twice - stdin is already consumed. " "Pass the payload down instead of re-reading it.") return {} _PAYLOAD_READ = True try: raw = sys.stdin.read() return json.loads(raw) if raw.strip() else {} except Exception: return {} def _sid(p: dict) -> str: v = str(p.get("session_id") or "") return "".join(c for c in v if c.isalnum())[:8] def _carry(p: dict) -> Path: """Where PreCompact leaves the full brief for the SessionStart that follows it. Kept apart from the handoff queue on purpose. A handoff travels BETWEEN sessions and belongs to whoever continues the work; a carry belongs to one session crossing its own compaction boundary and to nobody else. Same file shape, two different questions about ownership - so two different drawers. """ # paths.root(), never a hand-rolled home directory. The wipe command promises # that deleting one directory leaves nothing of Recall behind, and a carry file # is verbatim conversation text. Writing it anywhere else would make that promise # false in the one place it matters most, and quietly. import paths d = paths.root() / "handoff" d.mkdir(parents=True, exist_ok=True) return d / f"carry--{_sid(p) or 'unknown'}.md" def _standing(brief: str) -> str: """Just the binding-rules section of a brief.""" out, keep = [], False for line in brief.splitlines(): if line.startswith("## STANDING INSTRUCTIONS"): keep = True elif line.startswith("## ") and keep: break if keep: out.append(line) return "\n".join(out).strip() # A session that is killed, crashes, or has its terminal closed never fires # SessionEnd, so the handoff written there is the one you do not get on exactly the # days you needed it. This keeps a current one on disk while the session is alive. # It is cheap enough to do that: local text extraction, no model call, no network. AUTO_MIN_BYTES = 400_000 # below this a fresh session costs less than a handoff AUTO_EVERY_SEC = 600.0 def _auto_handoff(p: dict, tpath: str) -> None: try: sid = _sid(p) if not sid or not tpath: return src = Path(tpath) if not src.exists() or src.stat().st_size < AUTO_MIN_BYTES: return import paths marker = paths.root() / "handoff" / f".auto--{sid}" now = time.time() if marker.exists() and (now - marker.stat().st_mtime) < AUTO_EVERY_SEC: return marker.parent.mkdir(parents=True, exist_ok=True) marker.write_text(str(int(now)), encoding="utf-8") r = subprocess.run( [sys.executable, str(BIN / "handoff.py"), "write", tpath, "--session", sid], capture_output=True, text=True, timeout=TIMEOUT) _log(f"auto-handoff rc={r.returncode} sid={sid} src={src.stat().st_size:,}B") except Exception as e: _log(f"auto-handoff skipped: {e}") def do_session_start(p: dict) -> None: """SessionStart fires for startup, resume, clear, compact AND fork. It used to treat all five the same and swallow whatever handoff was waiting. That is how an unrelated session's 18 KB conversation arrived here on 2026-08-21 at 13:35: this session merely COMPACTED in a folder where somebody else had just left a handoff, and took it. Which door you came in by decides what you are owed, so read it. """ source = str(p.get("source") or "") try: out = subprocess.run( [sys.executable, str(BIN / "recall.py"), "context", "--days", "2", "--limit", "40", "--project", os.environ.get("RECALL_PROJECT", "home")], capture_output=True, text=True, timeout=TIMEOUT) ctx = (out.stdout or "").strip() hand = "" if source == "compact": # Coming back from a compaction. The brief PreCompact computed is sitting # on disk; putting it in here is the whole point, and this is the only # channel that carries it silently - PreCompact's own stdout is echoed # to the terminal verbatim, so the long form cannot travel that way. c = _carry(p) try: if c.exists(): hand = c.read_text(encoding="utf-8").strip() c.replace(c.with_name(c.name + ".used")) except Exception as e: _log(f"carry read failed (continuing without it): {e}") elif source in ("startup", ""): # A genuinely new session, and the only one entitled to a handoff. resume, # clear and fork already hold their own history; handing them somebody # else's would be pure loss, since a taken handoff cannot be un-taken. try: args = [sys.executable, str(BIN / "handoff.py"), "take"] if _sid(p): args += ["--session", _sid(p)] ho = subprocess.run(args, capture_output=True, text=True, timeout=TIMEOUT) hand = (ho.stdout or "").strip() except Exception as e: _log(f"handoff take failed (continuing without it): {e}") if hand: ctx = hand + ("\n\n" + ctx if ctx else "") _log(f"session-start source={source or '?'} sid={_sid(p) or '?'} " f"carried={len(hand)}B ctx={len(ctx)}B") if not ctx: return print(json.dumps({ "hookSpecificOutput": { "hookEventName": "SessionStart", "additionalContext": ctx, } })) except Exception as e: _log(f"session-start FAILED: {e}") def do_stop(p: dict) -> None: tpath = p.get("transcript_path") if not tpath: # fall back to the newest transcript for this project sys.path.insert(0, str(BIN)) try: from recall import transcript_files files = list(reversed(transcript_files())) except Exception: files = [] if not files: _log("stop: no transcript found") return tpath = str(files[0]) if not Path(tpath).exists(): _log(f"stop: transcript missing {tpath}") return try: r = subprocess.run( [sys.executable, str(BIN / "capture.py"), tpath, "--project", os.environ.get("RECALL_PROJECT", "home")], capture_output=True, text=True, timeout=TIMEOUT) if r.returncode != 0: _log(f"stop: capture rc={r.returncode} {r.stderr.strip()[:400]}") else: try: # THESE KEY NAMES ARE THE ONES capture.py ACTUALLY RETURNS. # They were 'prompts' and 'observations', which capture.py has never # returned - so this line raised KeyError on every SUCCESSFUL capture # and fell into the bare fallback below. The counts nobody could find # in hook.log were never being written, and the log's most detailed # line was unreachable code that still looked perfectly fine. A # success path can die as quietly as a failure path. d = json.loads(r.stdout) _log(f"stop: {d['session'][:8]} prompts={d['prompts_stored']} " f"obs={d['observations_stored']} redactions={d['redactions']}") except Exception as e: _log(f"stop: captured (counts unavailable: {e})") except subprocess.TimeoutExpired: _log("stop: capture TIMEOUT") except Exception as e: _log(f"stop: FAILED {e}") # Outside the try above on purpose: a capture that timed out or blew up is # exactly the run where a current handoff is most worth having. _auto_handoff(p, tpath) def do_pre_compact(p: dict) -> None: """Compaction is about to run anyway. Make it read less. ⚠️ THIS HOOK PRINTS PLAIN TEXT. IT IS NOT A JSON HOOK. Getting that wrong is what made it useless from the day it shipped until 2026-08-21, and the failure was invisible from in here. The first version emitted the JSON shape the other hooks use: {"hookSpecificOutput": {"hookEventName": "PreCompact", "additionalContext": ...}} PreCompact has no additionalContext. The harness rejected the whole object with "Hook JSON output validation failed" and compaction proceeded with no help at all - every single time, for every session, while this function raised nothing, exited 0 and logged nothing, because from in here a discarded stdout and a hook that was never called look exactly the same. There was no test that asserted the harness ACCEPTED the output, only that we had produced some. What it actually does, from the shipped client (function vFe): l = hooks.filter(succeeded && !blocked && output.trim()).map(output.trim()) return { newCustomInstructions: l.join("\n\n"), userDisplayMessage: ... } Stdout, verbatim, becomes the summariser's custom instructions. ⚠️ AND IT IS ALSO ECHOED. The same stdout goes into userDisplayMessage, which the client prints to the terminal. A 12 KB brief here would wallpaper the screen on every compaction. So the long form goes to disk for SessionStart(compact) to inject silently, and only a short steer travels this way. If anything below fails it fails quietly and compaction proceeds exactly as it would have. A broken optimiser must never break the session. """ try: t = p.get("transcript_path") or "" if not t or not Path(t).exists(): return out = subprocess.run([sys.executable, str(BIN / "brief.py"), t, "--budget", "12000"], capture_output=True, text=True, timeout=TIMEOUT) b = (out.stdout or "").strip() if not b: return try: c = _carry(p) c.write_text(b, encoding="utf-8") os.chmod(c, 0o600) except Exception as e: _log(f"pre-compact: carry write failed: {e}") rules = _standing(b) heads = [l[3:].split(" (")[0].strip() for l in b.splitlines() if l.startswith("## ")] steer = [ "Recall has already extracted this conversation's spine locally - no model " "call, no network - and the full extract is being handed to the session that " "resumes after this compaction. Do not spend effort re-deriving it from tool " "output; summarise the reasoning and the decisions instead.", f"Sections already captured verbatim: {', '.join(heads)}." if heads else "", "", "PRESERVE THESE EXACTLY. They are rules the user is still bound by, not history:", rules or " (none found in this transcript)", ] text = "\n".join(x for x in steer if x != "") # Echoed to the user's terminal, so keep it to something a person can read. if len(text) > 2400: text = text[:2400].rsplit("\n", 1)[0] + "\n ... (full extract carried separately)" print(text) _log(f"pre-compact: brief={len(b)}B carried, steer={len(text)}B on stdout " f"(plain text -> newCustomInstructions)") except Exception as e: _log(f"pre-compact FAILED (compaction continues normally): {e}") def do_session_end(p: dict) -> None: """Leave a handoff so the NEXT session can start clean instead of compacting.""" try: t = p.get("transcript_path") or "" args = [sys.executable, str(BIN / "handoff.py"), "write"] if t and Path(t).exists(): args.append(t) # Stamp the writer. An anonymous handoff in a shared folder cannot be routed. if _sid(p): args += ["--session", _sid(p)] r = subprocess.run(args, capture_output=True, text=True, timeout=TIMEOUT) _log(f"session-end: handoff write rc={r.returncode} sid={_sid(p) or '?'}") except Exception as e: # lower-case for the same reason: the session still ended and the transcript # is still on disk for the Stop lane to capture. Nothing was lost. _log(f"session-end handoff failed (the session was still captured): {e}") def main() -> int: event = sys.argv[1] if len(sys.argv) > 1 else "" p = _payload() if event == "session-start": do_session_start(p) elif event == "pre-compact": do_pre_compact(p) elif event == "session-end": do_session_end(p) elif event == "stop": do_stop(p) else: # FAILED, deliberately: a hook wired to an event name this does not handle # captures nothing, forever, and used to say so in a line no check read. _log(f"FAILED unknown event {event!r} - a hook is wired to something this " f"does not handle, so it captures nothing. Rewire: python3 install.py") return 0 # never break the session if __name__ == "__main__": try: sys.exit(main()) except Exception as e: _log(f"fatal (suppressed): {e}") sys.exit(0)