interactive-story/backend/tools/v11_compat_check.py

"""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 <v1 campaign.db> \\
        --tree <v1.0.0 worktree>/backend --label v100 --out "$HOME/v11-evidence/compat"
    .venv/bin/python -m tools.v11_compat_check --db <v1 campaign.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())