Artifex/tests/test_standalone_scenes.py

from __future__ import annotations

import json
import re
from pathlib import Path

import pytest

from control_plane.authoring.models import (
    DocumentAuthority,
    DocumentType,
    SceneDraftStatus,
    Series,
    StoryProject,
    Work,
    WorkType,
)
from control_plane.authoring.scene_context import build_scene_context_pack
from control_plane.authoring.sources import register_source
from control_plane.authoring.standalone_scenes import StandaloneSceneService
from control_plane.projects.models import Project
from model_router.router import (
    ModelCapability,
    ModelChunk,
    ModelRequestContract,
    ModelResponseContract,
    ModelRouter,
)


class FakeSceneProvider:
    provider_name = "test"

    def complete(self, request: ModelRequestContract) -> ModelResponseContract:
        if str(request.purpose) == str(ModelCapability.STORY_PLANNING):
            content = json.dumps(
                {
                    "purpose": "Sabine gives Corin a precise answer.",
                    "pov_character": "Sabine",
                    "tense": "past",
                    "location": "Corin's office",
                    "time_context": "ordinary daytime work",
                    "present": ["Sabine", "Corin"],
                    "beats": [
                        {"text": "Sabine crosses the room.", "required": True},
                        {"text": "Corin waits for her answer.", "required": True},
                        {"text": "Sabine leaves on her own timing.", "required": True},
                        {"text": "Corin does not turn it into a negotiation.", "required": True},
                        {"text": "Ordinary work remains unfinished.", "required": True},
                    ],
                    "exact_values": [],
                    "constraints": [],
                    "forbidden_events": [],
                    "ending_state": "Sabine has left the office.",
                    "final_image": "Corin remains beside the unfinished paperwork.",
                    "boundary_constraints": [],
                    "continuity_questions": [],
                }
            )
        else:
            requirement_ids = re.findall(r'"id":\s*"([^"]+)"', request.prompt)
            content = json.dumps(
                {
                    "passed": True,
                    "requirement_results": [
                        {
                            "requirement_id": requirement_id,
                            "status": "HIT",
                            "evidence_quote": "Sabine crossed the room.",
                            "details": "Satisfied in the scene.",
                        }
                        for requirement_id in requirement_ids
                    ],
                    "findings": [],
                    "observed_state": {"scene_end": {"location": "office"}},
                    "proposed_changes": [],
                }
            )
        return ModelResponseContract(model="test-model", content=content, metadata={})

    def stream(self, _request: ModelRequestContract):
        prose = "Sabine crossed the room. " + "She waited. " * 245
        yield ModelChunk(prose + "[[END_OF_SCENE]]")

    def health(self) -> str:
        return "AVAILABLE"


def scene_work(tmp_path: Path) -> Work:
    series = Series.objects.create(title="Labyrinth Hero", slug="labyrinth-hero")
    work = Work.objects.create(series=series, title="The Fortune Below", slug="the-fortune-below")
    project = Project.objects.create(name="The Fortune Below", project_type="STORY", goal="Write")
    StoryProject.objects.create(
        project=project,
        work=work,
        title=work.title,
        series=series.title,
        slug=work.slug,
        artifact_root=str(tmp_path / "artifacts"),
    )
    return work


def add_canon_source(work: Work, tmp_path: Path) -> None:
    root = tmp_path / "sources"
    root.mkdir()
    source = root / "relationship-canon.md"
    source.write_text(
        "# Relationship Canon\n\n"
        "Sabine and Corin preserve explicit choice and ordinary work boundaries.\n",
        encoding="utf-8",
    )
    register_source(
        work=work,
        path=source,
        root=root,
        authority=DocumentAuthority.CANON,
        document_type=DocumentType.CANON,
    )


def test_context_pack_only_uses_latest_allowed_authority(tmp_path: Path) -> None:
    work = scene_work(tmp_path)
    root = tmp_path / "sources"
    root.mkdir()
    source = root / "facts.md"
    source.write_text("# Facts\n\nSabine works at the estate.\n", encoding="utf-8")
    register_source(
        work=work,
        path=source,
        root=root,
        authority=DocumentAuthority.CANON,
        document_type=DocumentType.CANON,
    )
    register_source(
        work=work,
        path=source,
        root=root,
        authority=DocumentAuthority.SUPERSEDED,
        document_type=DocumentType.CANON,
    )

    pack, ranked = build_scene_context_pack(work=work, query="Sabine estate")

    assert ranked == []
    assert pack["citations"] == []


def test_context_pack_includes_series_reference_but_not_sibling_book(tmp_path: Path) -> None:
    work = scene_work(tmp_path)
    reference = Work.objects.create(
        series=work.series,
        title="Series Reference",
        slug="series-reference",
        work_type=WorkType.SERIES_REFERENCE,
    )
    sibling = Work.objects.create(
        series=work.series,
        title="Sibling Book",
        slug="sibling-book",
    )
    root = tmp_path / "series-sources"
    root.mkdir()
    reference_path = root / "shared.md"
    sibling_path = root / "sibling.md"
    reference_path.write_text("Sabine follows the shared household rule.", encoding="utf-8")
    sibling_path.write_text("Sabine ignores a sibling-only invention.", encoding="utf-8")
    for source_work, path in [(reference, reference_path), (sibling, sibling_path)]:
        register_source(
            work=source_work,
            path=path,
            root=root,
            authority=DocumentAuthority.CANON,
            document_type=DocumentType.CANON,
        )

    pack, _ranked = build_scene_context_pack(work=work, query="Sabine shared sibling invention")

    keys = {citation["document_key"] for citation in pack["citations"]}
    assert "shared.md" in keys
    assert "sibling.md" not in keys


def test_pinned_planning_sources_do_not_crowd_out_canon(tmp_path: Path) -> None:
    work = scene_work(tmp_path)
    root = tmp_path / "mixed-sources"
    root.mkdir()
    canon_path = root / "canon.md"
    planning_path = root / "planning.md"
    canon_path.write_text(
        "\n\n".join(f"Sabine canon boundary {index}." for index in range(4)),
        encoding="utf-8",
    )
    planning_path.write_text(
        "\n\n".join(f"Sabine planning detail {index}." for index in range(10)),
        encoding="utf-8",
    )
    register_source(
        work=work,
        path=canon_path,
        root=root,
        authority=DocumentAuthority.CANON,
        document_type=DocumentType.CANON,
    )
    register_source(
        work=work,
        path=planning_path,
        root=root,
        authority=DocumentAuthority.PLANNING,
        document_type=DocumentType.PLANNING,
    )

    pack, _ranked = build_scene_context_pack(
        work=work,
        query="Sabine boundary planning detail",
        authorities=[DocumentAuthority.CANON, DocumentAuthority.PLANNING],
        pinned_document_keys=["planning.md"],
        limit=6,
    )

    authorities = {citation["authority"] for citation in pack["citations"]}
    assert authorities == {DocumentAuthority.CANON, DocumentAuthority.PLANNING}


def test_governing_document_is_supplied_in_full_before_rag(tmp_path: Path) -> None:
    work = scene_work(tmp_path)
    root = tmp_path / "governing-sources"
    root.mkdir()
    governing_path = root / "rules.md"
    canon_path = root / "canon.md"
    governing_text = "# Rules\n\nFirst governing rule.\n\nFinal governing rule.\n"
    governing_path.write_text(governing_text, encoding="utf-8")
    canon_path.write_text("Sabine has an additional canon fact.", encoding="utf-8")
    register_source(
        work=work,
        path=governing_path,
        root=root,
        authority=DocumentAuthority.PLANNING,
        document_type=DocumentType.PLANNING,
    )
    register_source(
        work=work,
        path=canon_path,
        root=root,
        authority=DocumentAuthority.CANON,
        document_type=DocumentType.CANON,
    )

    pack, _ranked = build_scene_context_pack(
        work=work,
        query="Sabine canon",
        authorities=[DocumentAuthority.CANON, DocumentAuthority.PLANNING],
        governing_document_keys=["rules.md"],
        limit=2,
    )

    assert pack["governing_document_keys"] == ["rules.md"]
    assert pack["citations"][0]["kind"] == "governing_document"
    assert pack["citations"][0]["sha256"]
    assert "# Rules" in pack["rendered_context"]
    assert "First governing rule." in pack["rendered_context"]
    assert "Final governing rule." in pack["rendered_context"]
    assert "additional canon fact" in pack["rendered_context"]


def test_standalone_scene_runs_from_cited_plan_to_provisional_approval(tmp_path: Path) -> None:
    work = scene_work(tmp_path)
    add_canon_source(work, tmp_path)
    provider = FakeSceneProvider()
    service = StandaloneSceneService(ModelRouter({"terra": provider, "luna": provider}))
    scene = service.create(
        work=work,
        title="Fourteen Seconds",
        brief="Sabine interrupts Corin's ordinary work, gives him a deliberate answer, and leaves.",
        target_words=300,
        constraints=["Sabine owns the timing."],
        forbidden_events=["Do not turn this into a scored turn."],
        boundary_constraints=["Stop when Sabine leaves the office."],
    )

    service.prepare_context(scene)
    assert scene.status == SceneDraftStatus.PLANNING
    assert scene.context_citations.count() == 1

    service.plan(scene)
    assert scene.status == SceneDraftStatus.PLAN_REVIEW
    assert scene.context_citations.count() == 1
    assert scene.context_pack["citations"][0]["authority"] == DocumentAuthority.CANON
    assert scene.contract_requirements
    blocking_beats = [
        item
        for item in scene.contract_requirements
        if item["type"] == "BEAT" and item["blocking"]
    ]
    assert len(blocking_beats) == 5

    service.approve_plan(scene)
    service.write(scene)
    scene.prose = scene.prose.removeprefix("# Fourteen Seconds\n\n")
    scene.save()
    service.review(scene)
    service.approve(scene, actor="test")
    scene.refresh_from_db()

    assert scene.status == SceneDraftStatus.APPROVED
    assert scene.prose.startswith("# Fourteen Seconds\n\n")
    assert scene.word_count >= 250
    assert scene.review["passed"] is True
    assert scene.source_version.authority == DocumentAuthority.PROVISIONAL
    assert len(scene.generation_metadata["planning"]["prompt_sha256"]) == 64
    assert len(scene.generation_metadata["prose"]["prompt_sha256"]) == 64
    assert scene.generation_metadata["approval"]["forced"] is False
    assert Path(scene.artifact_uri).exists()
    assert Path(scene.review_artifact_uri).exists()

    scene.prose += " Changed after approval."
    with pytest.raises(ValueError, match="immutable"):
        scene.save()