bin/recall.py

#!/usr/bin/env python3 """ recall.py - the query surface. Replaces the predecessor's MCP tools with a local CLI that talks to one SQLite file and nothing else. recall search "<query>" full-text across observations, summaries, prompts recall timeline [--days N] what happened, newest first recall show <id> [<id>...] expand specific rows (the get_observations lane) recall context the SessionStart block recall stats store health and size recall health [--strict] does capture actually work? exits nonzero if not """ from __future__ import annotations import argparse import json import re import sqlite3 import os import shutil import sys import time from datetime import datetime, timezone, timedelta from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import store # noqa: E402 from console import ICONS as TYPE_ICON, icon as _icon, rule as _rule # noqa: E402 def transcript_dirs() -> list[Path]: """Every Claude project dir on this box, not a guessed one. This was once hardcoded to a single guessed directory name. On a second machine the real directories were named differently - and there were two of them - so the guess matched NEITHER and capture silently did nothing while reporting success. Deriving the name from $USER fails the same way: a machine can have several project directories and none of them need match the user name. Scan them all; a directory that exists is not a guess. """ root = Path.home() / ".claude/projects" if not root.is_dir(): return [] return [d for d in root.iterdir() if d.is_dir()] def transcript_files() -> list[Path]: """All transcripts on this box, newest last.""" out: list[Path] = [] for d in transcript_dirs(): out.extend(d.glob("*.jsonl")) return sorted(out, key=lambda f: f.stat().st_mtime) def _terms(q: str) -> list[str]: """The user's words, punctuation-safe. FTS5 treats bare punctuation as syntax.""" return [x for x in q.strip().replace("/", " ").split() if x] def _quote(term: str) -> str: return '"' + term.replace('"', '""') + '"' def _is_raw_fts(q: str) -> bool: """The caller is writing real FTS5 syntax; hand it through untouched.""" return any(c in q for c in '"*()') or " OR " in q or " NEAR" in q def _fts_escape(q: str) -> str: """Kept for callers that want the plain AND form of a query.""" q = q.strip() if not q: return '""' if _is_raw_fts(q): return q return " ".join(_quote(t) for t in _terms(q)) # HOW A QUERY IS RELAXED, AND WHY THIS IS THE MOST IMPORTANT CODE IN THE FILE # --------------------------------------------------------------------------- # Every term used to be quoted and joined with a space, which FTS5 reads as AND. # So the whole sentence had to appear in one row. A person does not search that # way. They type "why did the cron timer stop firing" and every one of those six # words has to be present or they get: # # no matches for 'why did the cron timer stop firing' # # on a store that plainly contains the answer. That reply is worse than useless: # it tells them Recall has nothing, so they go and re-explain the whole thing to # a fresh session by hand - spending exactly the tokens this tool exists to save. # For someone on a fixed monthly allowance that is the entire cost of the product # being wrong. # # So a miss is not accepted until three passes have failed, and whichever pass # answers is NAMED in the output. Widening a search silently would be its own # kind of lie - the user must be able to see that "all your words" became "some # of your words", or they cannot judge the results. # ORDER MATTERS AS MUCH AS THE PASSES DO. Widening happens one notch at a time and # stops at the first pass that answers, so the narrowest useful reading of the query # always wins. Dropping straight to OR would "succeed" on the word "the" and bury the # real answer in noise - a wall of irrelevant hits is its own kind of no-result. _PASSES = ( ("all of your words", lambda ts, meaty: " ".join(_quote(x) for x in ts)), ("your words as prefixes", lambda ts, meaty: " ".join(_quote(x) + "*" for x in ts)), ("the meaningful words", lambda ts, meaty: " ".join(_quote(x) for x in meaty)), ("meaningful words as prefixes", lambda ts, meaty: " ".join(_quote(x) + "*" for x in meaty)), ("any meaningful word", lambda ts, meaty: " OR ".join(_quote(x) for x in meaty)), ) # Words that carry no signal but, under AND, are enough on their own to force a # miss. Dropped only in the last-resort pass, and only when something remains. _STOP = {"a", "an", "the", "is", "was", "were", "did", "do", "does", "why", "what", "when", "how", "to", "of", "in", "on", "for", "and", "it", "that", "this", "i", "we", "my", "our", "me", "you", "with", "from", "at", "by", "be"} _WARNED_TABLES: set[str] = set() def _search_tables(con, match: str, a) -> list[tuple]: """Run one MATCH expression across every indexed table. Returns tagged rows. Summaries are queried here. They were indexed, populated by a working trigger, documented at the top of this file as searchable - and no command ever read the index. For anyone who keeps notes across sessions, that table holds the most condensed account of what was decided, and it was the one table search could not see. """ where, params = [], [] if getattr(a, "project", None): where.append("x.project = ?"); params.append(a.project) if getattr(a, "since_epoch", None): where.append("x.created_at_epoch >= ?"); params.append(a.since_epoch) filt = (" AND " + " AND ".join(where)) if where else "" cap = max(a.limit * 4, 40) # over-fetch, then rank and trim once out: list[tuple] = [] specs = [ ("obs", f"""SELECT x.id, x.type, x.title, x.subtitle, x.created_at, x.created_at_epoch, x.project, x.session_uid, bm25(observations_fts, 8.0, 4.0, 2.0, 1.0) AS rank, snippet(observations_fts, -1, '<<', '>>', ' ... ', 12) AS snip FROM observations_fts JOIN observations x ON x.id = observations_fts.rowid WHERE observations_fts MATCH ?{filt} ORDER BY rank LIMIT ?"""), ("summary", f"""SELECT x.id, x.project, x.created_at, x.created_at_epoch, x.request, x.learned, x.completed, bm25(summaries_fts) AS rank, snippet(summaries_fts, -1, '<<', '>>', ' ... ', 12) AS snip FROM summaries_fts JOIN summaries x ON x.id = summaries_fts.rowid WHERE summaries_fts MATCH ?{filt} ORDER BY rank LIMIT ?"""), ] if getattr(a, "prompts", False): specs.append(("prompt", f"""SELECT x.id, x.body, x.created_at, x.created_at_epoch, x.project, bm25(prompts_fts) AS rank, snippet(prompts_fts, -1, '<<', '>>', ' ... ', 12) AS snip FROM prompts_fts JOIN prompts x ON x.id = prompts_fts.rowid WHERE prompts_fts MATCH ?{filt} ORDER BY rank LIMIT ?""")) for kind, sql in specs: try: for r in con.execute(sql, (match, *params, cap)): out.append((kind, r)) except sqlite3.OperationalError as e: # A table that is missing or not indexed must not silently remove a whole # class of result. Say so, keep the rest. if "no such table" in str(e): # Once per query, not once per relaxation pass - five identical # warnings for one search reads like five separate faults. if kind not in _WARNED_TABLES: _WARNED_TABLES.add(kind) print(f" (the {kind} index is missing - those rows cannot be " f"searched. Rebuild it with: python3 bin/store.py)", file=sys.stderr) continue raise return out def _tag(kind: str, rid: int) -> str: """Display id. Prompts and summaries autoincrement from 1 in their own tables, exactly like observations, so a bare number was ambiguous: search printed prompt 1, the user typed `recall show 1`, and got observation 1 instead - a different row, no error, nothing to notice.""" return str(rid) if kind == "obs" else f"{kind[0]}{rid}" def _dedupe(rows: list[tuple]) -> list[tuple]: """Identical captures land repeatedly - INSERT OR IGNORE only stops exact hash collisions, not the same title recorded in two sessions. Printing the same line three times spends the reader's attention for nothing.""" seen, out = set(), [] for kind, r in rows: keys = r.keys() text = (r["title"] if "title" in keys else r["request"] if "request" in keys else r["body"] if "body" in keys else "") or "" k = (kind, " ".join(text.lower().split())[:160]) if k in seen: continue seen.add(k); out.append((kind, r)) return out def _clean_snip(snip: str, title: str) -> str: """The snippet is evidence, but only when it says something the title did not. The old guard compared the raw snippet against the title. The snippet always carries the << >> highlight markers, so it could never equal the title, so the guard never once fired and every title match printed its own title back as proof of itself. """ s = " ".join((snip or "").split()) if not s: return "" bare = s.replace("<<", "").replace(">>", "").strip(" .") if not bare or bare.lower() in " ".join((title or "").lower().split()): return "" if len(s) > 160: # trim on a space, never mid-word cut = s[:160].rsplit(" ", 1)[0] s = cut + " ..." return s def cmd_search(a) -> int: con = store.connect(readonly=True) raw = a.query.strip() a.since_epoch = None if getattr(a, "since", None): a.since_epoch = int( (datetime.now(timezone.utc) - timedelta(days=a.since)).timestamp() * 1000) rows: list[tuple] = [] how = "" try: if _is_raw_fts(raw): rows = _search_tables(con, raw, a) else: terms = _terms(raw) if not terms: print("give me something to search for: recall search \"cron timer\"") return 2 meaty = [x for x in terms if x.lower() not in _STOP] or terms seen_expr, attempts = set(), [] for lbl, fn in _PASSES: expr = fn(terms, meaty) if expr in seen_expr: # identical to an earlier pass; skip continue seen_expr.add(expr) attempts.append((lbl, expr)) for lbl, expr in attempts: rows = _search_tables(con, expr, a) if rows: how = lbl break except sqlite3.OperationalError as e: print(f"search syntax not understood by FTS5: {e}", file=sys.stderr) print(f'try quoting it: recall search \'"{a.query}"\'', file=sys.stderr) return 1 except Exception as e: print(f"search error: {e}", file=sys.stderr) return 1 if not rows: print(f"no matches for {a.query!r}") # An empty store and an empty result look identical to the person reading # them, and the fix is completely different. Say which one this is. try: tot = con.execute("SELECT COUNT(*) FROM observations").fetchone()[0] except Exception: tot = 0 if not tot: print("\n Your store is empty - nothing has been captured yet.") print(" Recall saves a session when it ENDS, so the first entry appears") print(" after you finish a session with the hooks installed.") print(" To bring in what you already have: recall import") elif getattr(a, "project", None) or getattr(a, "since", None): print(f" ({tot:,} observations stored, but the filters excluded them -") print(" try again without --project / --since)") return 0 rows = _dedupe(rows) # RANK WITHIN A TABLE, THEN MERGE. # # bm25 scores are only comparable inside one index. Observations are weighted # (title counts for more than body), summaries and prompts are not, so sorting the # raw numbers together let observations outrank everything else by construction - # and with a small --limit the summaries were pushed off the end of the list # regardless of how well they matched. Convert each table's score to a position # within its own results, and merge on that. per: dict[str, list] = {} for kind, r in rows: per.setdefault(kind, []).append(r) merged = [] for kind, rs in per.items(): rs.sort(key=lambda r: (r["rank"], -(r["created_at_epoch"] or 0))) for i, r in enumerate(rs): merged.append((i, kind != "obs", kind, r)) # position first; ties break towards observations, then newest. merged.sort(key=lambda m: (m[0], m[1], -(m[3]["created_at_epoch"] or 0))) rows = [(kind, r) for _, _, kind, r in merged] # ONE budget across all tables, not each. Floored at 1: --limit 0 used to print # "0 match(es)" while hits existed, which reads exactly like "nothing found". rows = rows[:max(1, a.limit)] if a.json: # The spread goes FIRST. Written the other way round, dict(r) put the raw # integer back over the tagged id and every JSON consumer saw the ambiguous # value the tag exists to remove. print(json.dumps([{**dict(r), "id": _tag(k, r["id"]), "row_id": r["id"], "kind": k} for k, r in rows], indent=2, default=str)) return 0 if how and how != _PASSES[0][0]: print(f" (no row had all of your words - showing matches on {how})\n") multi = len({r["project"] for _, r in rows if r["project"]}) > 1 for kind, r in rows: keys = r.keys() when = (r["created_at"] or "")[:16].replace("T", " ") tag = _tag(kind, r["id"]) proj = f" [{r['project']}]" if multi and r["project"] else "" if kind == "obs": print(f"{tag:>7} {when} {_icon(r['type'])} {r['title']}{proj}") snip = _clean_snip(r["snip"], r["title"]) if snip and not a.quiet: print(f" {snip}") if "subtitle" in keys and r["subtitle"] and a.verbose: print(f" {r['subtitle']}") elif kind == "summary": head = " ".join((r["request"] or r["learned"] or "session summary").split()) print(f"{tag:>7} {when} \U0001f4dd {head[:110]}{proj}") snip = _clean_snip(r["snip"], head) if snip and not a.quiet: print(f" {snip}") else: body = " ".join((r["body"] or "").split())[:110] print(f"{tag:>7} {when} \U0001f5e3 {body}{proj}") print(f"\n{len(rows)} match(es). Expand with: recall show <id>") if any(k != "obs" for k, _ in rows): print(" ids starting s = session summary, p = your prompt") return 0 def cmd_timeline(a) -> int: con = store.connect(readonly=True) since = int((datetime.now(timezone.utc) - timedelta(days=a.days)).timestamp() * 1000) cur = con.execute(""" SELECT id,type,title,subtitle,created_at,project,session_uid FROM observations WHERE created_at_epoch >= ? ORDER BY created_at_epoch DESC LIMIT ?""", (since, a.limit)) day = None n = 0 for r in cur: d = r["created_at"][:10] if d != day: day = d print(f"\n{_rule(2)} {d} " + _rule(40)) ic = _icon(r["type"]) print(f"{r['id']:>7} {r['created_at'][11:16]} {ic} {r['title']}") n += 1 print(f"\n{n} entries over {a.days}d") return 0 def _show_summary(r) -> None: print("=" * 72) print(f"#s{r['id']} session summary {r['created_at']} project={r['project']}") for f in ("request", "investigated", "learned", "completed", "next_steps", "notes"): if f in r.keys() and r[f]: print(f"\n[{f}]\n{r[f][:3000]}") def _show_prompt(r) -> None: print("=" * 72) print(f"#p{r['id']} your prompt {r['created_at']} project={r['project']}") print(f"\n{r['body'][:3000]}") def cmd_show(a) -> int: con = store.connect(readonly=True) for raw in a.ids: oid = str(raw).strip().lstrip("#") # show prints "#s1"; accept it back # Tagged ids from search. Without this, `recall show 1` on a prompt hit # returned observation 1 - a different row, silently, with no error. table, fetch = "observations", _show_prompt if oid[:1] == "s" and oid[1:].isdigit(): table, oid, fetch = "summaries", oid[1:], _show_summary elif oid[:1] == "p" and oid[1:].isdigit(): table, oid, fetch = "prompts", oid[1:], _show_prompt elif not oid.isdigit(): print(f"{raw}: not an id (expected a number, or s123 / p123 from search)") continue try: r = con.execute(f"SELECT * FROM {table} WHERE id=?", (oid,)).fetchone() except sqlite3.OperationalError as e: print(f"{raw}: cannot read {table} ({e})") continue if not r: print(f"{raw}: not found") continue if table != "observations": fetch(r) continue print("=" * 72) print(f"#{r['id']} {r['type']} {r['created_at']} project={r['project']}") print(f"session: {r['session_uid']}") print(f"\n{r['title']}") if r["subtitle"]: print(f" {r['subtitle']}") for f in ("body", "narrative", "facts"): if r[f]: print(f"\n[{f}]\n{r[f][:3000]}") for f in ("files_modified", "files_read"): if r[f] and r[f] not in ("[]", "null"): try: fl = json.loads(r[f]) except Exception: fl = [r[f]] if fl: print(f"\n[{f}] {len(fl)}") for x in fl[:15]: print(f" {x}") if r["redactions"]: print(f"\n[{r['redactions']} value(s) redacted at capture]") return 0 def _local(iso: str) -> datetime: """Rows are stored UTC. A context block that prints UTC times under a local-time header reads as a clock that is hours wrong.""" try: d = datetime.fromisoformat(iso.replace("Z", "+00:00")) except Exception: return datetime.now() if d.tzinfo is None: d = d.replace(tzinfo=timezone.utc) return d.astimezone() def cmd_context(a) -> int: """The SessionStart block. Cheap by construction: headlines only, with ids the model can expand on demand - the same 'read little, fetch on demand' trade the predecessor made, without the daemon.""" con = store.connect(readonly=True) now = datetime.now(timezone.utc) since = int((now - timedelta(days=a.days)).timestamp() * 1000) # --project FILTERS. It used to appear only in the header line, so a session in # one project was handed a block titled with that project's name and filled with # every other project's rows. This block is charged to every single session that # starts, so that was both wrong and the most expensive place to be wrong. # Falls back to the whole store when the project is unknown - a labelled block of # other people's work is worse than an honest unfiltered one. known = {r[0] for r in con.execute("SELECT DISTINCT project FROM observations")} scoped = a.project in known if scoped: tot = con.execute("SELECT COUNT(*) FROM observations WHERE project=?", (a.project,)).fetchone()[0] recent = con.execute("""SELECT id,type,title,created_at FROM observations WHERE created_at_epoch>=? AND project=? ORDER BY created_at_epoch DESC LIMIT ?""", (since, a.project, a.limit)).fetchall() else: tot = con.execute("SELECT COUNT(*) FROM observations").fetchone()[0] recent = con.execute("""SELECT id,type,title,created_at FROM observations WHERE created_at_epoch>=? ORDER BY created_at_epoch DESC LIMIT ?""", (since, a.limit)).fetchall() if not recent: print("") return 0 # %-I is a glibc extension and raises ValueError on Windows; pad then trim the # hour ourselves so the same line works on both. _lt = now.astimezone() _stamp = f"{_lt.strftime('%Y-%m-%d')} {_lt.strftime('%I').lstrip('0') or '12'}{_lt.strftime(':%M%p')}" scope = f"[{a.project}]" if scoped else "[all projects]" out = [f"# {scope} recent context, {_stamp}", "", f"Legend: {_icon('change')}change {_icon('discovery')}discovery {_icon('operation')}operation {_icon('bugfix')}bugfix {_icon('feature')}feature", "Fetch details: recall show <id> | Search: recall search \"<query>\"", ""] day = None for r in recent: lt = _local(r["created_at"]) d = lt.strftime("%Y-%m-%d") if d != day: day = d out.append(f"### {d}") ic = _icon(r["type"]) out.append(f"{r['id']:>6} {lt.strftime('%H:%M')} {ic} {r['title'][:95]}") out += ["", f"{tot:,} observations on file. Nothing leaves this box."] text = "\n".join(out) # A context block is charged to every session that starts. The predecessor claimed "92% # savings" by showing headlines and fetching bodies on demand; the same trade only # holds if the headline block itself stays small, so it is BOUNDED here rather than # left to grow with the store. ~4 chars/token is the usual rough ratio. budget_chars = a.max_tokens * 4 if len(text) > budget_chars: kept, total_lines = [], 0 for line in out: if total_lines + len(line) > budget_chars: break kept.append(line) total_lines += len(line) + 1 kept.append(f"... trimmed to ~{a.max_tokens} tokens. " f"Full history: recall timeline --days 7") text = "\n".join(kept) print(text) return 0 def cmd_stats(a) -> int: con = store.connect(readonly=True) db = Path(store.DB_PATH) size = sum(f.stat().st_size for f in db.parent.glob(db.name + "*") if f.exists()) print(f"store {db}") print(f"size {size/1e6:.1f} MB") for t in ("sessions", "observations", "summaries", "prompts"): n = con.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0] print(f"{t:11}{n:>9,}") r = con.execute("SELECT MIN(created_at),MAX(created_at) FROM observations").fetchone() # an empty store is a legitimate state (fresh peer, restored box) - report it, # do not crash on it print(f"span {r[0][:10]} .. {r[1][:10]}" if r and r[0] else "span (empty)") red = con.execute("SELECT SUM(redactions) FROM observations").fetchone()[0] or 0 red += con.execute("SELECT SUM(redactions) FROM prompts").fetchone()[0] or 0 print(f"redacted {red} values scrubbed at capture") print("network NONE. This build has no network code at all.") return 0 def _hook_log() -> Path: """hook.log, wherever THIS install keeps it. The internal build this grew from hardcodes /opt/friday/recall/hook.log. A stranger's install has no such path, so the streak check would have silently read nothing and reported clean forever - the exact failure it exists to catch. paths.py is the single source of truth for where this install's data lives. """ from paths import log_file return log_file() def _capture_writepath_faults(con) -> list[str]: """Compile every INSERT capture.py can run, against the schema actually on disk. EXPLAIN *prepares* a statement -- column names and value arity are both validated -- and executes nothing, so this is safe against the live store. This check exists because of 2026-08-20. In the internal build, capture.py's sessions INSERT listed 14 columns and supplied 12 values. Every Stop hook failed for three hours, every session in that window was lost, and `recall health` reported OK the entire time -- because it only ever asked whether the store LOOKED stale, and a store that is merely three hours cold looks exactly like a quiet afternoon. Freshness heuristics cannot see a broken write path; compiling it can. It is also the check that refuses a fix copied from the wrong build. This build's schema has no origin/ingested_at columns, so a patch written for that internal build would fail here as '14 values for 12 columns' -- caught before a release, not after someone's memory is gone. """ import ast faults = [] cap = Path(__file__).resolve().parent / "capture.py" try: tree = ast.parse(cap.read_text(encoding="utf-8")) except Exception as e: return [f"cannot parse capture.py: {e}"] for node in ast.walk(tree): if not (isinstance(node, ast.Constant) and isinstance(node.value, str)): continue sql = node.value if not re.match(r"\s*INSERT\b", sql, re.I): continue try: con.execute("EXPLAIN " + sql, ("x",) * sql.count("?")) except (sqlite3.OperationalError, sqlite3.ProgrammingError) as e: t = re.search(r"INTO\s+(\w+)", sql, re.I) faults.append(f"capture.py:{node.lineno} INSERT INTO " f"{t.group(1) if t else '?'} will not run: {e}") except Exception: pass # dynamic/partial SQL - not ours to judge return faults def _hook_failure_streak() -> tuple[int, str]: """Capture failures recorded since the last clean ingest. Self-clearing by design: one successful ingest resets the count to zero, so old scars never keep this red, and a live break cannot be waited out. The hook is right to exit 0 always -- it must never break a session -- but that makes hook.log the ONLY place a capture failure is written down, and nothing read it. A watchdog that never reads the log it writes is not a watchdog. """ log = _hook_log() if not log.exists(): return 0, "" try: lines = log.read_text(encoding="utf-8", errors="replace").splitlines() except Exception as e: # FAIL CLOSED. Returning 0 here made an unreadable log indistinguishable from # a clean one - the "empty reads as absent" trap. If the record cannot be # read, that is itself the finding. return 1, f"hook.log exists but could not be read: {e}" # EVERY failure line the hook can write, not just the capture ones. # # This used to require "stop:" AND one of three capture strings, which meant the # hook could report `session-start FAILED` - the context block never injected, the # whole point of the product - and health called it clean. A failure the watchdog # is not looking for is a failure nobody sees. # Upper-case FAILED is deliberate. The hook writes "FAILED" for a step that lost # something and lower-case "failed (continuing without it)" for one that degraded # and carried on. Matching case-insensitively would light this up on every session # that merely skipped an optional handoff, and a warning that is always on is one # nobody reads. "fatal (suppressed)" is listed explicitly: the hook must exit 0 so # it can never break a session, which means the most serious line it can write is # also the quietest. _FAIL = ("capture rc=", "capture TIMEOUT", "FAILED", "no transcript found", "transcript missing", "BUG:", "Traceback", "fatal (suppressed)") _CLEAN = re.compile(r"stop: \S+ prompts=") # PER LANE. A single counter could not do this job: a Stop always follows a # SessionStart in the same session, so one clean ingest wiped the SessionStart # failures too - and a permanently broken context block, which is the whole # product, could never raise the streak above zero. A success in one lane is not # evidence about another lane. lanes: dict[str, list] = {"stop": [0, ""], "other": [0, ""]} for ln in lines: lane = "stop" if ln.lstrip().startswith("stop:") else "other" if _CLEAN.search(ln) or "stop: captured" in ln: lanes["stop"] = [0, ""] # a clean ingest clears the CAPTURE lane only elif any(f in ln for f in _FAIL): lanes[lane][0] += 1 lanes[lane][1] = ln.strip() total = lanes["stop"][0] + lanes["other"][0] last = lanes["other"][1] or lanes["stop"][1] return total, last def _hook_health() -> list[str]: """Check every Recall hook in the settings file points at a file that exists.""" import json as _json out: list[str] = [] settings = Path.home() / ".claude" / "settings.json" if not settings.exists(): # This used to land in the "ok" column. No settings file means no hooks, # which means nothing is ever captured automatically - reported as healthy. out.append("WARN no ~/.claude/settings.json - nothing is wired, so no session " "will be captured automatically. Run: python3 install.py") return out try: d = _json.loads(settings.read_text(encoding="utf-8")) except Exception as e: out.append(f"HOOK settings.json unreadable: {e}") return out found = 0 wired: set[str] = set() try: _events = list((d.get("hooks") or {}).items()) except Exception: _events = [] out.append("WARN ~/.claude/settings.json has a 'hooks' section in a shape this " "does not understand - checking it by hand is the only option") for event, blocks in _events: # The file is the user's, and nothing guarantees its shape. This used to walk # it assuming dicts all the way down, so a hand-edited settings file made # `recall health` die with an AttributeError and mint a crash report - the # health command itself becoming the failure it exists to report. try: blocks = list(blocks) except Exception: out.append(f"WARN hooks for {event} are not a list - skipped") continue for b in blocks: if not isinstance(b, dict) or not isinstance(b.get("hooks"), list): out.append(f"WARN a hook entry under {event} has an unexpected shape " f"- skipped") continue for h in b.get("hooks", []): if not isinstance(h, dict): continue # Look in the ARGUMENTS as well as the command. # # Hooks are written in exec form now - the interpreter in `command` # and the script in `args` - because a single quoted command string # is a PowerShell parser error on Windows and never runs at all. # This detector was not updated with it, so it stopped finding the # very hooks the installer had just written, and `recall health` # answered a brand new, correct installation with # "no Recall hooks wired - nothing will be captured" # That is the first command a new user runs. Being told the install # failed when it worked is worse than most real failures. cmd = h.get("command", "") or "" argv = h.get("args") or [] if not isinstance(argv, list): argv = [] blob = cmd + " " + " ".join(str(x) for x in argv) if "hook.py" not in blob: continue m = (re.search(r'"([^"]*hook\.py)"', blob) or re.search(r"(\S*hook\.py)", blob)) if not m: continue path = Path(m.group(1)) found += 1 wired.add(event) if not path.exists(): out.append(f"HOOK {event} points at a MISSING file: {path} " f"- nothing is being captured automatically. " f"Rewire with: python3 install.py") continue # THE INTERPRETER HAS TO EXIST TOO. A hook wired to a python that was # removed - a deleted venv is the ordinary way this happens - fails on # every session, and counting the entry as "present" reported that as # healthy. The command is the thing that runs; check the command. # # ⚠️ DO NOT PUT shlex.split(cmd) BACK WITHOUT posix=False. # In its default POSIX mode shlex treats a backslash as an escape # character, so it eats every separator in a Windows path: # shlex.split(r"C:\Users\me\python.exe") # -> ["C:Usersmepython.exe"] # Health then reported that the interpreter "does not exist" and # declared a perfectly good, brand new installation FAILED - on every # Windows machine, for the first command a new user types. Measured on # Windows, not guessed. # # With hooks in exec form there is nothing to parse at all: the # interpreter IS the command and the arguments are already a list. # Parsing only remains for the older one-string form. exe = "" try: if argv: exe = cmd.strip() elif cmd: import shlex parts = shlex.split(cmd, posix=(os.name != "nt")) exe = (parts[0] if parts else "").strip('"') except Exception: exe = "" if exe and not (Path(exe).exists() or shutil.which(exe)): out.append(f"HOOK {event} runs {exe!r}, which does not exist " f"- that hook fails every time. Rewire: python3 install.py") # WHICH EVENTS, BY NAME. Counting hooks could not tell the difference between # "both wired" and "SessionStart wired, Stop missing" - and Stop is the only one # that writes anything, so that second case captures NOTHING while reporting a # tidy "1 hook(s) wired and present". required = {"SessionStart": "no context is injected into new sessions", "Stop": "NOTHING IS EVER CAPTURED"} for ev, consequence in required.items(): if ev not in wired: out.append(f"HOOK {ev} is not wired - {consequence}. " f"Fix with: python3 install.py") if found == 0: out.append("HOOK no Recall hooks wired - nothing will be captured automatically. " "Fix with: python3 install.py") elif not any(x.startswith("HOOK") for x in out): out.append(f"hooks wired and runnable: {', '.join(sorted(wired))}") return out def _search_probe(con) -> str | None: """Prove SEARCH ACTUALLY RETURNS A ROW. Returns a problem string, or None. Search is the command people use every day, and it could be one hundred percent dead while every other check stayed green: the FTS index is an external-content table, so if its rows are lost - an interrupted rebuild, a partial restore - then `SELECT count(*)` reads through to the content table and still says 6, and FTS5's own 'integrity-check' passes as well. Both of the obvious checks report healthy. Meanwhile every query returns nothing and exits 0, so a broken index and a genuine miss look identical to the person typing. The only thing that catches it is asking the index a question it must answer: take a row we know is there, search for a word from its own title, and require that row back. """ try: r = con.execute("""SELECT id, title FROM observations WHERE title IS NOT NULL AND length(title) > 3 ORDER BY created_at_epoch DESC LIMIT 1""").fetchone() except Exception as e: return f"cannot read observations: {e}" if not r: return "UNPROVEN: no rows yet, so search has nothing to be tested against" # [A-Za-z] pulled an ASCII FRAGMENT out of a non-English word - "Uberprufung" # yielded "berpr", which of course matches nothing - and this check then declared # a perfectly healthy store broken. A probe that invents failures on other # people's languages is worse than no probe: it teaches them to ignore health. # [^\W\d_] is "any letter in any script". words = [w for w in re.findall(r"[^\W\d_]{4,}", r["title"] or "", re.UNICODE)] if not words: return "UNPROVEN: no row has a word long enough to probe with" try: hit = con.execute( """SELECT 1 FROM observations_fts WHERE observations_fts MATCH ? AND rowid = ? LIMIT 1""", ('"' + words[0].replace('"', '""') + '"', r["id"])).fetchone() except Exception as e: return f"SEARCH INDEX UNUSABLE: {e} - rebuild with: python3 bin/store.py" if not hit: return ("SEARCH IS BROKEN: observation " f"{r['id']} is in the store but its own title word " f"{words[0]!r} does not find it. Every search is silently returning " "less than it should. Rebuild the index with: python3 bin/store.py") return None def cmd_health(a) -> int: """Answers 'is capture actually working' with evidence, not a green light. The predecessor's failure mode was silence: its worker died, capture stopped, and the only trace was a CAPTURE_BROKEN file nobody read. So this checks freshness against real transcript activity and fails loudly when the store is stale.""" problems, notes, warnings = [], [], [] con = store.connect(readonly=True) from redact import selftest if selftest(verbose=False) != 0: problems.append("redaction selftest FAILS - capture would write cleartext") else: notes.append("redaction selftest passes") last = con.execute("SELECT MAX(created_at_epoch) FROM observations").fetchone()[0] # "No observations" is NOT the same as "capture never ran". A session whose turns # used no tools produces prompts and a session row but no observation, which is # correct behaviour — and keying the alarm on observations alone made a fresh # Windows install report FAILED on its very first run. Ask whether ANYTHING was # captured, not whether one particular table filled. any_capture = ( con.execute("SELECT COUNT(*) FROM sessions").fetchone()[0] + con.execute("SELECT COUNT(*) FROM prompts").fetchone()[0] ) if not last and any_capture: notes.append(f"no observations yet, but {any_capture} session/prompt row(s) " f"captured - those turns used no tools") elif not last: # An empty store is only a fault if work already happened and was not # captured. On a fresh install it is the correct state, and reporting # FAILED there trains people to ignore this check. old_t = [f for f in transcript_files() if time.time() - f.stat().st_mtime > 3600] if old_t: problems.append( f"store is EMPTY but {len(old_t)} transcript(s) predate it - capture never ran") else: notes.append("store is empty - fresh install, capture starts at next Stop") else: age_h = (time.time() * 1000 - last) / 3.6e6 notes.append(f"newest observation {age_h:.1f}h old") tfiles = transcript_files() if not tfiles: # Saying nothing here meant "the check passed" and "the check never ran" # printed identically. warnings.append("no transcripts on this box - the staleness check could " "not run, so capture is unproven") if tfiles: newest = max(f.stat().st_mtime for f in tfiles) t_age_h = (time.time() - newest) / 3600 notes.append(f"newest transcript {t_age_h:.1f}h old") # Real work happened but nothing was captured => capture is broken. if t_age_h < 2 and age_h > 24: problems.append( f"CAPTURE STALE: transcripts {t_age_h:.1f}h old but store {age_h:.1f}h old") # ARE THE HOOKS ACTUALLY WIRED AND RUNNABLE? # # This check exists because its absence cost 2h14m of automatic capture and # health reported OK the whole time. An installer run from a temporary copy # rewired the live hooks to that copy; the copy was then deleted, so both hooks # pointed at a path that did not exist and every session silently captured # nothing. Store freshness could not see it, because someone happened to be # running capture by hand for unrelated reasons - the store looked alive while # the automation was dead. # # Freshness of the DATA is not evidence that the MECHANISM works. for note in _hook_health(): if note.startswith("HOOK"): problems.append(note) elif note.startswith("WARN "): warnings.append(note[5:]) else: notes.append(note) ing = con.execute("SELECT COUNT(*) FROM observations").fetchone()[0] notes.append(f"{ing:,} observations") # Does the write path still COMPILE against this store? Asked directly, # because staleness could not answer it. See _capture_writepath_faults. faults = _capture_writepath_faults(con) if faults: problems.extend(faults) else: notes.append("capture write path compiles against this schema") # Can search still answer? Asked end to end, because nothing else can see this. probe = _search_probe(con) if probe and probe.startswith("UNPROVEN"): # Not a fault - but not a pass either. Printing "ok" here would be the # check reporting a result it never obtained. warnings.append(probe.removeprefix("UNPROVEN: ")) elif probe: problems.append(probe) else: notes.append("search index answers for a known row") # Did the hook record failures nobody read? streak, last_fail = _hook_failure_streak() if streak: problems.append(f"{streak} capture failure(s) since the last clean ingest " f"- hook.log last says: {last_fail[:160]}") else: notes.append("no capture failures since the last clean ingest") # A minted Signal Code sitting unread is a failure that was never surfaced. from paths import report_file rep = report_file() # NOT gated on the streak. A report minted by a crash that left no capture-failure # line was never mentioned at all - which is the one case where the report is the # only record that anything went wrong. if rep.exists(): problems.append(f"an unread failure report is waiting: {rep} " f"- read it with: recall signal (then: recall signal --clear)") for n in notes: print(f" ok {n}") for p in problems: print(f" FAIL {p}", file=sys.stderr) for w in warnings: print(f" WARN {w}", file=sys.stderr) if problems: print("\nRECALL HEALTH: FAILED", file=sys.stderr) return 1 if warnings and getattr(a, "strict", False): # --strict was documented, accepted by the parser, and never read - so anyone # gating a script on `recall health --strict` was gating it on nothing. print("\nRECALL HEALTH: FAILED (--strict: warnings count)", file=sys.stderr) return 1 if warnings: print("\nRECALL HEALTH: OK, with warnings") return 0 print("\nRECALL HEALTH: OK") return 0 def cmd_signal(a) -> int: """Show or clear the last failure report. See friday_signal.py for the promise.""" import importlib.util spec = importlib.util.spec_from_file_location( "friday_signal", Path(__file__).resolve().parent / "friday_signal.py") m = importlib.util.module_from_spec(spec) spec.loader.exec_module(m) p = m.report_dir() / m.REPORT_FILE if a.clear: if p.exists(): p.unlink(); print(f"cleared {p}") else: print("no report to clear") return 0 if not p.exists(): print("no failure recorded. Nothing has broken since the last clear.") return 0 print(p.read_text(encoding="utf-8")) print(f"[{p}]") return 0 def cmd_import(a) -> int: """Bring an existing history in, from whichever source is present. THE FLOW, in this order and no other: look -> show what was found -> import -> PROVE it landed -> only then offer to set aside what is left over Proving the import landed before offering to touch anything is the whole safety design. "Import, remove the source, then discover the import failed" is the one sequence that must be impossible, so the cleanup step is gated on a count read back out of the new store rather than on the importer's own say-so. """ import importlib.util here = Path(__file__).resolve().parent def _mod(name): spec = importlib.util.spec_from_file_location(name, here / f"{name}.py") m = importlib.util.module_from_spec(spec) spec.loader.exec_module(m) return m sources = _mod("sources") explicit = Path(a.path).expanduser() if a.path else None found = sources.detect_all(explicit) if not found: print("Nothing to import from.") print(" Claude Code writes transcripts to ~/.claude/projects - if you have used") print(" it at all, they are the source, and there were none there.") return 1 print("Found:\n") for e in found: print(sources.describe(e)) if e["id"] == "legacy-store": try: ins = _mod("inspect_store") info = ins.inspect(e["found"]["path"]) e["inspection"] = info print() for line in ins.sentence(info): print(line) except Exception: pass print() if a.source: found = [e for e in found if e["id"] == a.source] if not found: print(f"no source named {a.source!r} was found") return 1 if not a.yes: names = ", ".join(e["label"] for e in found) try: ans = input(f"Import from {names}? [Y/n] ").strip().lower() except (EOFError, KeyboardInterrupt): print("\naborted."); return 1 if ans in ("n", "no"): print("aborted - nothing imported.") return 1 _TABLES = ("sessions", "observations", "summaries", "prompts") def _counts() -> dict: """Row counts read straight out of the store. Never what an importer claims.""" c = store.connect(readonly=True) try: return {t: c.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0] for t in _TABLES} finally: c.close() before = _counts() # COUNT AROUND EACH SOURCE SEPARATELY. # # This used to take one before/after pair around the whole loop and derive a # single `verified` flag from the total. That flag then gated the offer to move # a source aside - which meant rows landing from the TRANSCRIPTS could mark the # legacy database as "verified imported" when it had in fact imported nothing. # Three real paths reached it: the legacy reader aborting on a failed redaction # selftest (it RETURNS, it does not raise, so no failure was even printed), the # reader raising and being caught below, and a partial import that committed one # batch and then died. In each case the user was invited to move aside a database # whose contents were not in the store. # # Rule 5 says a source is never offered for removal until the import FROM IT is # verified. "From it" is the whole rule, so the count has to be per source. for e in found: print(f"\n importing {e['label']} ...") b = _counts() try: e["result"] = e["read"](e["found"], a.project) except Exception as exc: print(f" import FAILED: {exc}") e["result"] = None aft = _counts() e["gained"] = sum(aft[k] - b[k] for k in _TABLES) if e["gained"] <= 0: print(f" nothing new landed from {e['label']}" " - it will not be offered for cleanup.") # PROVE IT LANDED - counted out of the store itself, not reported by the importer. after = _counts() gained = {k: after[k] - before[k] for k in after} total = sum(gained.values()) con = store.connect(readonly=True) red = con.execute("SELECT COALESCE(SUM(redactions),0) FROM observations").fetchone()[0] print("\n verified in the store:") for k, v in gained.items(): print(f" {k:14} +{v:,}") print(f" {'total':14} +{total:,}") if red: print(f" {red} secret(s) were removed on the way in.") verified = total > 0 if not verified: print("\n Nothing new landed. Either it was already imported, or the import") print(" failed. Nothing else will be offered while that is true.") return 1 # Only now, and only for sources that are ours to offer. if not a.no_cleanup: lo = _mod("leftovers") for e in found: if not e.get("removable"): continue # Per-source, never the aggregate. See the note above the import loop. lo.offer(e["found"]["path"], verified=e.get("gained", 0) > 0, assume_no=a.keep_all) print("\n Recall is now the memory. Transcripts are all it needs -") print(" no other tool has to be installed for it to keep working.") return 0 def cmd_delegate(a) -> int: """Run a sibling script as a subcommand. The README documented `recall digest` and `recall wipe`; both existed only as separate scripts, so the documentation was false for the two commands a person is most likely to reach for - one of them being how you delete your data. Making them real was the honest fix; rewording the README to match a clumsier invocation would have been the lazy one. """ import subprocess script = Path(__file__).resolve().parent / a._script return subprocess.run([sys.executable, str(script), *a.rest]).returncode def _welcome() -> int: """What a person sees when they type `recall` on its own. It used to be an argparse error on stderr, exit 2 - the first contact anyone had with this tool was a usage failure. Someone installing at 1am needs to know three things: whether it is working, what to type, and how to get their history in. """ print("Recall - your own memory of what you worked on, kept on this machine.\n") installed = False try: con = store.connect(readonly=True) obs = con.execute("SELECT COUNT(*) FROM observations").fetchone()[0] sess = con.execute("SELECT COUNT(*) FROM sessions").fetchone()[0] installed = True except Exception: obs = sess = 0 if not installed: print(" No store yet. Set it up with:\n") print(" python3 install.py\n") print(" That creates the database and wires the two hooks. Nothing else.") return 0 if obs: print(f" {obs:,} things remembered across {sess:,} session(s).\n") print(" Try:") print(" recall search \"what you half-remember\"") print(" recall timeline --days 7") print(" recall context what a new session would be told") else: print(" The store is set up, but nothing has been captured yet.\n") print(" Recall saves a session when it ENDS, so the first entry appears after") print(" you finish one. To bring in the history you already have:\n") print(" recall import reads transcripts already on this disk\n") print(" If sessions end and nothing appears, run: recall health") print("\n recall --help every command recall health is it working?") return 0 def main() -> int: ap = argparse.ArgumentParser( prog="recall", description="Your own memory of what you worked on. One file, on this machine.", epilog="Nothing here talks to the network. `recall wipe` deletes all of it.") # NOT required: bare `recall` should explain itself, not fail. sub = ap.add_subparsers(dest="cmd") s = sub.add_parser("search", help="find something you worked on before") s.add_argument("query") s.add_argument("--limit", type=int, default=25); s.add_argument("--json", action="store_true") s.add_argument("--prompts", action="store_true", help="also search what you typed") s.add_argument("--project", help="only this project") s.add_argument("--since", type=int, metavar="DAYS", help="only the last N days") s.add_argument("-v", "--verbose", action="store_true") s.add_argument("-q", "--quiet", action="store_true", help="titles only, no match evidence") s.set_defaults(fn=cmd_search) t = sub.add_parser("timeline", help="what happened, newest first") t.add_argument("--days", type=int, default=3) t.add_argument("--limit", type=int, default=80); t.set_defaults(fn=cmd_timeline) sh = sub.add_parser("show", help="expand rows by id (123, s123, p123)") sh.add_argument("ids", nargs="+") sh.set_defaults(fn=cmd_show) c = sub.add_parser("context", help="the block a new session is given") c.add_argument("--days", type=int, default=2) c.add_argument("--limit", type=int, default=40) c.add_argument("--project", default=os.environ.get("RECALL_PROJECT", "home")) c.add_argument("--max-tokens", type=int, default=900, help="hard ceiling on the injected block") c.set_defaults(fn=cmd_context) sub.add_parser("stats", help="how big the store is and what is in it" ).set_defaults(fn=cmd_stats) for name, script, helptext in ( ("digest", "digest.py", "the story of a subject, file or day"), ("brief", "brief.py", "squeeze a transcript to its spine, no model call"), ("handoff", "handoff.py", "hand this session to a fresh one instead of compacting"), ("wipe", "wipe.py", "delete everything Recall has stored"), ): d = sub.add_parser(name, help=helptext) d.add_argument("rest", nargs=argparse.REMAINDER) d.set_defaults(fn=cmd_delegate, _script=script) im = sub.add_parser("import", help="bring an existing history in") im.add_argument("--path", help="a database, if it is not in the usual place") im.add_argument("--source", help="only this source id (see the table in sources.py)") im.add_argument("--project", default="imported") im.add_argument("--yes", action="store_true") im.add_argument("--keep-all", action="store_true", help="import, but keep every leftover without asking") im.add_argument("--no-cleanup", action="store_true", help="never mention leftovers at all") im.set_defaults(fn=cmd_import) sg = sub.add_parser("signal", help="the last failure report (nothing is ever sent)") sg.add_argument("--clear", action="store_true"); sg.set_defaults(fn=cmd_signal) h = sub.add_parser("health", help="is capture actually working?") h.add_argument("--strict", action="store_true", help="also fail on warnings, for use in a script") h.set_defaults(fn=cmd_health) a = ap.parse_args() if not getattr(a, "fn", None): return _welcome() return a.fn(a) try: # Signal Code: a crash mints a from signal_hook import arm # short code and a local report instead arm() # of a wall of traceback. Nothing is sent. except Exception: pass if __name__ == "__main__": sys.exit(main())