interactive-story/backend/tests/test_worldstate_integration.py

"""M5: the RPG world state is demoted, and this file is what holds it there.

This was the end-to-end test for Phase 12's RPG world state: a scenario with a
`stat_schema`, a turn whose reply carried a relative-delta block, the referee
applying it, and undo rolling it back.

**M5 removed that pipeline from the turn path** (ADR 010). Relative deltas were
ambiguous by construction — a number in a delta field is syntactically legal
whether the model meant "add 50" or "set to 50" — and the genre-neutral typed
events in `app/narrative/` replaced them. The engine in `app/worldstate/` is
retained so a pre-M5 database opens unchanged and its numbers stay readable, and
its own logic is still covered by `test_worldstate.py`; what no longer exists is
its authority over new turns.

So this file now tests the demotion, which is worth a test precisely because it
is an absence: nothing else would notice if the delta pipeline were quietly
rewired, and a second authoritative state engine is the specific failure the M5
brief names.

What was deleted, and why, rather than being re-instrumented:

* `test_turn_applies_clamped_delta_and_strips_block`, `test_action_world_changes_summary`,
  `test_world_state_endpoint`, `test_undo_reverts_world_state`,
  `test_retry_does_not_double_apply` — each asserted that a turn applies a
  delta through the referee. That is the removed capability itself, not
  instrumentation for something else. The invariants they *also* touched —
  undo restoring state, a retry not double-applying — are alive and better
  covered in `test_head_cursor.py`, `test_take_state.py` and
  `test_narrative_state.py`, now measured through the production state path.

    python -m pytest tests/test_worldstate_integration.py -v
"""
import pytest
from fastapi import Depends
from fastapi.testclient import TestClient

from app import auth, limits, models, worldstate
from app.database import Base, SessionLocal, engine, get_db
from app.main import app
from app.routers import adventures

from fakes import ScriptedProvider

SCHEMA = {
    "player": {"hp": {"min": 0, "max": 100, "initial": 100, "max_delta_per_turn": 30}},
    "npcs": {
        "gwen": {
            "name": "Gwen", "keys": "Gwen",
            "desc": "A loyal ranger ally.",
            "stats": {"trust": {"min": -100, "max": 100, "initial": 0}},
        },
    },
    "flags": {"alarm": {"desc": "The enemy is alerted", "initial": False}},
    "milestones": {"win": {"desc": "Win the fight"}},
}

# The faked model narrates and appends a delta that exceeds the per-turn
# cap, so the test can confirm the engine clamps it. It also flips a flag
# and completes a milestone.
AI_REPLY = (
    "The goblin's blade bites deep and Gwen nods at your resolve.\n\n"
    '```state\n{"player.hp": -80, "npc.gwen.trust": 15, "flags.alarm": true, "milestones.win": true}\n```'
)


@pytest.fixture()
def client(monkeypatch):
    Base.metadata.create_all(bind=engine)
    setup = SessionLocal()
    user = models.User(is_guest=False, email="rpg@example.com")
    setup.add(user)
    setup.flush()
    setup.add(models.Settings(user_id=user.id, api_key="enc:dummy", model="test-model"))
    scenario = models.Scenario(user_id=user.id, title="Dungeon", stat_schema=SCHEMA)
    setup.add(scenario)
    setup.flush()
    adv = models.Adventure(
        user_id=user.id, scenario_id=scenario.id, title="Run",
        world_state=worldstate.instantiate(SCHEMA),
    )
    setup.add(adv)
    setup.flush()
    # "Gwen" in the story text makes her NPC in-scene (matches the "gwen" npc's keys).
    setup.add(models.Action(adventure_id=adv.id, type="start",
                            text="You face a goblin. Gwen watches."))
    setup.commit()
    adv_id, user_id = adv.id, user.id
    setup.close()

    ScriptedProvider.replies = [AI_REPLY]
    monkeypatch.setattr(adventures.turns, "OpenAICompatibleProvider", ScriptedProvider)
    monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)

    def _current_user(db=Depends(get_db)):
        return db.get(models.User, user_id)

    app.dependency_overrides[auth.get_current_user] = _current_user

    c = TestClient(app)
    c.adv_id = adv_id
    try:
        yield c
    finally:
        app.dependency_overrides.clear()
        adventures.turns._active_turns.clear()
        Base.metadata.drop_all(bind=engine)


def _world(adv_id):
    db = SessionLocal()
    try:
        return db.get(models.Adventure, adv_id).world_state
    finally:
        db.close()


def _last_ai_text(adv_id):
    db = SessionLocal()
    try:
        adv = db.get(models.Adventure, adv_id)
        return adv.actions[-1].text
    finally:
        db.close()


def _play(client, text="attack the goblin"):
    r = client.post(f"/api/adventures/{client.adv_id}/actions", json={"type": "do", "text": text})
    assert r.status_code == 200, r.text
    return r


def test_a_turn_no_longer_applies_a_relative_delta_block(client):
    """The demotion, asserted.

    The fixture's reply carries the old protocol. Under Phase 12 this moved hp
    from 100 to 70. Under M5 the referee is not in the turn path at all, so the
    numbers stay where they were — and the block still leaves the prose, because
    a reader must never see the protocol whichever protocol it is.
    """
    _play(client)

    world = _world(client.adv_id)
    assert world["player"]["hp"] == 100, "the delta pipeline is still wired into turns"
    assert world.get("milestones", {}) == {}
    assert world.get("flags", {}) in ({}, {"alarm": False})

    text = _last_ai_text(client.adv_id)
    assert "```state" not in text, "the protocol reached the reader"
    assert "goblin's blade" in text

    # The raw reply is still kept, so the Insights view and the audit can show
    # what the model actually sent.
    db = SessionLocal()
    try:
        snapshot = db.get(models.Adventure, client.adv_id).actions[-1].context_snapshot
    finally:
        db.close()
    assert '"player.hp": -80' in snapshot["raw_output"]


def test_an_old_style_block_proposes_no_narrative_events(client):
    """A delta block is not a typed proposal, and must not be read as one.

    `{"player.hp": -80}` is a JSON object with no `events` list. It parses, so
    it is not malformed; it simply proposes nothing. What matters is that no
    part of it is coerced into an event — a state engine that guessed here would
    reintroduce exactly the ambiguity ADR 010 removed.
    """
    _play(client)
    db = SessionLocal()
    try:
        proposal = (
            db.query(models.StateProposal)
            .filter_by(adventure_id=client.adv_id)
            .order_by(models.StateProposal.id.desc())
            .first()
        )
        assert proposal is not None, "no proposal was recorded for the turn"
        assert db.query(models.StateEvent).filter_by(
            adventure_id=client.adv_id).count() == 0
    finally:
        db.close()


def test_the_narrative_state_is_what_the_turn_now_establishes(client):
    """And the replacement works on the same campaign, in the same turn shape."""
    ScriptedProvider.replies = [
        'The blade turns aside.\n```state\n{"events": ['
        '{"type": "create_entity", "entity": "gwen", "entity_type": "character",'
        ' "name": "Gwen"},'
        '{"type": "set_entity_status", "entity": "gwen", "status": "wary"}'
        ']}\n```'
    ]
    _play(client, "parry")

    r = client.get(f"/api/adventures/{client.adv_id}/state")
    assert r.status_code == 200, r.text
    document = r.json()["document"]
    assert document["entities"]["gwen"]["status"] == "wary"


def test_override_world_state_endpoint(client):
    r = client.put(f"/api/adventures/{client.adv_id}/world-state",
                    json={"player.hp": 5, "flags.alarm": True, "npc.bogus.trust": 1})
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["state"]["player"]["hp"] == 5
    assert body["state"]["flags"]["alarm"] is True
    assert body["report"]["rejected"][0]["reason"] == "unknown npc"
    # persisted to the DB, not just the response.
    assert _world(client.adv_id)["player"]["hp"] == 5

    # This bypasses max_delta_per_turn (30) because it is a direct correction, not a turn.
    r = client.put(f"/api/adventures/{client.adv_id}/world-state", json={"player.hp": 100})
    assert r.json()["state"]["player"]["hp"] == 100