428 lines
16 KiB
Python
428 lines
16 KiB
Python
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",
|
|
]
|