bin/digest.py

#!/usr/bin/env python3 """ digest.py - answer "what happened with X" instead of "which rows mention X". THE GAP THIS FILLS ------------------ The predecessor shipped a "corpora / knowledge base" feature: build a corpus, prime it, query it. In practice the useful shape of that is narrower and cheaper - you have a subject (a host, a file, a subsystem, an incident) and you want its STORY: when it started, what changed, what broke, where it stands. A search gives you 25 rows and leaves the assembling to you. This assembles it. No model is involved: it is grouping and ordering over rows we already have, which means it cannot hallucinate a history that did not happen - the failure mode that makes a "knowledge base" worse than no knowledge base. recall-digest "<subject>" the story of a subject recall-digest --file <path> everything that touched a file recall-digest --day 2026-08-19 one day, grouped """ from __future__ import annotations import argparse import json import sys from collections import Counter, defaultdict from datetime import datetime, timezone from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import store # noqa: E402 from recall import _fts_escape, TYPE_ICON, _local # noqa: E402 def _rows_for_subject(con, subject: str, limit: int): q = _fts_escape(subject) return con.execute(""" SELECT o.id,o.type,o.title,o.subtitle,o.body,o.created_at,o.files_modified, o.session_uid, bm25(observations_fts) AS rank FROM observations_fts JOIN observations o ON o.id=observations_fts.rowid WHERE observations_fts MATCH ? ORDER BY o.created_at_epoch LIMIT ?""", (q, limit)).fetchall() def _rows_for_file(con, path: str, limit: int): like = f"%{path}%" return con.execute(""" SELECT id,type,title,subtitle,body,created_at,files_modified,session_uid FROM observations WHERE files_modified LIKE ? OR files_read LIKE ? ORDER BY created_at_epoch LIMIT ?""", (like, like, limit)).fetchall() def _rows_for_day(con, day: str, limit: int): return con.execute(""" SELECT id,type,title,subtitle,body,created_at,files_modified,session_uid FROM observations WHERE created_at LIKE ? ORDER BY created_at_epoch LIMIT ?""", (day + "%", limit)).fetchall() def main() -> int: ap = argparse.ArgumentParser(prog="recall-digest") ap.add_argument("subject", nargs="?") ap.add_argument("--file") ap.add_argument("--day") ap.add_argument("--limit", type=int, default=400) a = ap.parse_args() con = store.connect(readonly=True) if a.file: rows, label = _rows_for_file(con, a.file, a.limit), f"file {a.file}" elif a.day: rows, label = _rows_for_day(con, a.day, a.limit), f"day {a.day}" elif a.subject: rows, label = _rows_for_subject(con, a.subject, a.limit), f"subject {a.subject!r}" else: ap.error("give a subject, --file, or --day") if not rows: print(f"nothing on record for {label}") return 0 first, last = _local(rows[0]["created_at"]), _local(rows[-1]["created_at"]) kinds = Counter(r["type"] for r in rows) files = Counter() for r in rows: try: for f in json.loads(r["files_modified"] or "[]"): files[f] += 1 except Exception: pass sessions = len({r["session_uid"] for r in rows if r["session_uid"]}) print(f"=== {label}") print(f" {len(rows)} entries across {sessions} session(s)") print(f" {first.strftime('%Y-%m-%d %H:%M')} -> {last.strftime('%Y-%m-%d %H:%M')}") print(f" " + " ".join(f"{TYPE_ICON.get(k,'-')}{k}={v}" for k, v in kinds.most_common())) if files: print(f"\n files most touched:") for f, n in files.most_common(8): print(f" {n:>3}x {f}") # The story, thinned. Everything is too much; the turning points are the value. notable = [r for r in rows if r["type"] in ("bugfix", "security_alert", "security_note", "decision", "feature")] if len(notable) < 6: notable = rows print(f"\n timeline ({len(notable)} shown):") day = None for r in notable[:60]: lt = _local(r["created_at"]) d = lt.strftime("%Y-%m-%d") if d != day: day = d print(f"\n -- {d}") ic = TYPE_ICON.get(r["type"], "-") print(f" {r['id']:>7} {lt.strftime('%H:%M')} {ic} {(r['title'] or '')[:88]}") print(f"\n expand any of these: recall show <id>") return 0 if __name__ == "__main__": sys.exit(main())