"""M10 §6 and §18: the media layer cannot write the story. The architectural claim is one sentence — *media is derived presentation, story state is authoritative, and there is no reverse path* — and this file is the part of it that is checked by running things rather than by reading imports. Every test here follows the same shape, which is the shape that makes it evidence rather than assertion: record the authoritative document, byte for byte do the media-layer thing record it again require them to be identical That catches a write nobody intended as well as one somebody did, and it does not depend on knowing *how* a violation would have happened. `test_m10_media_hooks.py` covers what the boundary carries; this covers what it must never push back through. python -m pytest tests/test_m10_authority.py -v """ import copy 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="m10auth@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="Authority") 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) @pytest.fixture() def office(client): return m10_fixture.build(client, client.adv_id) def authoritative(adv_id) -> dict: """Everything the story counts as true, read straight from the database.""" with SessionLocal() as db: adventure = db.get(models.Adventure, adv_id) return { "state": copy.deepcopy(adventure.narrative_state), "head_branch": adventure.head_branch_id, "head_depth": adventure.head_depth, "events": db.query(models.StateEvent).filter( models.StateEvent.adventure_id == adv_id).count(), "proposals": db.query(models.StateProposal).filter( models.StateProposal.adventure_id == adv_id).count(), "actions": db.query(models.Action).filter( models.Action.adventure_id == adv_id).count(), } # ------------------------------------------------------ writes that must not def test_writing_a_visual_profile_changes_no_story_state(client, office): before = authoritative(client.adv_id) response = client.put( f"/api/adventures/{client.adv_id}/visual-profiles/bill", json={"descriptors": {"build": "heavyset", "clothing": "navy suit"}, "features": ["signet ring"], "style_notes": "photographic"}, ) assert response.status_code == 200, response.text[:300] assert authoritative(client.adv_id) == before def test_updating_a_visual_profile_creates_no_state_fact(client, office): """§6's example, made concrete. A profile saying Alice wears a blue coat must not make it true that Alice owns or wears a blue coat. Checked by looking for the words in the authoritative document afterwards, not only by comparing counts. """ before = authoritative(client.adv_id) client.put(f"/api/adventures/{client.adv_id}/visual-profiles/alice", json={"descriptors": {"clothing": "blue coat"}}) after = authoritative(client.adv_id) assert after == before assert "blue coat" not in repr(after["state"]) document = client.get( f"/api/adventures/{client.adv_id}/state").json()["document"] assert not any("blue coat" in repr(f) for f in document["facts"]) assert "blue coat" not in repr(document["entities"]["alice"]) def test_deleting_a_visual_profile_changes_no_story_state(client, office): before = authoritative(client.adv_id) assert client.delete( f"/api/adventures/{client.adv_id}/visual-profiles/alice" ).status_code == 204 assert authoritative(client.adv_id) == before def test_building_a_scene_packet_changes_nothing(client, office): """A packet is a read. Built repeatedly, it must still be a read.""" before = authoritative(client.adv_id) for _ in range(5): assert client.get( f"/api/adventures/{client.adv_id}/scene-packet" ).status_code == 200 assert authoritative(client.adv_id) == before def test_a_scene_packet_does_not_move_the_head(client, office): before = authoritative(client.adv_id) client.get(f"/api/adventures/{client.adv_id}/scene-packet?start=0&end=4") after = authoritative(client.adv_id) assert after["head_branch"] == before["head_branch"] assert after["head_depth"] == before["head_depth"] def test_a_dummy_media_result_cannot_reach_the_story(client, office): """§18: adding a depiction, even a wrong one, changes nothing. The result claims Alice is wearing a red coat and standing in a corridor. None of that is true in the campaign, and after registering, generating and holding the result, none of it has become true. """ import asyncio before = authoritative(client.adv_id) packet = client.get( f"/api/adventures/{client.adv_id}/scene-packet").json() class WrongProvider: def capabilities(self): return providers.ProviderCapabilities( provider_id="wrong", kinds=(providers.IMAGE,)) async def generate(self, request): return providers.MediaResult( kind=providers.IMAGE, media_type="image/png", data=b"\x89PNG\r\n\x1a\n", provenance={"scene_id": request.scene["scene_id"]}, details={"depicts": "Alice in a red coat in a corridor"}, ) providers.register("wrong", WrongProvider()) try: result = asyncio.run(WrongProvider().generate( providers.MediaRequest(kind=providers.IMAGE, scene=packet))) assert "red coat" in result.details["depicts"] finally: providers.unregister("wrong") after = authoritative(client.adv_id) assert after == before assert "red coat" not in repr(after["state"]) assert "corridor" not in repr(after["state"]) def test_a_provider_failure_cannot_advance_the_head(client, office): """§18: a media failure is not a story event.""" import asyncio before = authoritative(client.adv_id) class FailingProvider: def capabilities(self): return providers.ProviderCapabilities( provider_id="failing", kinds=(providers.IMAGE,)) async def generate(self, request): raise providers.MediaProviderError("the local generator is not running") providers.register("failing", FailingProvider()) try: with pytest.raises(providers.MediaProviderError): asyncio.run(FailingProvider().generate(providers.MediaRequest( kind=providers.IMAGE, scene=client.get( f"/api/adventures/{client.adv_id}/scene-packet").json()))) finally: providers.unregister("failing") assert authoritative(client.adv_id) == before def test_a_scene_derivation_failure_does_not_corrupt_an_accepted_turn(client, office): """§18: if building a packet raised, the story would be untouched. The failure is induced in the packet builder itself, which is the only place derivation happens, and the accepted turn either side is compared whole. """ before = authoritative(client.adv_id) original = scene_packet.build def explode(*args, **kwargs): raise RuntimeError("scene derivation failed") scene_packet.build = explode try: response = client.get(f"/api/adventures/{client.adv_id}/scene-packet") assert response.status_code >= 500 except RuntimeError: pass # the TestClient re-raises; either way the story must be intact finally: scene_packet.build = original assert authoritative(client.adv_id) == before # And the campaign still plays. m10_fixture.play(client, client.adv_id, "carry on", []) assert authoritative(client.adv_id)["actions"] == before["actions"] + 2 # ------------------------------------------------- rebuilding derived data def test_deleting_every_visual_profile_leaves_the_campaign_intact(client, office): """§18's last clause: derived data can go without taking the story with it. Profiles are the only thing M10 persists, and they are recoverable only from a bundle or by being written again — so the promise here is narrower than M9's rebuildable indexes, and the test states the narrow thing: removing them costs the descriptions and nothing else. """ before = authoritative(client.adv_id) with SessionLocal() as db: db.query(models.VisualProfile).filter( models.VisualProfile.adventure_id == client.adv_id ).delete(synchronize_session=False) db.commit() assert authoritative(client.adv_id) == before assert client.get( f"/api/adventures/{client.adv_id}/visual-profiles").json()["profiles"] == [] # The packet still builds; it simply describes nobody's appearance. p = client.get(f"/api/adventures/{client.adv_id}/scene-packet").json() assert [c["name"] for c in p["characters"]] == ["Bill", "Alice", "Roger"] assert all(c["visual_profile"] is None for c in p["characters"]) def test_the_story_survives_a_profile_naming_a_vanished_entity(client, office): """A profile whose entity is gone is inert, not a corruption. Reachable through an import: a bundle may carry a profile for an entity that only exists on a branch the campaign has left. """ with SessionLocal() as db: db.add(models.VisualProfile( adventure_id=client.adv_id, entity_key="nobody_at_all", descriptors={"hair": "green"}, features=[], style_notes="")) db.commit() before = authoritative(client.adv_id) p = client.get(f"/api/adventures/{client.adv_id}/scene-packet").json() assert "green" not in repr(p) assert authoritative(client.adv_id) == before m10_fixture.play(client, client.adv_id, "carry on", []) # ---------------------------------------------- the separation, structurally def test_the_media_package_imports_nothing_that_writes_state(client): """The guarantee behind every test above, checked as an import rule. `narrative.apply` and `narrative.store` are the only modules that write the authoritative document, and `media/` reaching either of them would make the separation a convention rather than a fact. `narrative.model` and `narrative.store.current` are reads and are used. """ import pathlib seam = pathlib.Path(__file__).resolve().parent.parent / "app" / "media" for path in seam.rglob("*.py"): body = path.read_text() assert "narrative.apply" not in body, path.name assert "from ..narrative import apply" not in body, path.name assert "set_current" not in body, path.name assert "head.move_to" not in body, path.name assert "tree.place_action" not in body, path.name def test_no_state_event_type_was_added_for_media(client): """M10 adds no way for the media layer to speak in the story's vocabulary.""" from app.narrative import events assert not any( name.startswith("media") or "visual" in name or "asset" in name for name in events.ALLOWED )