bin/capture.py

#!/usr/bin/env python3 """ capture.py - turn a Claude Code transcript into Recall rows. DESIGN DECISION THAT MATTERS ---------------------------- The predecessor generated observations by spawning a hosted-model subprocess per session - roughly 450MB of memory each, and it shipped the transcript off the machine to do it. When that path failed it captured NOTHING, which is why "capture is broken" went unnoticed for hours. Recall does the whole job the other way, with one lane: FACT LANE (the only lane: deterministic, no model, no network) prompts, tool calls, files read/modified, commands, session boundaries. Titles and narratives are derived from those facts by the code in this file. Nothing is sent anywhere to produce them, so there is no second path that can be down, slow, cost money, or quietly fail. That is the whole trade: descriptions are plainer than a model would write, and in exchange capture cannot fail, cannot bill you, and cannot leak. Everything crosses redact.scrub() before it touches disk. """ from __future__ import annotations import argparse import hashlib import json import os import re import sys from datetime import datetime, timezone from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import time as _time # noqa: E402 import store # noqa: E402 from redact import scrub, selftest # noqa: E402 # Tools whose results we deliberately never store the body of. The old store # kept everything a PostToolUse "*" hook saw; that is how 9 chat-bot tokens and # a provider API key ended up in a 203MB file with a cloud outbox attached. _NEVER_BODY = {"Read", "NotebookRead"} _WRITE_TOOLS = {"Edit", "Write", "NotebookEdit", "MultiEdit"} # In auto mode most file work happens through Bash, not Edit/Write. A capture # system that only watches the Edit tool reports "0 files modified" for a session # that rewrote a service - the same class of blind spot as a check that cannot # fire. These patterns recover the paths Bash actually touched. _BASH_WRITE = re.compile( r"""(?x) (?:^|[|;&]|then|do)\s* (?: (?:cat|tee|printf|echo)\s[^|;&]*?>{1,2}\s*(?P<redir>[^\s|;&<>]+) | sed\s+-[a-zA-Z]*i[a-zA-Z]*\s[^|;&]*?\s(?P<sed>[^\s|;&<>]+)\s*$ | (?:cp|mv|install)\s+(?:-[^\s]+\s+)*[^\s]+\s+(?P<dest>[^\s|;&<>]+) | (?:mkdir|touch)\s+(?:-[^\s]+\s+)*(?P<touch>[^\s|;&<>]+) | tee\s+(?:-a\s+)?(?P<tee>[^\s|;&<>]+) )""") def _bash_writes(cmd: str) -> set[str]: """Best-effort recovery of paths a shell command wrote to.""" out = set() for m in _BASH_WRITE.finditer(cmd): for g in ("redir", "sed", "dest", "touch", "tee"): v = m.group(g) if not v: continue v = v.strip("\"'") # /dev/null and friends are not artifacts if v.startswith("/dev/") or v in ("-", "") or v.startswith("$"): continue out.add(v) return out _READ_TOOLS = {"Read", "Grep", "Glob", "NotebookRead"} def _iso(ts: str | None) -> tuple[str, int]: if not ts: d = datetime.now(timezone.utc) else: try: d = datetime.fromisoformat(ts.replace("Z", "+00:00")) except Exception: d = datetime.now(timezone.utc) return d.isoformat(), int(d.timestamp() * 1000) # Harness-injected text arrives as type "user" but is not the user talking. # Storing it means the context block fills with <task-notification> noise # instead of what was actually asked for. _NOISE_PREFIX = ( "<system-reminder>", "<task-notification>", "<local-command-stdout>", "<local-command-stderr>", "<command-name>", "<command-message>", "<command-args>", "[Request interrupted", "Caveat: The messages below", "<user-prompt-submit-hook>", ) def _is_noise(txt: str) -> bool: t = txt.lstrip() return (not t) or t.startswith(_NOISE_PREFIX) def _blocks(msg) -> list: if isinstance(msg, dict): c = msg.get("content") if isinstance(c, list): return c if isinstance(c, str): return [{"type": "text", "text": c}] return [] _REDACTION_TAG = re.compile(r"\[REDACTED:[a-z0-9-]+\]") def _redactions_in(*values: str) -> int: """How many redactions are visible in the text actually stored. Counted from the written value, never from how many times scrub() ran. The boundary scrub runs over every file path and every command in a transcript, and those collapse into sets and per-turn digests before they reach a row -- so counting calls reported 87 redactions for a store containing 3. A number a user can verify by eye in their own store is the only honest one to print. """ return sum(len(_REDACTION_TAG.findall(v)) for v in values if isinstance(v, str)) def parse_transcript(path: Path) -> dict: """Deterministic extraction. No model involved.""" prompts, tools = [], [] files_read, files_mod, commands = set(), set(), [] # Per-turn attribution. Session-wide sets are still kept for the session row, # but an OBSERVATION must only claim what ITS turn touched — otherwise every # observation in a session inherits the whole session's file list, and # `recall-digest --file X` returns every turn of every session that ever # touched X. Observed 2026-08-20: one session had 30 observations all # claiming the same 21 files. per_turn_read: dict[int, set] = {} per_turn_mod: dict[int, set] = {} started = ended = None session_uid = path.stem title = None pn = 0 red_total = 0 # scrub() calls made at the boundary; NOT what a row # reports - see _redactions_in for why those differ def _clean(v): """Redact BEFORE anything downstream truncates or stores this. Two real leaks, both found 2026-08-20 against this build with planted values, are closed by scrubbing here instead of further down: * TRUNCATE-BEFORE-SCRUB. _title_for cuts the title to 120 chars, and ingest scrubbed only afterwards. A cut landing inside a token left a prefix too short for its own pattern to match, so it survived the scrub: 'ghp_<36 chars>' was stored as 'ghp_C'. Slicing raw text can only ever make a secret harder to recognise, never safer. * FILE PATHS WERE NEVER SCRUBBED AT ALL. files_read / files_modified went into the row raw, so a path like /tmp/AKIA<16>/out.txt stored a complete, working credential. Redacting once at the boundary makes both structural: nothing raw reaches a truncation, and no column can leak what never entered. """ nonlocal red_total if not v: return v out, hits, _ = scrub(v) red_total += hits return out # encoding NAMED, not left to the platform. Without it, Windows reads the # transcript in the machine's legacy code page and errors="replace" then # SILENTLY substitutes every character it cannot map - so a conversation in # any language but English would be stored corrupted, with no error and no # way to tell afterwards. Quiet damage to the memory itself. for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): if not line.strip(): continue try: d = json.loads(line) except Exception: continue t = d.get("type") ts = d.get("timestamp") if ts: started = started or ts ended = ts if t == "ai-title" and not title: title = _clean(d.get("title") or d.get("aiTitle")) elif t == "user": for b in _blocks(d.get("message")): if b.get("type") == "text": raw = (b.get("text") or "").strip() txt, hits, _ = scrub(raw) if raw else ("", 0, None) red_total += hits if _is_noise(txt): continue pn += 1 prompts.append({"n": pn, "text": txt, "ts": ts}) elif t == "assistant": for b in _blocks(d.get("message")): if b.get("type") != "tool_use": continue name = b.get("name", "?") inp = b.get("input") or {} tools.append({"name": name, "ts": ts, "prompt_number": pn}) tmod = per_turn_mod.setdefault(pn, set()) tread = per_turn_read.setdefault(pn, set()) fp = _clean(inp.get("file_path") or inp.get("notebook_path")) if fp: if name in _WRITE_TOOLS: files_mod.add(fp); tmod.add(fp) else: files_read.add(fp); tread.add(fp) elif name in _READ_TOOLS and inp.get("path"): _p = _clean(inp["path"]) files_read.add(_p); tread.add(_p) if name == "Bash" and inp.get("command"): _cmd = _clean(inp["command"]) w = _bash_writes(_cmd) files_mod.update(w); tmod.update(w) commands.append({ "cmd": _cmd[:2000], "desc": _clean(inp.get("description", "")), "ts": ts, "prompt_number": pn, }) return { "session_uid": session_uid, "title": title, "prompts": prompts, "tools": tools, "files_read": sorted(files_read), "files_modified": sorted(files_mod), "commands": commands, "started_at": started, "ended_at": ended, "per_turn_read": per_turn_read, "per_turn_mod": per_turn_mod, "redactions": red_total, } # --- title construction ----------------------------------------------------- # The first version used the raw first line of the prompt. That reads like a chat # log ("run it all you pick order"), not a work log, and six months later it tells # you nothing about what actually happened. A title should survive the loss of its # context: intent first, outcome appended when the intent alone is not self-carrying. _STOPWORDS = re.compile( r"(?i)^(?:ok(?:ay)?|now|so|and|also|then|can you|could you|please|pls|" r"lets?|let's|go ahead and|i want (?:you )?to|i need (?:you )?to|" r"we need to|you (?:can|should)|hey|yo)\s+") def _clean_intent(text: str) -> str: line = "" for raw in text.strip().splitlines(): raw = raw.strip() if raw and not raw.startswith(("<", "#", "```")): line = raw break if not line: line = " ".join(text.split()) prev = None while prev != line: # strip stacked lead-ins prev = line line = _STOPWORDS.sub("", line).strip() line = " ".join(line.split()).rstrip(" .,:;!?-") return line[:1].upper() + line[1:] if line else "" def _outcome(files_mod: list[str], cmds: list[dict]) -> str: bits = [] if files_mod: names = [Path(f).name for f in files_mod] head = ", ".join(names[:3]) bits.append(head + (f" +{len(names)-3} more" if len(names) > 3 else "")) if cmds: bits.append(f"{len(cmds)} command{'s' if len(cmds) != 1 else ''}") return " | ".join(bits) def _title_for(prompt: str, files_mod: list[str], cmds: list[dict]) -> str: intent = _clean_intent(prompt) out = _outcome(files_mod, cmds) # A short or vague intent cannot stand alone - graft the outcome onto it. if out and (len(intent) < 34 or len(intent.split()) < 5): combined = f"{intent} - {out}" if intent else out return combined[:120] return (intent or out or "(no description)")[:120] def _classify(cmds: list[dict], files_mod: list[str]) -> str: joined = " ".join(c["cmd"] for c in cmds).lower() if any(k in joined for k in ("pytest", "selftest", "--test", "npm test")): return "verification" if any(k in joined for k in ("git commit", "git push")): return "change" if any(k in joined for k in ("systemctl", "service ", "restart")): return "operation" if files_mod: return "change" return "discovery" def build_observations(p: dict) -> list[dict]: """One observation per prompt turn - the unit the user actually thinks in.""" obs = [] by_pn: dict[int, dict] = {} for c in p["commands"]: by_pn.setdefault(c["prompt_number"], {"cmds": [], "tools": []})["cmds"].append(c) for t in p["tools"]: by_pn.setdefault(t["prompt_number"], {"cmds": [], "tools": []})["tools"].append(t) for pr in p["prompts"]: n = pr["n"] bucket = by_pn.get(n, {"cmds": [], "tools": []}) if not bucket["tools"]: continue # pure conversation turn, nothing done tool_names = [t["name"] for t in bucket["tools"]] counts: dict[str, int] = {} for tn in tool_names: counts[tn] = counts.get(tn, 0) + 1 mod = sorted(p["per_turn_mod"].get(n, set())) read = sorted(p["per_turn_read"].get(n, set())) title = _title_for(pr["text"], mod, bucket["cmds"]) obs.append({ "prompt_number": n, "type": _classify(bucket["cmds"], mod), "title": title, "subtitle": _outcome(mod, bucket["cmds"]) or ", ".join(f"{k}x{v}" for k, v in sorted(counts.items())), "body": pr["text"][:4000], "facts": json.dumps({ "tools": counts, "commands": [c["desc"] or c["cmd"][:120] for c in bucket["cmds"]][:20], }), "files_modified": json.dumps(mod[:60]), "files_read": json.dumps(read[:60]), "ts": pr["ts"], }) return obs def import_cutover(con) -> int: """Newest moment already covered by imported history. An earlier tool summarised everything up to its cutover; those rows live here as source='legacy-import' with better, model-written titles. Re-deriving the same turns from the transcripts would double every entry. This guard used to live ONLY in rebuild.sh, so `capture.py --all` bypassed it and silently re-added 3,229 duplicates. A rule that governs what may be written belongs where the writing happens, not in one script that happens to call it. """ r = con.execute( "SELECT MAX(created_at_epoch) FROM observations WHERE source='legacy-import'" ).fetchone() return int(r[0]) if r and r[0] else 0 def ingest(path: Path, project: str, con=None) -> dict: if selftest(verbose=False) != 0: raise SystemExit("ABORT: redaction selftest failed - refusing to capture") p = parse_transcript(path) own = con is None con = con or store.connect() cutover = import_cutover(con) s_iso, s_ep = _iso(p["started_at"]) e_iso, e_ep = _iso(p["ended_at"]) title, _, _ = scrub(p["title"] or "") # scrub-then-cut, never cut-then-scrub (see _clean in parse_transcript) opening = (scrub(p["prompts"][0]["text"])[0][:500]) if p["prompts"] else "" con.execute("""INSERT INTO sessions (session_uid,content_session_id,project,platform,title,opening_prompt, started_at,started_at_epoch,ended_at,ended_at_epoch,status,prompt_count) VALUES (?,?,?,'claude',?,?,?,?,?,?,'completed',?) ON CONFLICT(session_uid) DO UPDATE SET ended_at=excluded.ended_at, ended_at_epoch=excluded.ended_at_epoch, prompt_count=excluded.prompt_count, title=COALESCE(sessions.title,excluded.title)""", (p["session_uid"], p["session_uid"], project, title or None, opening, s_iso, s_ep, e_iso, e_ep, len(p["prompts"]))) # Redaction happens once, at the parse boundary; this is that count. # The scrubs below are idempotent re-checks and add 0 on clean input. n_p = n_o = 0 red = _redactions_in(title or "", opening) for pr in p["prompts"]: if cutover and _iso(pr["ts"])[1] <= cutover: continue body, _, _ = scrub(pr["text"]) # idempotent re-check on boundary-clean text h = _redactions_in(body) i_iso, i_ep = _iso(pr["ts"]) key = int(hashlib.sha1( f"{p['session_uid']}:{pr['n']}".encode()).hexdigest()[:12], 16) con.execute("""INSERT OR IGNORE INTO prompts (session_uid,project,prompt_number,body,created_at,created_at_epoch, redactions,source,legacy_id) VALUES (?,?,?,?,?,?,?,'transcript',?)""", (p["session_uid"], project, pr["n"], body, i_iso, i_ep, h, -key)) red += h # rowcount, not +1. INSERT OR IGNORE silently stores nothing when the row is # already there, so counting attempts made a re-ingest of the same transcript # report rows it did not write. That number is what `recall import` prints # back as proof the import landed - a count that overstates its work is the # one number here that must never be generous. n_p += con.execute("SELECT changes()").fetchone()[0] for o in build_observations(p): if cutover and _iso(o["ts"])[1] <= cutover: continue parts = {} for f in ("title", "subtitle", "body", "facts"): v, _, _ = scrub(o[f] or "") parts[f] = v # Every redaction visible in this row. The old code stored only the LAST # field's count, so a secret in the title vanished from the tally. h = _redactions_in(*parts.values(), o["files_read"], o["files_modified"]) i_iso, i_ep = _iso(o["ts"]) key = int(hashlib.sha1( f"{p['session_uid']}:obs:{o['prompt_number']}".encode()).hexdigest()[:12], 16) con.execute("""INSERT OR IGNORE INTO observations (session_uid,project,type,title,subtitle,body,facts,files_read, files_modified,prompt_number,created_at,created_at_epoch,redactions, source,legacy_id) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,'transcript',?)""", (p["session_uid"], project, o["type"], parts["title"], parts["subtitle"], parts["body"], parts["facts"], o["files_read"], o["files_modified"], o["prompt_number"], i_iso, i_ep, h, -key)) red += h n_o += con.execute("SELECT changes()").fetchone()[0] # see the note above con.commit() if own: con.close() # Report what was STORED, not what was parsed. Returning len(p["prompts"]) # made a run that wrote 42 rows print "prompts: 7256" — a counter that # overstates its work is how people stop believing the output. return {"session": p["session_uid"], "prompts_seen": len(p["prompts"]), "prompts_stored": n_p, "observations_stored": n_o, "files_modified": len(p["files_modified"]), "redactions": red} def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("transcript", nargs="?") ap.add_argument("--project", default=None) ap.add_argument("--all", action="store_true", help="ingest every transcript for the project") a = ap.parse_args() proj = a.project or Path.cwd().name or "home" if a.all: sys.path.insert(0, str(Path(__file__).resolve().parent)) from recall import transcript_files con = store.connect() tot = {"prompts_seen": 0, "prompts_stored": 0, "observations_stored": 0, "redactions": 0} files = transcript_files() for i, f in enumerate(files, 1): try: r = ingest(f, proj, con) for k in tot: tot[k] += r[k] except Exception as e: print(f" skip {f.name}: {e}", file=sys.stderr) if i % 25 == 0: print(f" {i}/{len(files)}...", flush=True) con.close() print(json.dumps({"transcripts": len(files), **tot}, indent=2)) return 0 if not a.transcript: ap.error("give a transcript path or --all") print(json.dumps(ingest(Path(a.transcript), proj), indent=2)) return 0 if __name__ == "__main__": sys.exit(main())