"""v1.1: does a real v1.0.0 database open unchanged? # the same database, opened by each tree, snapshotted read-only .venv/bin/python -m tools.v11_compat_check --db \\ --tree /backend --label v100 --out "$HOME/v11-evidence/compat" .venv/bin/python -m tools.v11_compat_check --db \\ --label v11 --exercise --out "$HOME/v11-evidence/compat" Run from `backend/`. The source database is never opened. It is copied into `--out` first, and the copy is what the application opens. A read-only snapshot is taken through the API, the same way a reader sees the campaign: - the export bundle, which carries the whole tree, the head, the Save Points, state, events, summaries, memories and knowledge, and has no timestamp of its own; - the narrative state and its events; - the Save Points, the imported knowledge, the memories, the derived status and the settings; - the database schema and `PRAGMA user_version`, before and after the application opened it. Two snapshots of the same database from two trees are then compared. Identical means v1.1 read it exactly as v1.0.0 did, and a matching schema and version mean nothing migrated. `--exercise` then uses the v1.1 copy: undo, redo, a Save Point restore, a context dry run (knowledge retrieval), an export, and an import of that export. It first points the copy's endpoint at a loopback port that refuses, so nothing here reaches an inference server. """ from __future__ import annotations import argparse import json import os import shutil import sqlite3 import sys from pathlib import Path def _schema(path: Path) -> dict: connection = sqlite3.connect(f"file:{path}?mode=ro", uri=True) try: version = connection.execute("PRAGMA user_version").fetchone()[0] rows = connection.execute( "SELECT type, name, sql FROM sqlite_master WHERE name NOT LIKE 'sqlite_%' " "ORDER BY type, name").fetchall() finally: connection.close() return {"user_version": version, "objects": [list(r) for r in rows]} def main() -> int: parser = argparse.ArgumentParser(description=__doc__.split("\n")[0]) parser.add_argument("--db", required=True) parser.add_argument("--tree", default="", help="a backend/ directory to import the app from") parser.add_argument("--label", required=True) parser.add_argument("--exercise", action="store_true") parser.add_argument("--out", required=True) args = parser.parse_args() out = Path(args.out) out.mkdir(parents=True, exist_ok=True) copy = out / f"{args.label}.db" if copy.exists(): print(f"{copy} exists; choose a new --label or --out") return 2 shutil.copy2(args.db, copy) schema_before = _schema(copy) if args.tree: sys.path.insert(0, str(Path(args.tree).resolve())) os.environ["AIDND_DB_PATH"] = str(copy) os.environ.pop("AIDND_DATABASE_URL", None) os.environ.pop("DATABASE_URL", None) from fastapi import Depends from fastapi.testclient import TestClient from app import auth, limits, models from app.database import SessionLocal, get_db from app.main import app print(f"app imported from {Path(sys.modules['app'].__file__).parent}") limits.check_row_cap = lambda *a, **k: None with SessionLocal() as db: owner = db.query(models.Adventure.user_id).order_by(models.Adventure.id).first() user_id = owner[0] if owner else db.query(models.User.id).first()[0] app.dependency_overrides[auth.get_current_user] = ( lambda db=Depends(get_db): db.get(models.User, user_id) ) def call(client, method, url, body=None, expect=200): response = client.request(method, f"/api{url}", json=body) if response.status_code != expect: raise SystemExit(f"{method} {url}: HTTP {response.status_code} {response.text[:300]}") return response.json() if response.content else None report: dict = {"label": args.label, "schema_before": schema_before} with TestClient(app) as client: with SessionLocal() as db: adventure_ids = [a for (a,) in db.query(models.Adventure.id) .filter(models.Adventure.user_id == user_id) .order_by(models.Adventure.id)] snapshot = {"settings": call(client, "GET", "/settings"), "adventures": {}} for adv in adventure_ids: snapshot["adventures"][str(adv)] = { "export": call(client, "GET", f"/adventures/{adv}/export"), "state": call(client, "GET", f"/adventures/{adv}/state"), "state_events": call(client, "GET", f"/adventures/{adv}/state/events"), "checkpoints": call(client, "GET", f"/adventures/{adv}/checkpoints"), "knowledge": call(client, "GET", f"/adventures/{adv}/knowledge"), "memories": call(client, "GET", f"/adventures/{adv}/memories"), "derived": call(client, "GET", f"/adventures/{adv}/derived"), "newest_actions": call(client, "GET", f"/adventures/{adv}/actions?limit=5"), } report["snapshot"] = snapshot if args.exercise and adventure_ids: adv = adventure_ids[0] ex: dict = {} call(client, "PUT", "/settings", {"endpoint_url": "http://127.0.0.1:9/v1", "embedding_model": ""}) before = call(client, "GET", f"/adventures/{adv}/actions?limit=1") ex["before"] = {k: before[k] for k in ("total", "can_undo", "can_redo")} undone = call(client, "POST", f"/adventures/{adv}/undo") ex["after_undo"] = {k: undone[k] for k in ("total", "can_undo", "can_redo")} redone = call(client, "POST", f"/adventures/{adv}/redo") ex["after_redo"] = {k: redone[k] for k in ("total", "can_undo", "can_redo")} points = call(client, "GET", f"/adventures/{adv}/checkpoints") if points: point = points[0] restored = call(client, "POST", f"/adventures/{adv}/checkpoints/{point['id']}/restore") page = call(client, "GET", f"/adventures/{adv}/actions?limit=1") ex["restore"] = {"save_point": point["name"], "total": page["total"], "can_redo": page["can_redo"], "response_keys": sorted(restored or {})} context = client.get(f"/api/adventures/{adv}/context") body = context.json() ex["context"] = { "status": context.status_code, "knowledge_used": len(((body.get("knowledge") or {}).get("used")) or []), "canon_section": any(s["label"] == "campaign_canon" for s in body.get("sections") or []), "state_section": any(s["label"] == "narrative_state" for s in body.get("sections") or []), "summary": body.get("summary"), "tokens": body.get("tokens"), "window": body.get("window"), } bundle = call(client, "GET", f"/adventures/{adv}/export") imported = call(client, "POST", "/adventures/import", bundle, expect=201) new_id = imported["id"] reimport = call(client, "GET", f"/adventures/{new_id}/export") ex["import"] = { "new_id": new_id, "actions_in_bundle": len(bundle.get("actions") or []), "actions_after_import": len(reimport.get("actions") or []), "head_same": (bundle.get("headBranch") is not None and bundle.get("headDepth") == reimport.get("headDepth")), "checkpoints": [len(bundle.get("checkpoints") or []), len(reimport.get("checkpoints") or [])], "memories": [len(bundle.get("memories") or []), len(reimport.get("memories") or [])], "narrative_state_same": bundle.get("narrativeState") == reimport.get("narrativeState"), } report["exercise"] = ex app.dependency_overrides.clear() report["schema_after"] = _schema(copy) (out / f"{args.label}.json").write_text(json.dumps(report, indent=2, sort_keys=True, default=str)) same_schema = report["schema_before"] == report["schema_after"] print(f"schema unchanged by opening: {same_schema} " f"(user_version {report['schema_before']['user_version']} -> " f"{report['schema_after']['user_version']})") if "exercise" in report: print(json.dumps(report["exercise"], indent=2, default=str)[:3000]) return 0 if __name__ == "__main__": sys.exit(main())