"""M10 §4 and §17: scene data obeys the history rules, because it *is* story data.

The claim this file makes is unusual, and worth stating plainly before the
tests: **M10 wrote no lineage code.** There is no media head, no `active` flag,
no scene branch table and no separate restore path. The scene lives in the
authoritative narrative state document, which M3 gave a head, M4 gave Save
Points, M5 gave per-position snapshots and M9 gave portability — so it inherits
every one of those rules by being the same data rather than by copying them.

That makes these tests a check on an inheritance rather than on an
implementation, and they are written to fail loudly if the inheritance were ever
broken by a future scene store appearing beside the state document. The M10
brief's §4 sequence is exercised literally, including the restart, and the
Mara-in-the-cellar example it names is the first test.

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

import json
import os
import shutil
import sqlite3
import tempfile
import urllib.request
from pathlib import Path

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

from app import auth, limits, memorybank, models
from app.database import Base, SessionLocal, engine, get_db
from app.knowledge import embeddings
from app.main import app
from app.media import packet as scene_packet
from app.routers import adventures

import m10_fixture
from fakes import ScriptedProvider
from test_process_restart import Server, _free_port


class StubDerived:
    async def complete(self, system, prompt, **kwargs):
        return "A memory."

    async def embed(self, texts):
        return [[1.0, 0.5, 0.25] for _ in texts]


@pytest.fixture()
def client(monkeypatch):
    Base.metadata.create_all(bind=engine)
    memorybank._vector_cache.clear()
    embeddings._cache.clear()
    setup = SessionLocal()
    user = models.User(is_guest=False, email="m10lin@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=400,
    ))
    adventure = models.Adventure(user_id=user.id, title="Lineage")
    setup.add(adventure)
    setup.flush()
    setup.add(models.Action(
        adventure_id=adventure.id, type="start", text="It begins.",
    ))
    setup.commit()
    adv_id, user_id = adventure.id, user.id
    setup.close()

    monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
    monkeypatch.setattr(adventures.turns, "OpenAICompatibleProvider", ScriptedProvider)
    monkeypatch.setattr(memorybank, "embedding_provider", lambda s: StubDerived())
    monkeypatch.setattr(memorybank, "summary_provider", lambda s: StubDerived())
    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
    try:
        yield test_client
    finally:
        app.dependency_overrides.clear()
        adventures.turns._active_turns.clear()
        memorybank._vector_cache.clear()
        embeddings._cache.clear()
        Base.metadata.drop_all(bind=engine)


def scene_of(client, adv_id=None):
    return client.get(
        f"/api/adventures/{adv_id or client.adv_id}/state"
    ).json()["document"].get("scene") or {}


def packet_of(client, adv_id=None):
    r = client.get(f"/api/adventures/{adv_id or client.adv_id}/scene-packet")
    assert r.status_code == 200, r.text[:300]
    return r.json()


def retained_scenes(adv_id) -> list[tuple]:
    """Every scene the tree still holds, as (branch, depth, summary).

    Read from the per-position snapshots, which is where a retained scene lives
    — the point being that a scene the story left is still on disk, attached to
    the position that established it.
    """
    from sqlalchemy.orm import undefer

    with SessionLocal() as db:
        rows = (
            db.query(models.Action)
            .filter(models.Action.adventure_id == adv_id)
            .options(undefer(models.Action.narrative_state_after))
            .order_by(models.Action.branch_id, models.Action.depth, models.Action.id)
            .all()
        )
    out = []
    for row in rows:
        state = row.narrative_state_after or {}
        summary = (state.get("scene") or {}).get("summary")
        if summary:
            out.append((row.branch_id, row.depth, summary))
    return out


# --------------------------------------------- the brief's own §4 example

def test_a_scene_from_an_abandoned_line_does_not_become_current(client):
    """§4, literally: Mara in the cellar, then Mara upstairs.

    Path A's scene must remain stored, must not be current on Path B, and
    Path B's scene must be Path B's.
    """
    m10_fixture.play(client, client.adv_id, "set up", [
        m10_fixture.entity("mara", "character", "Mara"),
        m10_fixture.entity("cellar", "location", "The cellar"),
        m10_fixture.entity("upstairs", "location", "Upstairs"),
    ])
    m10_fixture.play(client, client.adv_id, "go down", [
        {"type": "set_scene", "summary": "Mara enters the cellar.",
         "location": "cellar", "present": ["mara"]},
    ])
    assert scene_of(client)["summary"] == "Mara enters the cellar."
    path_a = packet_of(client)["scene_id"]

    assert client.post(f"/api/adventures/{client.adv_id}/undo").status_code == 200
    m10_fixture.play(client, client.adv_id, "stay put", [
        {"type": "set_scene", "summary": "Mara remains upstairs.",
         "location": "upstairs", "present": ["mara"]},
    ])

    current = scene_of(client)
    assert current["summary"] == "Mara remains upstairs."
    assert current["location"] == "upstairs"
    assert packet_of(client)["location"]["name"] == "Upstairs"
    assert packet_of(client)["scene_id"] != path_a

    # Path A's scene is still on disk, on the branch it belongs to.
    kept = retained_scenes(client.adv_id)
    assert ("Mara enters the cellar." in [s for _, _, s in kept]), kept
    assert ("Mara remains upstairs." in [s for _, _, s in kept]), kept
    branches = {s: b for b, _, s in kept}
    assert branches["Mara enters the cellar."] != branches["Mara remains upstairs."]


def test_divergence_deletes_no_scene(client):
    """§4: diverging retains the old line rather than replacing it."""
    m10_fixture.play(client, client.adv_id, "set up", [
        m10_fixture.entity("mara", "character", "Mara"),
        m10_fixture.entity("cellar", "location", "The cellar"),
    ])
    m10_fixture.play(client, client.adv_id, "down", [
        {"type": "set_scene", "summary": "Scene A.", "location": "cellar",
         "present": ["mara"]}])
    before = len(retained_scenes(client.adv_id))
    client.post(f"/api/adventures/{client.adv_id}/undo")
    m10_fixture.play(client, client.adv_id, "elsewhere", [
        {"type": "set_scene", "summary": "Scene C.", "location": "cellar",
         "present": ["mara"]}])
    after = retained_scenes(client.adv_id)
    assert len(after) == before + 1
    assert "Scene A." in [s for _, _, s in after]


# ------------------------------------------------- the brief's §17 sequence

def test_the_full_scene_lineage_sequence(client):
    """§17, step by step, in one test so the order is the thing under test.

    Scene A, Save Point, Scene B, Undo, Redo, restore, diverge to Scene C — and
    at every step the active scene must be the one the head is on, while the
    scenes the story left must still be on disk.
    """
    adv = client.adv_id
    m10_fixture.play(client, adv, "set up", [
        m10_fixture.entity("mara", "character", "Mara"),
        m10_fixture.entity("hall", "location", "The hall"),
    ])
    client.put(f"/api/adventures/{adv}/visual-profiles/mara",
               json={"descriptors": {"build": "sturdy"}})

    # 1-2. Scene A, persisted.
    m10_fixture.play(client, adv, "scene a", [
        {"type": "set_scene", "summary": "Scene A.", "location": "hall",
         "present": ["mara"]}])
    assert scene_of(client)["summary"] == "Scene A."

    # 3. Save Point at Scene A.
    point = client.post(f"/api/adventures/{adv}/checkpoints",
                        json={"name": "At scene A", "note": ""})
    assert point.status_code == 201, point.text[:300]
    point_id = point.json()["id"]

    # 4. Advance to Scene B.
    m10_fixture.play(client, adv, "scene b", [
        {"type": "set_scene", "summary": "Scene B.", "location": "hall",
         "present": ["mara"]}])
    assert scene_of(client)["summary"] == "Scene B."

    # 5. Undo -> back at Scene A.
    assert client.post(f"/api/adventures/{adv}/undo").status_code == 200
    assert scene_of(client)["summary"] == "Scene A."

    # 6. Redo -> Scene B again.
    assert client.post(f"/api/adventures/{adv}/redo").status_code == 200
    assert scene_of(client)["summary"] == "Scene B."

    # 7. Restore the Save Point -> Scene A, and Scene B is still retained.
    restored = client.post(f"/api/adventures/{adv}/checkpoints/{point_id}/restore")
    assert restored.status_code == 200, restored.text[:300]
    assert scene_of(client)["summary"] == "Scene A."
    assert "Scene B." in [s for _, _, s in retained_scenes(adv)]

    # 8. Diverge to Scene C.
    m10_fixture.play(client, adv, "scene c", [
        {"type": "set_scene", "summary": "Scene C.", "location": "hall",
         "present": ["mara"]}])
    assert scene_of(client)["summary"] == "Scene C."

    # Scene B is retained and is NOT current on Scene C's line.
    kept = [s for _, _, s in retained_scenes(adv)]
    assert "Scene B." in kept and "Scene A." in kept and "Scene C." in kept
    assert scene_of(client)["summary"] == "Scene C."

    # 9-11. Restart, then inspect again. Nothing about eligibility moved.
    with SessionLocal() as fresh:
        adventure = fresh.get(models.Adventure, adv)
        assert adventure.narrative_state["scene"]["summary"] == "Scene C."

    # The profile is stable across every one of those movements.
    profile = client.get(f"/api/adventures/{adv}/visual-profiles/mara").json()
    assert profile["descriptors"] == {"build": "sturdy"}


def test_a_visual_profile_is_stable_across_divergence(client):
    """§17: a character does not change appearance because the story forked.

    This is the one place M10's storage choice is directly observable: profiles
    are campaign-scoped, so the same profile is visible from both lines.
    """
    m10_fixture.play(client, client.adv_id, "set up", [
        m10_fixture.entity("mara", "character", "Mara"),
        m10_fixture.entity("hall", "location", "The hall"),
    ])
    client.put(f"/api/adventures/{client.adv_id}/visual-profiles/mara",
               json={"descriptors": {"hair": "dark auburn"}})
    m10_fixture.play(client, client.adv_id, "a", [
        {"type": "set_scene", "summary": "A.", "location": "hall",
         "present": ["mara"]}])
    on_a = packet_of(client)["characters"][0]["visual_profile"]

    client.post(f"/api/adventures/{client.adv_id}/undo")
    m10_fixture.play(client, client.adv_id, "b", [
        {"type": "set_scene", "summary": "B.", "location": "hall",
         "present": ["mara"]}])
    on_b = packet_of(client)["characters"][0]["visual_profile"]

    assert on_a == on_b == {"descriptors": {"hair": "dark auburn"},
                            "features": [], "style_notes": ""}


def test_a_profile_survives_redo_and_a_save_point_restore(client):
    """The other two history operations, for the profile rather than the scene.

    Divergence is covered above and is the interesting case; Redo and a Save
    Point restore are covered here because K02 claims stability across all of
    them, and a claim in a report should have a test under it rather than an
    argument. Both move the head, and a profile that moved with it would be the
    per-position storage M10 deliberately did not build.
    """
    m10_fixture.play(client, client.adv_id, "set up", [
        m10_fixture.entity("mara", "character", "Mara"),
        m10_fixture.entity("hall", "location", "The hall"),
    ])
    profile = {"descriptors": {"hair": "dark auburn"}, "features": ["a scar"],
               "style_notes": "candlelight"}
    client.put(f"/api/adventures/{client.adv_id}/visual-profiles/mara",
               json=profile)
    point = client.post(f"/api/adventures/{client.adv_id}/checkpoints",
                        json={"name": "Before the hall", "note": ""})
    assert point.status_code == 201, point.text[:300]

    m10_fixture.play(client, client.adv_id, "into the hall", [
        {"type": "set_scene", "summary": "Mara stands in the hall.",
         "location": "hall", "present": ["mara"]}])
    expected = {"descriptors": {"hair": "dark auburn"}, "features": ["a scar"],
                "style_notes": "candlelight"}
    assert packet_of(client)["characters"][0]["visual_profile"] == expected

    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 packet_of(client)["characters"][0]["visual_profile"] == expected

    restored = client.post(
        f"/api/adventures/{client.adv_id}/checkpoints/{point.json()['id']}/restore")
    assert restored.status_code == 200, restored.text[:300]
    # The scene is gone — it was set after the Save Point — and the profile is
    # not, which is exactly the difference between story state and presentation
    # metadata.
    assert packet_of(client)["characters"] == []
    assert client.get(
        f"/api/adventures/{client.adv_id}/visual-profiles/mara"
    ).json()["descriptors"] == {"hair": "dark auburn"}


def test_nothing_relies_on_a_mutable_active_flag(client):
    """§4's last clause, checked structurally rather than by behaviour.

    The scene follows the head because it *is* the state at the head. If a
    future change introduced a scene table with its own `active` column, this
    would be the test that noticed.
    """
    assert not hasattr(models, "Scene")
    columns = {c.name for c in models.VisualProfile.__table__.columns}
    assert "active" not in columns
    assert "branch_id" not in columns
    assert "depth" not in columns


# ------------------------------------------------- a genuine process restart

@pytest.fixture()
def spawned():
    """A real server process against a real database file, twice.

    `test_process_restart.py` owns the harness; M10 reuses it because "survives
    a restart" is a claim about bytes on disk, and a same-process fixture cannot
    tell durable state from a live object.
    """
    directory = tempfile.mkdtemp(prefix="m10-restart-")
    db_path = os.path.join(directory, "campaign.db")
    started: list[Server] = []

    def start() -> Server:
        server = Server(db_path, _free_port())
        started.append(server)
        server.wait_until_ready()
        return server

    try:
        yield start, db_path
    finally:
        for server in started:
            server.stop()
        shutil.rmtree(directory, ignore_errors=True)


def test_scene_and_profile_survive_a_genuine_process_restart(spawned):
    """K01/K02/K03's durability clause, across a real PID boundary.

    The spawned server narrates with a deterministic provider that emits no
    state events, so the scene and the entities are established through the
    ordinary correction endpoint — which is a real, validated write path, not a
    fixture reaching into the ORM.
    """
    start, db_path = spawned
    first = start()
    campaign = first.call("POST", "/adventures", {
        "title": "Restarted", "opening": "The office is quiet.",
    }, expect=201)
    adv = campaign["id"]

    first.call("POST", f"/adventures/{adv}/state/corrections", {
        "events": [
            {"type": "create_entity", "entity": "alice",
             "entity_type": "character", "name": "Alice"},
            {"type": "create_entity", "entity": "office",
             "entity_type": "location", "name": "The office"},
            {"type": "set_scene", "summary": "Alice waits in the office.",
             "location": "office", "present": ["alice"]},
        ],
        "note": "setting the scene",
    }, expect=201)

    first.call("PUT", f"/adventures/{adv}/visual-profiles/alice",
               {"descriptors": {"hair": "short black"},
                "features": ["tortoiseshell glasses"]}, expect=200)

    before_scene = first.call("GET", f"/adventures/{adv}/state",
                              expect=200)["document"]["scene"]
    before_packet = first.call("GET", f"/adventures/{adv}/scene-packet", expect=200)
    first.stop()
    assert not first.is_listening()

    second = start()
    after_scene = second.call("GET", f"/adventures/{adv}/state",
                              expect=200)["document"]["scene"]
    after_packet = second.call("GET", f"/adventures/{adv}/scene-packet", expect=200)
    after_profile = second.call(
        "GET", f"/adventures/{adv}/visual-profiles/alice", expect=200)

    assert after_scene == before_scene
    assert after_scene["summary"] == "Alice waits in the office."
    assert after_packet == before_packet
    assert after_profile["descriptors"] == {"hair": "short black"}
    assert after_packet["characters"][0]["visual_profile"]["features"] == [
        "tortoiseshell glasses"
    ]


def test_the_restarted_database_holds_the_profile_row(spawned):
    """Read out of the file itself, so "persisted" is not taken on trust."""
    start, db_path = spawned
    server = start()
    campaign = server.call("POST", "/adventures",
                           {"title": "Rows", "opening": "Start."}, expect=201)
    adv = campaign["id"]
    server.call("POST", f"/adventures/{adv}/state/corrections", {
        "events": [{"type": "create_entity", "entity": "ship",
                    "entity_type": "vehicle", "name": "The Persephone"}],
        "note": "",
    }, expect=201)
    server.call("PUT", f"/adventures/{adv}/visual-profiles/ship",
                {"descriptors": {"hull": "pitted white composite"}}, expect=200)
    server.stop()

    connection = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
    try:
        row = connection.execute(
            "SELECT entity_key, descriptors FROM visual_profiles "
            "WHERE adventure_id = ?", (adv,)
        ).fetchone()
    finally:
        connection.close()
    assert row is not None
    assert row[0] == "ship"
    assert json.loads(row[1]) == {"hull": "pitted white composite"}