interactive-story/backend/tests/test_knowledge_migration.py

"""M7: opening a genuine pre-M7 database, and playing on afterwards.

Two databases are exercised, because they fail differently:

* **Fresh.** Everything is built by `create_all`, which is the path a new
  install takes — and the path the FTS5 index nearly missed, because a virtual
  table is not something SQLAlchemy's metadata describes.
* **A real M6 database.** Built by dropping every M7 table and index and
  rewinding the stamp to 91, so the M7 migration runs its real statements
  against a schema that genuinely lacks them. A current schema with an old stamp
  would skip the DDL and test half the change (the lesson
  `tests/schema_rewind.py` was written for).

What the second one has to prove is not "the migration completed". It is that a
campaign written before M7 existed still behaves: its history, head, branches,
Save Points, narrative state, summaries, memories, derived status and prompt
provenance are all intact, it needs no knowledge sources to play, and it can
then import one and use it.

    python -m pytest tests/test_knowledge_migration.py -v
"""

import pytest
from fastapi import Depends
from fastapi.testclient import TestClient
from sqlalchemy import inspect, select, text

from app import auth, limits, memorybank, migrations, models
from app.database import Base, SessionLocal, engine, get_db
from app.knowledge import fts
from app.main import app
from app.routers import adventures

from fakes import ScriptedProvider, state_block

M6_VERSION = 91
M7_VERSION = 92

#: Everything M7 adds to the schema. Dropping all of it and rewinding the stamp
#: is what makes the fixture a real M6 database rather than a current one
#: wearing an old number.
M7_TABLES = ("knowledge_embeddings", "knowledge_chunks", "knowledge_sources")


class StubEmbedder:
    async def embed(self, texts):
        return [[1.0, float(len(t) % 7), 0.5] for t in texts]


@pytest.fixture()
def client(monkeypatch):
    Base.metadata.create_all(bind=engine)
    memorybank._vector_cache.clear()
    monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
    monkeypatch.setattr(adventures.turns, "OpenAICompatibleProvider", ScriptedProvider)
    monkeypatch.setattr(memorybank, "embedding_provider", lambda s: StubEmbedder())
    monkeypatch.setattr(memorybank, "summary_provider", lambda s: StubEmbedder())
    try:
        yield _make_client()
    finally:
        app.dependency_overrides.clear()
        memorybank._vector_cache.clear()
        Base.metadata.drop_all(bind=engine)


def _make_client():
    setup = SessionLocal()
    user = models.User(is_guest=False, email="m7mig@example.com")
    setup.add(user)
    setup.flush()
    setup.add(models.Settings(
        user_id=user.id, model="test-model", embedding_model="",
        context_token_budget=4000, max_output_tokens=300,
    ))
    adventure = models.Adventure(user_id=user.id, title="Pre-M7 Campaign")
    setup.add(adventure)
    setup.flush()
    setup.add(models.Action(adventure_id=adventure.id, type="start",
                            text="The road forks at the Crooked Lantern."))
    setup.commit()
    adv_id, user_id = adventure.id, user.id
    setup.close()
    app.dependency_overrides[auth.get_current_user] = (
        lambda db=Depends(get_db): db.get(models.User, user_id)
    )
    test_client = TestClient(app)
    test_client.adv_id = adv_id
    test_client.user_id = user_id
    return test_client


def play(client, text_, prose="The road bends on past the treeline.", events=None):
    ScriptedProvider.replies = [f"{prose}\n{state_block(events or [])}"]
    response = client.post(f"/api/adventures/{client.adv_id}/actions",
                           json={"type": "do", "text": text_})
    assert response.status_code == 200, response.text[:300]
    return response


def rewind_to_m6():
    """Makes the database genuinely M6: no M7 tables, no M7 index, stamp 91."""
    with engine.begin() as conn:
        for table in M7_TABLES:
            conn.execute(text(f"DROP TABLE IF EXISTS {table}"))
        conn.execute(text(f"DROP TABLE IF EXISTS {fts.TABLE}"))
        conn.execute(text(f"PRAGMA user_version = {M6_VERSION}"))


def stamp():
    with engine.begin() as conn:
        return conn.execute(text("PRAGMA user_version")).scalar()


def upload(client, name, body, classification):
    return client.post(
        f"/api/adventures/{client.adv_id}/knowledge",
        files={"file": (name, body.encode(), "text/markdown")},
        data={"classification": classification},
    )


# ----------------------------------------------------------------- fresh

def test_a_fresh_database_gets_every_m7_table_and_the_fts_index(client):
    """The `create_all` path, including the virtual table it cannot describe."""
    tables = set(inspect(engine).get_table_names())
    for table in M7_TABLES:
        assert table in tables
    assert fts.TABLE in tables
    assert stamp() == migrations.LATEST_VERSION == M7_VERSION

    # And it works end to end on that fresh database.
    assert upload(client, "canon.md",
                  "# Abbey\n\nThe Old Abbey lies north of Westhaven.\n",
                  "canon").status_code == 201
    play(client, "Aldric asks about the Old Abbey north of Westhaven.")
    report = client.get(f"/api/adventures/{client.adv_id}/context").json()
    assert "canon.md" in [u["filename"] for u in report["knowledge"]["used"]]


# ------------------------------------------------------------ a real M6 db

def test_a_real_m6_database_migrates_and_keeps_everything_it_had(client):
    """The migration, against a database that genuinely predates M7."""
    # --- build a campaign with one of everything M6 owns ---
    play(client, "Aldric leaves the tavern.")
    play(client, "Aldric walks the north road.",
         events=[{"type": "create_entity", "entity": "aldric", "name": "Aldric",
                  "entity_type": "character"}])
    play(client, "Aldric reaches the abbey gate.",
         events=[{"type": "add_fact", "fact_id": "at-gate", "subject": "aldric",
                  "predicate": "stands at", "value": "the abbey gate"}])
    save_point = client.post(f"/api/adventures/{client.adv_id}/checkpoints",
                             json={"name": "At the gate"}).json()
    assert client.post(f"/api/adventures/{client.adv_id}/undo").status_code == 200
    play(client, "Aldric turns back instead.", prose="He turns back toward the town.")

    with SessionLocal() as db:
        adventure = db.get(models.Adventure, client.adv_id)
        db.add(models.Summary(
            adventure_id=adventure.id, text="Aldric has been walking north.",
            branch_id=adventure.head_branch_id, depth=adventure.head_depth,
            source_start=0, source_end=adventure.head_depth, trigger="interval",
        ))
        memory = models.Memory(
            adventure_id=adventure.id, text="Aldric left the Crooked Lantern.",
            branch_id=adventure.head_branch_id, depth=adventure.head_depth,
        )
        memorybank.set_vector(memory, [1.0, 2.0, 3.0])
        db.add(memory)
        db.add(models.DerivedStatus(
            adventure_id=adventure.id, kind="summary", status="ok"))
        db.commit()

    before = {
        "actions": client.get(f"/api/adventures/{client.adv_id}/actions").json(),
        "branches": client.get(f"/api/adventures/{client.adv_id}/branches").json(),
        "checkpoints": client.get(f"/api/adventures/{client.adv_id}/checkpoints").json(),
        "state": client.get(f"/api/adventures/{client.adv_id}/state").json(),
        "derived": client.get(f"/api/adventures/{client.adv_id}/derived").json(),
        "memories": client.get(f"/api/adventures/{client.adv_id}/memories").json(),
    }
    with SessionLocal() as db:
        adventure = db.get(models.Adventure, client.adv_id)
        head_before = (adventure.head_branch_id, adventure.head_depth)
        state_before = adventure.narrative_state
    ai_action = next(a for a in reversed(before["actions"]["actions"])
                     if a["type"] == "ai")
    snapshot_before = client.get(
        f"/api/adventures/{client.adv_id}/actions/{ai_action['id']}/context"
    ).json()

    # --- make it an M6 database, then migrate it ---
    rewind_to_m6()
    tables = set(inspect(engine).get_table_names())
    assert not (set(M7_TABLES) & tables)
    assert fts.TABLE not in tables
    assert stamp() == M6_VERSION

    migrations.bootstrap(engine)

    assert stamp() == M7_VERSION
    tables = set(inspect(engine).get_table_names())
    for table in M7_TABLES + (fts.TABLE,):
        assert table in tables, table

    # --- everything M6 had still behaves ---
    assert client.get(f"/api/adventures/{client.adv_id}/actions").json() \
        == before["actions"]
    assert client.get(f"/api/adventures/{client.adv_id}/branches").json() \
        == before["branches"]
    assert client.get(f"/api/adventures/{client.adv_id}/checkpoints").json() \
        == before["checkpoints"]
    assert client.get(f"/api/adventures/{client.adv_id}/state").json() \
        == before["state"]
    assert client.get(f"/api/adventures/{client.adv_id}/memories").json() \
        == before["memories"]
    derived_after = client.get(f"/api/adventures/{client.adv_id}/derived").json()
    assert derived_after["summaries"] == before["derived"]["summaries"]
    assert derived_after["status"] == before["derived"]["status"]
    with SessionLocal() as db:
        adventure = db.get(models.Adventure, client.adv_id)
        assert (adventure.head_branch_id, adventure.head_depth) == head_before
        assert adventure.narrative_state == state_before

    # Prompt provenance from before the migration is still readable, and its
    # M6 components are unchanged.
    snapshot_after = client.get(
        f"/api/adventures/{client.adv_id}/actions/{ai_action['id']}/context"
    ).json()
    assert snapshot_after["sections"] == snapshot_before["sections"]
    assert snapshot_after["summary"] == snapshot_before["summary"]
    assert snapshot_after["memories"] == snapshot_before["memories"]

    # The campaign needs no knowledge sources to keep playing.
    assert client.get(f"/api/adventures/{client.adv_id}/knowledge").json() == []
    report = client.get(f"/api/adventures/{client.adv_id}/context").json()
    assert report["knowledge"]["used"] == []
    assert not any(s["label"].startswith("imported_") for s in report["sections"])
    play(client, "Aldric keeps walking.")

    # Undo, Redo and Save Point restore all still work after the migration.
    assert client.post(f"/api/adventures/{client.adv_id}/undo").status_code == 200
    assert client.post(f"/api/adventures/{client.adv_id}/redo").status_code == 200
    assert client.post(
        f"/api/adventures/{client.adv_id}/checkpoints/{save_point['id']}/restore"
    ).status_code == 200

    # --- and it can now use the new subsystem ---
    assert upload(client, "canon.md",
                  "# The Abbey\n\nThe Old Abbey lies five miles north of "
                  "Westhaven and its crypt bears a broken circle.\n",
                  "canon").status_code == 201
    play(client, "Aldric asks about the Old Abbey and its broken-circle symbol.")
    report = client.get(f"/api/adventures/{client.adv_id}/context").json()
    assert "canon.md" in [u["filename"] for u in report["knowledge"]["used"]]


def test_the_migration_is_idempotent(client):
    """Running it twice is not a second migration."""
    rewind_to_m6()
    migrations.bootstrap(engine)
    upload(client, "canon.md", "# Abbey\n\nThe abbey stands.\n", "canon")
    with SessionLocal() as db:
        rows = len(db.execute(select(models.KnowledgeChunk)).scalars().all())

    migrations.bootstrap(engine)
    assert stamp() == M7_VERSION
    with SessionLocal() as db:
        assert len(db.execute(select(models.KnowledgeChunk)).scalars().all()) == rows
    assert len(client.get(f"/api/adventures/{client.adv_id}/knowledge").json()) == 1


def test_the_fts_index_is_dropped_with_the_table_it_indexes():
    """`create_all`/`drop_all` carry the virtual table both ways.

    Without this, a teardown would leave the index holding rowids for chunks
    that no longer exist, and the next campaign's first passage would inherit a
    stranger's search results.
    """
    Base.metadata.create_all(bind=engine)
    assert fts.TABLE in inspect(engine).get_table_names()
    Base.metadata.drop_all(bind=engine)
    assert fts.TABLE not in inspect(engine).get_table_names()
    Base.metadata.create_all(bind=engine)
    with engine.begin() as conn:
        assert conn.execute(text(f"SELECT count(*) FROM {fts.TABLE}")).scalar() == 0
    Base.metadata.drop_all(bind=engine)