bin/archive.py

#!/usr/bin/env python3 """ archive.py - durable, redacted export of the Recall store. WHY --- Harvested history may originally exist only inside the previous tool's database. Losing that would leave recall.db as a single point of failure with no way to rebuild. This writes our own copy, in our own format, already redacted: gzipped JSONL, one object per line, no vendor schema and nothing to interpret. recall-archive export write archive/recall-<date>.jsonl.gz recall-archive verify <f> prove the archive can be read back and counted """ from __future__ import annotations import argparse import gzip import json import sys from datetime import datetime, timezone from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import store # noqa: E402 from paths import archive_dir ARCHIVE_DIR = archive_dir() TABLES = ("sessions", "observations", "summaries", "prompts") def cmd_export(a) -> int: con = store.connect(readonly=True) ARCHIVE_DIR.mkdir(parents=True, exist_ok=True) stamp = datetime.now(timezone.utc).strftime("%Y%m%d") out = Path(a.out) if a.out else ARCHIVE_DIR / f"recall-{stamp}.jsonl.gz" counts = {} with gzip.open(out, "wt", encoding="utf-8") as f: f.write(json.dumps({ "_meta": "friday-recall-archive", "version": 1, "exported_at": datetime.now(timezone.utc).isoformat(), "tables": list(TABLES), }) + "\n") for t in TABLES: n = 0 for row in con.execute(f"SELECT * FROM {t}"): d = dict(row) d["_t"] = t f.write(json.dumps(d, ensure_ascii=False) + "\n") n += 1 counts[t] = n size = out.stat().st_size print(f"archive {out}") print(f"size {size/1e6:.1f} MB") for k, v in counts.items(): print(f" {k:14}{v:>9,}") return 0 def _pick_archive(name: str | None) -> Path: """Resolve an archive path, or say plainly that there is none. A missing backup is a real state to report, not a stack trace to decode.""" if name: p = Path(name) if not p.is_file(): raise SystemExit(f"no archive at {p}") return p found = sorted(ARCHIVE_DIR.glob("recall-*.jsonl.gz")) if not found: raise SystemExit(f"no archives in {ARCHIVE_DIR} - run: recall-archive export") return found[-1] def cmd_verify(a) -> int: """An archive nobody has read back is a guess, not a backup.""" p = _pick_archive(a.file) counts: dict[str, int] = {} meta = None bad = 0 with gzip.open(p, "rt", encoding="utf-8") as f: for i, line in enumerate(f): try: d = json.loads(line) except Exception: bad += 1 continue if i == 0 and d.get("_meta"): meta = d continue t = d.get("_t") if not t: bad += 1 continue counts[t] = counts.get(t, 0) + 1 con = store.connect(readonly=True) live = {t: con.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0] for t in TABLES} print(f"archive {p}") print(f"meta {meta.get('exported_at') if meta else 'MISSING'}") print(f"{'table':16}{'archive':>10}{'store':>10} {'delta':>8}") ok = bad == 0 and meta is not None drift = 0 for t in TABLES: a_n, l_n = counts.get(t, 0), live[t] d = l_n - a_n drift += max(d, 0) print(f"{t:16}{a_n:>10,}{l_n:>10,} {d:>+8,}") if bad: print(f"unparseable lines: {bad}") if meta is None: print("archive header MISSING - not a recall archive") # Two different questions. Corruption is always a failure. Drift only matters # when the archive is about to become the ONLY copy - that is --strict, and # it is what the pre-burn gate uses. if drift: print(f"\nDRIFT: store has {drift:,} rows the archive does not " f"(live capture ran after export)") if a.strict and drift: print("ARCHIVE VERIFY: FAILED (--strict: archive does not cover the store)") return 1 print("\nARCHIVE VERIFY:", "OK" if ok else "FAILED") return 0 if ok else 1 def cmd_restore(a) -> int: """Rebuild the store from an archive alone - the path that makes the burn safe.""" p = _pick_archive(a.file) con = store.connect() cols = {t: [r[1] for r in con.execute(f"PRAGMA table_info({t})")] for t in TABLES} n = 0 with gzip.open(p, "rt", encoding="utf-8") as f: for line in f: try: d = json.loads(line) except Exception: continue t = d.pop("_t", None) if t not in TABLES: continue d.pop("id", None) if t != "sessions" else None keys = [k for k in d if k in cols[t]] if not keys: continue q = (f"INSERT OR IGNORE INTO {t} ({','.join(keys)}) " f"VALUES ({','.join('?' * len(keys))})") con.execute(q, [d[k] for k in keys]) n += 1 if n % 5000 == 0: con.commit() con.commit() print(f"restored {n:,} rows from {p.name}") return 0 def main() -> int: ap = argparse.ArgumentParser(prog="recall-archive") s = ap.add_subparsers(dest="cmd", required=True) e = s.add_parser("export"); e.add_argument("--out"); e.set_defaults(fn=cmd_export) v = s.add_parser("verify"); v.add_argument("file", nargs="?") v.add_argument("--strict", action="store_true", help="fail if the store holds rows the archive does not") v.set_defaults(fn=cmd_verify) r = s.add_parser("restore"); r.add_argument("file", nargs="?"); r.set_defaults(fn=cmd_restore) a = ap.parse_args() return a.fn(a) if __name__ == "__main__": sys.exit(main())