"""M10: the media seam — K01-K04, the packet, the profiles, the contracts.

Lineage behaviour has its own file (`test_m10_lineage.py`), as does the
authority separation (`test_m10_authority.py`) and the no-media claim
(`test_m10_no_media.py`), because those three are the claims a reviewer will
want to find whole rather than scattered.

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

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.media import profiles as visual_profiles
from app.media import providers
from app.routers import adventures

import m10_fixture
from fakes import ScriptedProvider


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="m10@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="The Office")
    setup.add(adventure)
    setup.flush()
    setup.add(models.Action(
        adventure_id=adventure.id, type="start",
        text="Bill badges in on a Tuesday morning.",
    ))
    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)


@pytest.fixture()
def office(client):
    return m10_fixture.build(client, client.adv_id)


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


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


# ------------------------------------------------------------------- K01

def test_k01_a_structured_scene_is_persisted_for_a_multi_character_scene(
    client, office
):
    """K01. A scene with several characters and a clear location, **persisted**.

    The acceptance text forbids satisfying this with an ephemeral dictionary
    built inside a test, so the assertion is made against what a *second*
    session reads out of the database — not against a value this test computed.
    """
    with SessionLocal() as db:
        adventure = db.get(models.Adventure, client.adv_id)
        stored = adventure.narrative_state["scene"]

    assert stored["summary"] == "Bill, Alice and Roger meet around the table."
    assert stored["location"] == "office"
    assert sorted(stored["present"]) == ["alice", "bill", "roger"]
    # The coordinate is what makes it a scene *snapshot* rather than a note: it
    # says which accepted position this describes.
    assert stored["at"]["branch_id"] is not None
    assert isinstance(stored["at"]["depth"], int)


def test_k01_the_persisted_scene_is_sufficient_to_depict(client, office):
    """Sufficiency, checked as "could something draw this?" rather than "is it non-empty?"."""
    p = packet_of(client)
    assert p["location"]["name"] == "The office"
    assert [c["name"] for c in p["characters"]] == ["Bill", "Alice", "Roger"]
    assert p["action_summary"] == "Bill, Alice and Roger meet around the table."
    assert p["objects"] and p["objects"][0]["name"] == "Security badge"
    assert p["scene_id"]


def test_the_scene_snapshot_is_per_position_and_survives_a_restart(client, office):
    """Persisted in the ordinary sense: a new session reads the same thing.

    A genuine process restart is exercised in `test_m10_lineage.py`; this is the
    cheaper claim that the value is on disk rather than in a live object.
    """
    with SessionLocal() as first:
        before = first.get(models.Adventure, client.adv_id).narrative_state["scene"]
    with SessionLocal() as second:
        after = second.get(models.Adventure, client.adv_id).narrative_state["scene"]
    assert before == after


# ------------------------------------------------------------------- K02/K03

def test_k02_a_character_keeps_stable_visual_descriptors(client, office):
    row = client.get(
        f"/api/adventures/{client.adv_id}/visual-profiles/alice"
    ).json()
    assert row["descriptors"]["hair"] == "short black"
    assert row["features"] == ["tortoiseshell glasses"]
    assert row["style_notes"] == "photographic, natural light"


def test_k03_a_location_keeps_stable_visual_descriptors(client, office):
    row = client.get(
        f"/api/adventures/{client.adv_id}/visual-profiles/office"
    ).json()
    assert row["descriptors"]["architecture"] == "open-plan floor"
    assert row["features"] == ["whiteboard covered in diagrams"]


def test_profiles_survive_more_turns(client, office):
    """K02/K03 across turns: playing on does not disturb a profile."""
    for i in range(3):
        m10_fixture.play(client, client.adv_id, f"talk {i}", [])
    row = client.get(
        f"/api/adventures/{client.adv_id}/visual-profiles/alice"
    ).json()
    assert row["descriptors"]["hair"] == "short black"


def test_an_item_may_have_a_profile_too(client, office):
    """§5's optional third kind, and proof the one table holds all three.

    There is no `kind` column: a character, a location and an item are all
    entities in the M5 model, and the profile attaches to the entity key.
    """
    response = client.put(
        f"/api/adventures/{client.adv_id}/visual-profiles/badge",
        json={"descriptors": {"material": "white plastic"},
              "features": ["photo in the corner"]},
    )
    assert response.status_code == 200, response.text[:300]
    assert packet_of(client)["objects"][0]["visual_profile"]["descriptors"] == {
        "material": "white plastic"
    }


def test_no_profile_is_distinguishable_from_an_empty_one(client, office):
    """A future provider must be able to tell "unstated" from "stated as nothing"."""
    p = packet_of(client)
    by_name = {c["name"]: c for c in p["characters"]}
    assert by_name["Roger"]["visual_profile"] is None
    assert by_name["Alice"]["visual_profile"] is not None

    client.put(f"/api/adventures/{client.adv_id}/visual-profiles/roger", json={})
    again = {c["name"]: c for c in packet_of(client)["characters"]}
    assert again["Roger"]["visual_profile"] == {
        "descriptors": {}, "features": [], "style_notes": ""
    }


def test_a_profile_must_name_an_entity_the_campaign_has(client, office):
    """A typo is an error, not a row describing nobody."""
    response = client.put(
        f"/api/adventures/{client.adv_id}/visual-profiles/alicce",
        json={"descriptors": {"hair": "short black"}},
    )
    assert response.status_code == 400
    assert "no entity called" in response.json()["detail"]


def test_a_profile_replaces_rather_than_merges(client, office):
    """So a descriptor can be removed, which a merge would make impossible."""
    client.put(f"/api/adventures/{client.adv_id}/visual-profiles/alice",
               json={"descriptors": {"hair": "short black"}})
    row = client.get(
        f"/api/adventures/{client.adv_id}/visual-profiles/alice"
    ).json()
    assert row["descriptors"] == {"hair": "short black"}
    assert row["features"] == []


def test_deleting_a_profile_leaves_the_entity_alone(client, office):
    """A profile is a description. Removing it removes a description."""
    assert client.delete(
        f"/api/adventures/{client.adv_id}/visual-profiles/alice"
    ).status_code == 204
    assert "alice" in state_of(client)["entities"]
    assert {c["name"] for c in packet_of(client)["characters"]} == {
        "Bill", "Alice", "Roger"
    }


@pytest.mark.parametrize("bad", [
    {"descriptors": {"hair": ["short", "black"]}},
    {"descriptors": "short black hair"},
    {"features": "glasses"},
    {"style_notes": {"note": "photographic"}},
    {"descriptors": {"hair": "x" * 5_000}},
])
def test_a_malformed_profile_is_refused(client, office, bad):
    response = client.put(
        f"/api/adventures/{client.adv_id}/visual-profiles/alice", json=bad
    )
    assert response.status_code == 400, response.text[:200]


# --------------------------------------------------------- scene identity

def test_scene_identity_resolves_back_to_a_position(client, office):
    """§3. A future asset holding this string can find the accepted scene again."""
    p = packet_of(client)
    resolved = scene_packet.parse_scene_id(p["scene_id"])
    assert resolved["adventure_id"] == client.adv_id
    assert resolved["branch_id"] == p["turn_range"]["branch_id"]
    assert resolved["start"] == p["turn_range"]["start"]
    assert resolved["end"] == p["turn_range"]["end"]


def test_a_scene_may_span_several_turns(client, office):
    """§3: one turn is not assumed to be one scene, which a video needs."""
    p = packet_of(client, start=0, end=4)
    assert p["turn_range"]["start"] == 0
    assert p["turn_range"]["end"] == 4
    assert p["scene_id"].endswith(":0-4")
    assert scene_packet.parse_scene_id(p["scene_id"])["end"] == 4


def test_a_reversed_range_is_read_in_order(client, office):
    assert packet_of(client, start=4, end=0)["turn_range"] == \
        packet_of(client, start=0, end=4)["turn_range"]


def test_several_assets_may_name_one_scene(client, office):
    """§3: nothing allocates or records a scene, so nothing bounds how many
    future assets refer to it. Two builds of the same scene agree exactly."""
    assert packet_of(client)["scene_id"] == packet_of(client)["scene_id"]


# ------------------------------------------------------- the packet's bounds

def test_the_packet_does_not_carry_the_transcript(client, office):
    """§12. A provider gets the scene, not the campaign."""
    for i in range(4):
        m10_fixture.play(client, client.adv_id, f"say something memorable {i}", [],
                         prose=f"Roger tells a long story about the printer {i}.")
    blob = repr(packet_of(client))
    assert "printer" not in blob
    assert "Bill badges in on a Tuesday morning" not in blob


def test_the_packet_carries_no_imported_knowledge_at_all(client, office):
    """Not just secrets: imported material as a class stays out.

    A positive control comes with it — the source really was imported and really
    does reach the narrator — so this cannot pass because the upload failed.
    """
    m10_fixture.upload_handbook(client, client.adv_id)
    m10_fixture.play(client, client.adv_id, "ask about the north wall panelling", [])

    report = client.get(f"/api/adventures/{client.adv_id}/context").json()
    assert any("handbook" in r["filename"] for r in report["knowledge"]["used"]), (
        "the control failed: the narrator never saw the handbook, so this "
        "proves nothing about the packet"
    )
    assert "refurbished" not in repr(packet_of(client))


def test_the_packet_is_bounded_when_the_state_is_large(client, office):
    """A scene with many entities does not produce an unbounded packet.

    Thirty extras rather than more, because `set_scene`'s `present` is itself
    capped at `validate.MAX_LABELS` (40) — asking for more gets the *event*
    refused and leaves the previous scene standing, which would make this test
    pass by measuring the wrong scene. The precondition is asserted first for
    exactly that reason.
    """
    extras = [f"extra_{i}" for i in range(30)]
    m10_fixture.play(client, client.adv_id, "the whole floor arrives",
                     [m10_fixture.entity(k, "character", f"Extra {k[-2:]}")
                      for k in extras])
    m10_fixture.play(client, client.adv_id, "everyone crowds in", [
        {"type": "set_scene", "summary": "The whole floor crowds in.",
         "location": "office",
         "present": ["bill", "alice", "roger"] + extras},
    ])
    present = state_of(client)["scene"]["present"]
    assert len(present) == 33, (
        f"the scene was not set as this test intends ({len(present)} present), "
        f"so the bound below would be measuring the wrong scene"
    )
    p = packet_of(client)
    assert len(p["characters"]) == scene_packet.MAX_CHARACTERS
    assert len(p["continuity_constraints"]) <= scene_packet.MAX_CONSTRAINTS


# ------------------------------------------------------- provider contracts

def test_no_provider_is_registered(client):
    """v1 ships none, and nothing registers one at import."""
    assert providers.registered() == {}
    for kind in providers.MEDIA_KINDS:
        assert providers.for_kind(kind) == []


def test_a_provider_can_be_added_without_touching_story_code(client, office):
    """M10's Definition of Done, as an executable claim.

    A provider is registered, asked to depict the current scene, and returns —
    and nothing in the story engine was modified, imported or subclassed to make
    that work. The adapter satisfies a `Protocol`, so it did not even have to
    import the base class.
    """
    seen = {}

    class FakeImageProvider:
        def capabilities(self):
            return providers.ProviderCapabilities(
                provider_id="fake-local", kinds=(providers.IMAGE,),
            )

        async def generate(self, request):
            seen["scene_id"] = request.scene["scene_id"]
            return providers.MediaResult(
                kind=providers.IMAGE, media_type="image/png",
                data=b"\x89PNG\r\n\x1a\n",
                provenance={"scene_id": request.scene["scene_id"]},
            )

    provider = FakeImageProvider()
    assert isinstance(provider, providers.MediaProvider)
    providers.register("fake-local", provider)
    try:
        assert providers.for_kind(providers.IMAGE) == [provider]
        import asyncio

        p = packet_of(client)
        result = asyncio.run(provider.generate(
            providers.MediaRequest(kind=providers.IMAGE, scene=p)
        ))
        assert result.media_type == "image/png"
        assert result.provenance["scene_id"] == p["scene_id"]
        assert seen["scene_id"] == p["scene_id"]
    finally:
        providers.unregister("fake-local")
    assert providers.registered() == {}


def test_every_required_media_kind_is_accommodated(client):
    assert set(providers.MEDIA_KINDS) == {"image", "video", "audio", "tts", "stt"}


def test_a_request_for_an_unknown_kind_is_refused(client, office):
    with pytest.raises(ValueError, match="hologram"):
        providers.MediaRequest(kind="hologram", scene=packet_of(client))


def test_stt_returns_a_draft_and_not_a_result(client):
    """§10, and the reason the return type differs.

    A transcription cannot be handed to something expecting a finished artefact,
    because it is not one — it is text the reader is going to edit.
    """
    class FakeStt:
        def capabilities(self):
            return providers.ProviderCapabilities(
                provider_id="fake-stt", kinds=(providers.STT,))

        async def transcribe(self, audio, hints=None):
            return providers.DraftTranscription(text="i open teh door")

    import asyncio

    stt = FakeStt()
    assert isinstance(stt, providers.TranscriptionProvider)
    draft = asyncio.run(stt.transcribe(b"\x00\x01"))
    assert isinstance(draft, providers.DraftTranscription)
    assert not isinstance(draft, providers.MediaResult)
    assert draft.editable is True


def test_an_stt_draft_has_no_route_into_the_story(client, office):
    """The corrected text enters the way anything the reader types does.

    Asserted by playing the edited draft through the ordinary action endpoint
    and observing that it is an ordinary turn — validated, refereed, snapshotted
    — rather than by asserting that some bypass does not exist.
    """
    draft = providers.DraftTranscription(text="i open teh door")
    corrected = draft.text.replace("teh", "the")

    before = len(client.get(f"/api/adventures/{client.adv_id}").json()["actions"])
    m10_fixture.play(client, client.adv_id, corrected, [])
    after = client.get(f"/api/adventures/{client.adv_id}").json()["actions"]
    assert len(after) == before + 2
    assert after[-2]["text"].endswith("i open the door.")


def test_the_story_engine_holds_no_provider_vocabulary(client):
    """§9. Provider syntax must not appear in Story Engine code.

    Greps rather than trusting the boundary, so a future adapter's vocabulary
    cannot leak in unnoticed.

    **`app/media/` is excluded, and the exclusion is the point rather than a
    hole.** §9's rule is about the *Story Engine*; `media/` is the seam, and its
    docstrings name ComfyUI, Whisper and `num_inference_steps` precisely in
    order to say that those belong to a future adapter and not here. A grep that
    failed on the sentence forbidding a thing would push the explanation out of
    the code, which is the opposite of what the rule wants.

    What would catch a violation inside `media/` is not this test but the shape
    of the package: it registers no provider (`test_no_provider_is_registered`),
    ships no adapter, and imports nothing that could reach one.
    """
    import pathlib

    root = pathlib.Path(__file__).resolve().parent.parent / "app"
    seam = root / "media"
    forbidden = ("comfyui", "stable diffusion", "stable-diffusion", "automatic1111",
                 "num_inference_steps", "cfg_scale", "denoising_strength",
                 "safetensors", "whisper", "kokoro", "flux.1")
    offenders = []
    for path in root.rglob("*.py"):
        if seam in path.parents:
            continue
        lowered = path.read_text().lower()
        for word in forbidden:
            if word in lowered:
                offenders.append(f"{path.relative_to(root)}: {word}")
    assert offenders == [], offenders


def test_the_seam_ships_no_adapter(client):
    """The other half of the rule above, for `app/media/` itself.

    The seam is allowed to *name* a provider in prose; it is not allowed to
    *be* one. Checked by what it does rather than by what it says: no provider
    registered, and no HTTP client imported anywhere in the package.
    """
    import pathlib

    assert providers.registered() == {}
    seam = pathlib.Path(__file__).resolve().parent.parent / "app" / "media"
    for path in seam.rglob("*.py"):
        body = path.read_text()
        for client_lib in ("import httpx", "import requests", "urllib.request",
                           "import socket", "subprocess"):
            assert client_lib not in body, f"{path.name} imports {client_lib}"


# ------------------------------------------------------------ endpoint policy

def test_a_media_endpoint_must_be_loopback(client):
    """§11 and contract §27-28: stricter than the narrator's policy, on purpose."""
    assert providers.endpoint_rejection_reason("http://127.0.0.1:8188") is None
    assert providers.endpoint_rejection_reason("http://localhost:8188") is None


def test_a_trusted_lan_media_endpoint_is_refused(client):
    """Allowed for narrator inference; not for media, which has no v1 use."""
    reason = providers.endpoint_rejection_reason("http://192.168.1.50:8188")
    assert reason is not None
    assert "on this machine" in reason


@pytest.mark.parametrize("url", [
    "https://api.example.com/v1",
    "http://8.8.8.8:8188",
    "",
    "not a url",
])
def test_a_non_local_media_endpoint_is_refused(client, url):
    assert providers.endpoint_rejection_reason(url) is not None


def test_check_endpoint_raises_for_a_refused_endpoint(client):
    with pytest.raises(providers.EndpointRejected):
        providers.check_endpoint("https://api.example.com/v1")
    providers.check_endpoint("http://127.0.0.1:8188")


# ------------------------------------------------------------------- K04

def test_k04_the_extension_point_a_future_asset_would_attach_through(client, office):
    """K04, on the acceptance text's **deferred** branch — see the M10 report §F.

    No media tables exist, so this demonstrates the equivalent extension point
    rather than a stored asset: a dummy local byte fixture is carried through
    the provider contract, and the association it needs is proved to resolve.

    What is actually asserted is the part that would matter to a real asset:
    the provenance it carries names a scene, that name resolves to an accepted
    position, and the story is untouched either side.
    """
    p = packet_of(client)
    before_state = state_of(client)
    before_actions = client.get(f"/api/adventures/{client.adv_id}").json()["actions"]

    dummy = providers.MediaResult(
        kind=providers.IMAGE,
        media_type="image/png",
        data=b"\x89PNG\r\n\x1a\n\x00fixture",
        provenance={"scene_id": p["scene_id"],
                    "turn_range": p["turn_range"],
                    "campaign_id": p["campaign"]["id"]},
    )

    resolved = scene_packet.parse_scene_id(dummy.provenance["scene_id"])
    assert resolved["adventure_id"] == client.adv_id
    assert resolved["branch_id"] == p["turn_range"]["branch_id"]

    # The position it names is a real accepted turn in this campaign.
    with SessionLocal() as db:
        found = db.query(models.Action).filter(
            models.Action.adventure_id == client.adv_id,
            models.Action.branch_id == resolved["branch_id"],
            models.Action.depth == resolved["end"],
        ).count()
    assert found >= 1

    # And nothing about the story moved.
    assert state_of(client) == before_state
    assert client.get(f"/api/adventures/{client.adv_id}").json()["actions"] == \
        before_actions