bin/legacy_store.py

#!/usr/bin/env python3 """ legacy_store.py - read a previous memory tool's database into Recall. Named for what it DOES, not for whose it is. This is one row in sources.py; adding another tool means another row, not another file with somebody's name on it. Identified by its TABLES, never by its filename - a folder name proves nothing, and guessing from one is how a tool ends up touching a file it never understood. Opened READ-ONLY. Importing reads; whether that store then goes is the owner's decision and is handled separately, after the import has been verified. Everything crosses the redactor on the way in. That matters more here than anywhere else in the package: stores like this have been found holding credentials in clear text, and importing without scrubbing would carry them straight into the new store. """ from __future__ import annotations import argparse import os import sqlite3 import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import store # noqa: E402 from redact import scrub, selftest # noqa: E402 # No default path here on purpose. sources.py owns WHERE a legacy store lives; # a second copy of that path would be a second place to keep in step, and it is # the only reason a vendor name would appear in this file at all. SRC = os.environ.get("RECALL_IMPORT_DB", "") def _clean(row: sqlite3.Row, fields: list[str]) -> tuple[dict, int]: out, hits = {}, 0 for f in fields: v = row[f] if f in row.keys() else None if isinstance(v, str) and v: v, n, _ = scrub(v) hits += n out[f] = v return out, hits def _run(args) -> dict: # A migration that runs with a broken redactor writes cleartext secrets # into the new store. Refuse rather than "probably fine". if selftest(verbose=False) != 0: print("ABORT: redaction selftest failed - refusing to migrate", file=sys.stderr) selftest(verbose=True) # RAISE, do not return a status code. This function is declared -> dict and # every caller treats it as one. Returning 2 here meant the refusal travelled # as a truthy value that looked like a result: `recall import` printed no # failure at all, and `main()` below still exited 0. A refusal that reports # success is the same defect as a check that cannot fail. raise RuntimeError("redaction selftest failed - refusing to migrate") src = sqlite3.connect(f"file:{args.src}?mode=ro&immutable=1", uri=True) src.row_factory = sqlite3.Row dst = store.connect() # WHICH TABLES THIS PARTICULAR DATABASE ACTUALLY HAS. # # Detection only requires the two core tables; session_summaries and # user_prompts are optional and older stores do not have them. This code read # them unconditionally anyway, so such a store passed detection, imported its # observations in 5000-row batches, and THEN raised on a missing table - leaving # a half-imported store behind. Ask the database what it has instead of assuming. present = {r[0] for r in src.execute( "SELECT name FROM sqlite_master WHERE type='table'")} stats: dict[str, int] = {} redactions = 0 # ---- sessions ----------------------------------------------------------- n = 0 for r in src.execute("SELECT * FROM sdk_sessions"): uid = r["memory_session_id"] or f"legacy-{r['id']}" c, h = _clean(r, ["user_prompt", "custom_title"]) redactions += h dst.execute("""INSERT OR IGNORE 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,legacy_id) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)""", (uid, r["content_session_id"], r["project"], r["platform_source"], c["custom_title"], c["user_prompt"], r["started_at"], r["started_at_epoch"], r["completed_at"], r["completed_at_epoch"], r["status"], r["prompt_counter"], r["id"])) n += 1 stats["sessions"] = n # ---- observations ------------------------------------------------------- TXT = ["text", "title", "subtitle", "facts", "narrative", "concepts"] n = 0 for r in src.execute("SELECT * FROM observations"): c, h = _clean(r, TXT) redactions += h dst.execute("""INSERT OR IGNORE INTO observations (session_uid,project,type,title,subtitle,body,facts,narrative,concepts, files_read,files_modified,prompt_number,model,agent_type, created_at,created_at_epoch,redactions,source,legacy_id) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,'legacy-import',?)""", (r["memory_session_id"], r["project"], r["type"], c["title"], c["subtitle"], c["text"], c["facts"], c["narrative"], c["concepts"], r["files_read"], r["files_modified"], r["prompt_number"], r["generated_by_model"], r["agent_type"], r["created_at"], r["created_at_epoch"], h, r["id"])) n += 1 if n % 5000 == 0: dst.commit() print(f" observations {n}...", flush=True) stats["observations"] = n # ---- summaries (optional table) ----------------------------------------- TXT = ["request", "investigated", "learned", "completed", "next_steps", "notes"] n = 0 for r in (src.execute("SELECT * FROM session_summaries") if "session_summaries" in present else ()): c, h = _clean(r, TXT) redactions += h dst.execute("""INSERT OR IGNORE INTO summaries (session_uid,project,request,investigated,learned,completed,next_steps, files_read,files_edited,notes,prompt_number,created_at,created_at_epoch, redactions,legacy_id) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (r["memory_session_id"], r["project"], c["request"], c["investigated"], c["learned"], c["completed"], c["next_steps"], r["files_read"], r["files_edited"], c["notes"], r["prompt_number"], r["created_at"], r["created_at_epoch"], h, r["id"])) n += 1 stats["summaries"] = n # ---- prompts (optional table) ------------------------------------------- n = 0 for r in (src.execute("""SELECT p.*, s.memory_session_id, s.project FROM user_prompts p LEFT JOIN sdk_sessions s ON p.session_db_id = s.id""") if "user_prompts" in present else ()): c, h = _clean(r, ["prompt_text"]) redactions += h dst.execute("""INSERT OR IGNORE INTO prompts (session_uid,project,prompt_number,body,created_at,created_at_epoch, redactions,source,legacy_id) VALUES (?,?,?,?,?,?,?,'legacy-import',?)""", (r["memory_session_id"], r["project"], r["prompt_number"], c["prompt_text"], r["created_at"], r["created_at_epoch"], h, r["id"])) n += 1 stats["prompts"] = n if args.dry_run: dst.rollback() print("DRY RUN - rolled back") else: dst.execute("INSERT OR REPLACE INTO meta(k,v) VALUES('migrated_from',?)", (args.src,)) dst.commit() dst.execute("INSERT INTO observations_fts(observations_fts) VALUES('optimize')") dst.execute("INSERT INTO summaries_fts(summaries_fts) VALUES('optimize')") dst.execute("INSERT INTO prompts_fts(prompts_fts) VALUES('optimize')") dst.commit() print("\n--- migrated ---") for k, v in stats.items(): print(f" {k:14} {v}") print(f" {'redactions':14} {redactions}") return {**stats, "redactions": redactions} def import_from(db) -> dict: """Entry point used by sources.py. Returns what was stored. Defined ABOVE the __main__ guard on purpose: below it, `sys.exit(main())` means this name is never bound when the file runs as a script, and a reader checking whether the entry point exists has to know that to get the right answer. """ import argparse as _a return _run(_a.Namespace(src=str(db), dry_run=False)) def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--src", default=SRC) ap.add_argument("--dry-run", action="store_true") a = ap.parse_args() try: _run(a) except Exception as exc: # Previously this returned 0 whatever happened, including the refusal above. print(f"migration failed: {exc}", file=sys.stderr) return 2 return 0 if __name__ == "__main__": sys.exit(main())