interactive-story/backend/tests/test_pre_m5_compatibility.py

"""M5 corrective pass: campaigns that existed before narrative state.

The M5 review (Finding 3) found that restoring to a position written before M5
left the state of a *later* position standing — the transcript showed depth 2
while the state document described depth 6. The head/state invariant this
project holds everywhere is:

    visible active transcript position == stored head == authoritative state

A position with no narrative snapshot cannot be exempt from it. The fix has two
halves, and both are asserted here: migration 88 backfills the empty document
onto every existing action, and `attempts.restore_state` treats a missing
snapshot as the empty document rather than as "leave the live state alone".

The fixture is a genuine pre-M5 database. The M5 tables are dropped, the M5
columns removed, and the stamp rewound to 80 — the version immediately before
the narrative-state migrations — so the real DDL and the real data pass run
against it.

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

import pytest
from fastapi import Depends
from fastapi.testclient import TestClient
from sqlalchemy import text

from app import auth, limits, migrations, models
from app.database import Base, SessionLocal, engine, get_db
from app.main import app
from app.narrative import model as narrative_model
from app.routers import adventures

from fakes import ScriptedProvider, state_block

# The stamp immediately before the M5 narrative-state migrations (81-88).
PRE_M5_VERSION = 80


@pytest.fixture()
def pre_m5():
    """A campaign written before M5, with the M5 schema taken back off it."""
    Base.metadata.create_all(bind=engine)
    db = SessionLocal()
    user = models.User(is_guest=False, email="prem5@example.com")
    db.add(user)
    db.flush()
    db.add(models.Settings(user_id=user.id, api_key="enc:x", model="test-model"))
    adventure = models.Adventure(
        user_id=user.id, title="Before M5", world_state={"player": {"gold": 70}}
    )
    db.add(adventure)
    db.flush()
    branch = models.Branch(adventure_id=adventure.id, lineage=[])
    db.add(branch)
    db.flush()
    branch.lineage = [[branch.id, None]]
    adventure.head_branch_id = branch.id
    adventure.head_depth = 4
    for depth in range(5):
        db.add(models.Action(
            adventure_id=adventure.id, branch_id=branch.id, depth=depth,
            type="ai" if depth % 2 else "do", text=f"old row {depth}", live=True,
            world_state_after={"player": {"gold": depth * 10}},
        ))
    db.add(models.Checkpoint(
        adventure_id=adventure.id, name="Old Save Point",
        branch_id=branch.id, depth=2,
    ))
    db.commit()
    ids = (adventure.id, user.id)
    db.close()

    # Take M5 back off the database, so the migration has real work to do.
    with engine.begin() as conn:
        conn.execute(text("DROP TABLE state_events"))
        conn.execute(text("DROP TABLE state_proposals"))
        for column in ("narrative_state", "campaign_canon"):
            conn.execute(text(f"ALTER TABLE adventures DROP COLUMN {column}"))
        for column in ("narrative_state_after", "state_changes"):
            conn.execute(text(f"ALTER TABLE actions DROP COLUMN {column}"))
        conn.execute(text(f"PRAGMA user_version = {PRE_M5_VERSION}"))

    try:
        yield ids
    finally:
        Base.metadata.drop_all(bind=engine)


@pytest.fixture()
def client(pre_m5, monkeypatch):
    adventure_id, user_id = pre_m5
    migrations.bootstrap(engine)
    monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
    monkeypatch.setattr(adventures.turns, "OpenAICompatibleProvider", ScriptedProvider)
    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 = adventure_id
    try:
        yield test_client
    finally:
        app.dependency_overrides.clear()


def _play(client, text, events):
    ScriptedProvider.replies = [f"A beat.\n{state_block(events)}"]
    r = client.post(f"/api/adventures/{client.adv_id}/actions",
                    json={"type": "do", "text": text})
    assert r.status_code == 200, r.text
    assert '"error"' not in r.text, r.text[:200]


def _document(client) -> dict:
    r = client.get(f"/api/adventures/{client.adv_id}/state")
    assert r.status_code == 200, r.text
    return r.json()["document"]


def _head(client) -> tuple[int, int]:
    with SessionLocal() as db:
        adventure = db.get(models.Adventure, client.adv_id)
        return adventure.head_branch_id, adventure.head_depth


def _state_matches_head(client) -> bool:
    """The invariant, read out of the database rather than out of the API."""
    from app import head as head_module

    with SessionLocal() as db:
        adventure = db.get(models.Adventure, client.adv_id)
        node = head_module.node_at(db, adventure, adventure.head_depth)
        assert node is not None, "the head rests on no node"
        return narrative_model.normalize(adventure.narrative_state) == \
            narrative_model.normalize(node.narrative_state_after)


CAST = [
    {"type": "create_entity", "entity": "mara", "entity_type": "character",
     "name": "Mara"},
    {"type": "set_entity_attribute", "entity": "mara", "attribute": "mood",
     "value": "wary"},
]


def test_the_migration_backfills_every_existing_action(client):
    """Migration 88. No row is left without a snapshot to restore to."""
    with SessionLocal() as db:
        rows = db.query(models.Action).filter_by(adventure_id=client.adv_id).all()
        assert rows, "the fixture wrote no rows"
        assert all(row.narrative_state_after == narrative_model.empty() for row in rows)


def test_the_migrated_campaign_still_opens_and_keeps_its_history(client):
    r = client.get(f"/api/adventures/{client.adv_id}")
    assert r.status_code == 200, r.text
    assert len(r.json()["actions"]) == 5
    assert client.get(f"/api/adventures/{client.adv_id}/checkpoints").json()[0]["name"] \
        == "Old Save Point"
    assert _document(client)["entities"] == {}


def test_restoring_a_pre_m5_save_point_leaves_no_later_state_standing(client):
    """The review's reproduction, end to end.

    Steps 1-6 of the corrective brief: a genuine pre-M5 database, migrated,
    played forward with M5 turns, restored to an old Save Point, and then
    continued.
    """
    _play(client, "go on", CAST)
    assert _document(client)["entities"]["mara"]["attributes"] == {"mood": "wary"}
    assert _state_matches_head(client)
    played_head = _head(client)

    save_point = client.get(f"/api/adventures/{client.adv_id}/checkpoints").json()[0]
    r = client.post(
        f"/api/adventures/{client.adv_id}/checkpoints/{save_point['id']}/restore")
    assert r.status_code == 200, r.text

    # The transcript is back at the old position, and so is the state.
    assert _head(client)[1] == save_point["depth"]
    assert _document(client)["entities"] == {}, \
        "state from a later position survived the restore"
    assert _state_matches_head(client)

    # Redo forward: the M5 state comes back with the position it belongs to.
    while client.post(f"/api/adventures/{client.adv_id}/redo").status_code == 200:
        assert _state_matches_head(client)
    assert _head(client) == played_head
    assert _document(client)["entities"]["mara"]["attributes"] == {"mood": "wary"}


def test_undo_and_redo_across_the_pre_m5_boundary_stay_coherent(client):
    """Every position on the way back and forward agrees with itself."""
    _play(client, "go on", CAST)
    _play(client, "and on", [
        {"type": "set_entity_attribute", "entity": "mara", "attribute": "mood",
         "value": "calm"}])

    seen = []
    while client.post(f"/api/adventures/{client.adv_id}/undo").status_code == 200:
        assert _state_matches_head(client)
        seen.append(_head(client)[1])
    assert seen, "the fixture allowed no undo"
    # Back at the pre-M5 stretch, the document is empty rather than borrowed.
    assert _document(client)["entities"] == {}

    while client.post(f"/api/adventures/{client.adv_id}/redo").status_code == 200:
        assert _state_matches_head(client)
    assert _document(client)["entities"]["mara"]["attributes"] == {"mood": "calm"}


def test_a_pre_m5_campaign_can_be_continued_normally(client):
    """No legacy RPG machinery becomes authoritative again on the way."""
    _play(client, "go on", CAST)
    document = _document(client)
    assert document["entities"]["mara"]["name"] == "Mara"
    with SessionLocal() as db:
        adventure = db.get(models.Adventure, client.adv_id)
        # The old numbers are still on the row, and still not the state.
        assert adventure.world_state == {"player": {"gold": 70}}
        assert "gold" not in str(adventure.narrative_state)