from __future__ import annotations import json from io import StringIO from pathlib import Path from unittest.mock import patch import pytest from django.core.management import call_command from django.test import Client from django.urls import reverse from control_plane.authoring.models import ( BookStateStatus, BookStateVersion, DocumentAuthority, DocumentType, SceneDraftStatus, Series, StandaloneScene, Work, ) from control_plane.authoring.sources import register_source from control_plane.authoring.standalone_scenes import ( SceneIdeationService, render_scene_ideation_compact_markdown, render_scene_ideation_markdown, ) from model_router.router import ModelRequestContract, ModelResponseContract, ModelRouter class FakeIdeationProvider: provider_name = "test" def complete(self, _request: ModelRequestContract) -> ModelResponseContract: return ModelResponseContract( model="test-ideas", content=json.dumps( { "candidates": [ { "title": "The Unused Chair", "brief": "Sabine and Corin negotiate an ordinary household choice.", "purpose": "Spend an unresolved agency question through action.", "placement": "After Sabine begins paid estate work.", "pov_character": "Sabine", "scene_type": "quiet_connection", "type_fit": "The scene changes meaning through a freely chosen quiet presence.", "scope_fit": "Sabine's paid estate work exists within Book Six.", "prerequisites": ["Sabine has begun paid estate work."], "target_words": 1600, "citations": ["SRC-01", "SRC-02"], "opportunity": "Turn a stated boundary into an ordinary decision.", "future_opportunities": [ "Sabine can later delegate authority on her own terms.", "Corin can face a higher-cost choice not to intervene.", ], "constraints": ["Sabine makes the final choice."], "forbidden_events": ["No sexual escalation."], "boundary_constraints": ["The decision remains reversible."], "continuity_questions": ["Confirm exact chronology."], "risks": ["Do not make employment repayment for rescue."], }, { "title": "A Route Left Open", "brief": "Corin declines to optimize a shared evening for Sabine.", "purpose": "Test whether restraint can produce a new romantic option.", "placement": "During an unassigned Book Two interval.", "pov_character": "Corin", "scene_type": "major_turn", "type_fit": "Corin's refusal changes their available relationship choices.", "scope_fit": "The established relationship question is active in Book Six.", "prerequisites": ["Sabine and Corin know each other."], "target_words": 1900, "citations": ["SRC-01"], "opportunity": ( "Use an unspent choice without resolving later milestones." ), "future_opportunities": [ "Their unfinished route can acquire a different meaning later." ], "constraints": [], "forbidden_events": ["No completed intercourse."], "boundary_constraints": ["Envelope After remains sealed."], "continuity_questions": [], "risks": [], }, ] } ), metadata={}, ) def health(self) -> str: return "AVAILABLE" def idea_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", ) root = tmp_path / "idea-sources" root.mkdir() sources = [ ("canon.md", "Sabine preserves explicit choice and independent authority.", "canon"), ("planning.md", "Sabine begins paid estate work only after freedom.", "planning"), ("experiment.md", "Sabine accepts an invented irreversible promise.", "provisional"), ] for name, content, authority in sources: path = root / name path.write_text(content, encoding="utf-8") register_source( work=work, path=path, root=root, authority=authority, document_type=( DocumentType.CANON if authority == DocumentAuthority.CANON else DocumentType.PLANNING ), ) return work def test_ideation_is_cited_read_only_and_selection_is_idempotent(tmp_path: Path) -> None: work = idea_work(tmp_path) service = SceneIdeationService(ModelRouter({"sol": FakeIdeationProvider()})) idea = service.propose( work=work, target_book="Book Six", focus="Sabine ordinary choice agency", candidate_count=2, governing_document_keys=["planning.md"], model_hint="sol", ) assert StandaloneScene.objects.count() == 0 assert {item["authority"] for item in idea.context_pack["citations"]} == { DocumentAuthority.CANON, DocumentAuthority.PLANNING, } assert [item["candidate_id"] for item in idea.candidates] == ["idea-01", "idea-02"] assert idea.candidates[0]["future_opportunities"] == [ "Sabine can later delegate authority on her own terms.", "Corin can face a higher-cost choice not to intervene.", ] assert len(idea.generation_metadata["prompt_sha256"]) == 64 assert idea.target_book == "Book Six" assert "physical_escalation" in idea.requested_scene_types assert idea.context_pack["governing_document_keys"] == ["planning.md"] assert idea.context_pack["citations"][0]["kind"] == "governing_document" scene, created = service.select_candidate(idea, candidate_id="idea-01") same_scene, created_again = service.select_candidate(idea, candidate_id="idea-01") assert created is True assert created_again is False assert same_scene.id == scene.id assert scene.status == SceneDraftStatus.PLANNING assert scene.title == "The Unused Chair" assert scene.constraints == [ "Sabine makes the final choice.", "Placement scope: Book Six. Sabine's paid estate work exists within Book Six.", ] assert StandaloneScene.objects.count() == 1 def test_scene_idea_api_proposes_and_selects_candidate(tmp_path: Path) -> None: work = idea_work(tmp_path) service = SceneIdeationService(ModelRouter({"sol": FakeIdeationProvider()})) with patch("control_plane.authoring.views.ideation_service", return_value=service): response = Client().post( reverse("scene_ideas"), data=json.dumps( { "series_slug": work.series.slug, "work_slug": work.slug, "target_book": "Book Six", "focus": "Sabine ordinary choice agency", "candidate_count": 2, "model": "sol", } ), content_type="application/json", ) idea_id = response.json()["id"] selected = Client().post( reverse("scene_idea_action", args=[idea_id]), data=json.dumps({"action": "select", "candidate_id": "idea-02"}), content_type="application/json", ) selected_again = Client().post( reverse("scene_idea_action", args=[idea_id]), data=json.dumps({"action": "select", "candidate_id": "idea-02"}), content_type="application/json", ) assert response.status_code == 201 assert response.json()["candidates"][0]["candidate_id"] == "idea-01" assert selected.status_code == 201 assert selected.json()["created"] is True assert selected.json()["scene"]["status"] == SceneDraftStatus.PLANNING assert selected_again.status_code == 200 assert selected_again.json()["created"] is False def test_book_state_ideation_binds_selected_scene_to_chapter(tmp_path: Path) -> None: work = idea_work(tmp_path) state = BookStateVersion.objects.create( work=work, version=1, status=BookStateStatus.APPROVED, content={ "title": work.title, "chapters": [ { "chapter_key": "chapter-1", "title": "Chapter 1", "purpose": "Open the route.", } ], }, ) service = SceneIdeationService(ModelRouter({"sol": FakeIdeationProvider()})) idea = service.propose( work=work, target_book="Book Two", candidate_count=2, model_hint="sol", book_state=state, ) with pytest.raises(ValueError, match="book_chapter_key"): service.select_candidate(idea, candidate_id="idea-01") scene, created = service.select_candidate( idea, candidate_id="idea-01", book_chapter_key="chapter-1", ) assert created is True assert idea.book_state == state assert idea.context_pack["book_state_sha256"] == state.sha256 assert scene.book_state == state assert scene.book_chapter_key == "chapter-1" def test_ideation_rejects_citations_outside_frozen_context() -> None: with pytest.raises(ValueError, match="supplied citation IDs"): SceneIdeationService._normalize_candidates( { "candidates": [ { "title": "Unsupported Idea", "brief": "A proposal grounded in evidence the model did not receive.", "purpose": "Verify citation validation.", "opportunity": "Spend an unsupported question.", "scene_type": "quiet_connection", "type_fit": "A quiet choice changes the relationship.", "scope_fit": "All prerequisites exist in the selected book.", "citations": ["SRC-99"], "future_opportunities": ["A later choice becomes available."], } ] }, candidate_count=1, citation_ids={"SRC-01"}, ) def test_ideation_requires_future_opportunities() -> None: with pytest.raises(ValueError, match="future_opportunities"): SceneIdeationService._normalize_candidates( { "candidates": [ { "title": "Closed Door", "brief": "A choice closes one route without resolving the relationship.", "purpose": "Spend an established question.", "opportunity": "The unresolved question can now be answered through action.", "scene_type": "quiet_connection", "type_fit": "The choice changes their physical distance.", "scope_fit": "All prerequisites exist in the selected book.", "citations": ["SRC-01"], } ] }, candidate_count=1, citation_ids={"SRC-01"}, ) def test_ideation_markdown_export_is_deterministic_and_read_only(tmp_path: Path) -> None: work = idea_work(tmp_path) service = SceneIdeationService(ModelRouter({"sol": FakeIdeationProvider()})) idea = service.propose( work=work, target_book="Book Six", focus="Sabine ordinary choice agency", candidate_count=2, model_hint="sol", ) expected = render_scene_ideation_markdown(idea) output = tmp_path / "reviews" / "ideas.md" stdout = StringIO() call_command( "fiction_ideas", "export", "--id", str(idea.id), "--output", str(output), stdout=stdout, ) assert output.read_text(encoding="utf-8") == expected assert "# Scene Ideas: The Fortune Below" in expected assert "## idea-01: The Unused Chair" in expected assert "**Scene type:** `quiet_connection`" in expected assert "### Future Opportunities Created" in expected assert "Sabine can later delegate authority on her own terms." in expected assert "- No candidate has been selected." in expected assert "- Target book: `Book Six`" in expected assert "- Requested scene types:" in expected assert "- Governing documents: (none)" in expected assert "### Book Scope Fit" in expected assert "### Scene Type Fit" in expected assert "## Frozen Citation Index" in expected assert "`canon.md`" in expected assert str(output) in stdout.getvalue() assert StandaloneScene.objects.count() == 0 def test_compact_ideation_normalization_and_export(tmp_path: Path) -> None: candidates = SceneIdeationService._normalize_candidates( { "candidates": [ { "title": "A Deliberate Threshold", "brief": "Corin makes a physical choice Sabine did not design for him.", "scene_type": "physical_escalation", "citations": ["SRC-01"], "opportunity": "Reveal what Corin independently wants.", "future_opportunities": [ "Sabine can recognize the choice when he makes it again.", "Corin can decide whether to cross the next threshold.", ], } ] }, candidate_count=1, citation_ids={"SRC-01"}, allowed_scene_types={"physical_escalation"}, compact=True, ) work = idea_work(tmp_path) service = SceneIdeationService(ModelRouter({"sol": FakeIdeationProvider()})) idea = service.propose( work=work, target_book="Book Six", candidate_count=2, model_hint="sol", ) idea.candidates = candidates idea.save(update_fields=["candidates", "updated_at"]) rendered = render_scene_ideation_compact_markdown(idea) output = tmp_path / "compact-ideas.md" call_command( "fiction_ideas", "export", "--id", str(idea.id), "--output", str(output), "--compact", ) assert output.read_text(encoding="utf-8") == rendered assert "### Brief" in rendered assert "### Opportunity Spent" in rendered assert "### Future Opportunities Created" in rendered assert "### Evaluation:" in rendered assert "### Feedback:" in rendered assert "### Purpose" not in rendered assert "### Constraints" not in rendered assert "## Frozen Citation Index" not in rendered assert set(candidates[0]) == { "candidate_id", "title", "brief", "scene_type", "citations", "opportunity", "future_opportunities", } def test_ideation_requires_target_book(tmp_path: Path) -> None: work = idea_work(tmp_path) service = SceneIdeationService(ModelRouter({"sol": FakeIdeationProvider()})) with pytest.raises(ValueError, match="target_book is required"): service.propose(work=work, target_book="", candidate_count=2, model_hint="sol") def test_ideation_allows_repeated_explicit_scene_type() -> None: data = json.loads(FakeIdeationProvider().complete(None).content) for candidate in data["candidates"]: candidate["scene_type"] = "physical_escalation" candidates = SceneIdeationService._normalize_candidates( data, candidate_count=2, citation_ids={"SRC-01", "SRC-02"}, allowed_scene_types={"physical_escalation"}, ) assert [candidate["scene_type"] for candidate in candidates] == [ "physical_escalation", "physical_escalation", ]