1286 lines
52 KiB
Python
1286 lines
52 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from django.conf import settings
|
|
from django.db import transaction
|
|
from django.utils import timezone
|
|
from django.utils.text import slugify
|
|
|
|
from control_plane.authoring.models import (
|
|
DocumentAuthority,
|
|
DocumentType,
|
|
PromptVersion,
|
|
SceneContextCitation,
|
|
SceneDraftStatus,
|
|
SceneIdeation,
|
|
StandaloneScene,
|
|
Work,
|
|
text_sha256,
|
|
)
|
|
from control_plane.authoring.prompts import (
|
|
SCENE_IDEA_TYPES,
|
|
SCENE_IDEATION_COMPACT_TEMPLATE,
|
|
SCENE_IDEATION_SYSTEM,
|
|
SCENE_IDEATION_TEMPLATE,
|
|
STANDALONE_SCENE_PLAN_SYSTEM,
|
|
STANDALONE_SCENE_PLAN_TEMPLATE,
|
|
STANDALONE_SCENE_PROSE_SYSTEM,
|
|
STANDALONE_SCENE_PROSE_TEMPLATE,
|
|
STANDALONE_SCENE_REVIEW_SYSTEM,
|
|
STANDALONE_SCENE_REVIEW_TEMPLATE,
|
|
)
|
|
from control_plane.authoring.scene_context import build_scene_context_pack
|
|
from control_plane.authoring.sources import register_source
|
|
from control_plane.authoring.state_management import (
|
|
build_contract_requirements,
|
|
evidence_is_present,
|
|
json_sha256,
|
|
)
|
|
from control_plane.authoring.streaming import ResumableDraftWriter, atomic_write_text, word_count
|
|
from model_router.providers import extract_json_object
|
|
from model_router.router import ModelCapability, ModelRequestContract, ModelRouter
|
|
|
|
|
|
def render_authoring_prompt(
|
|
purpose: str,
|
|
default_system: str,
|
|
default_template: str,
|
|
**values: Any,
|
|
) -> tuple[str, str | None]:
|
|
stored = PromptVersion.objects.filter(purpose=purpose, is_active=True).first()
|
|
system = stored.system_text if stored else default_system
|
|
template = stored.user_template if stored else default_template
|
|
return f"{system}\n\n{template.format(**values)}", str(stored.id) if stored else None
|
|
|
|
|
|
class SceneIdeationService:
|
|
def __init__(self, router: ModelRouter) -> None:
|
|
self.router = router
|
|
|
|
def propose(
|
|
self,
|
|
*,
|
|
work: Work,
|
|
target_book: str,
|
|
focus: str = "",
|
|
candidate_count: int = 10,
|
|
scene_types: list[str] | None = None,
|
|
authorities: list[str] | None = None,
|
|
pinned_document_keys: list[str] | None = None,
|
|
governing_document_keys: list[str] | None = None,
|
|
detail_level: str = "full",
|
|
model_hint: str | None = None,
|
|
book_state=None,
|
|
) -> SceneIdeation:
|
|
candidate_count = int(candidate_count)
|
|
if not 1 <= candidate_count <= 12:
|
|
raise ValueError("candidate_count must be between 1 and 12")
|
|
target_book = target_book.strip()
|
|
if not target_book:
|
|
raise ValueError("target_book is required")
|
|
if len(target_book) > 160:
|
|
raise ValueError("target_book must be 160 characters or fewer")
|
|
requested_scene_types = list(dict.fromkeys(scene_types or SCENE_IDEA_TYPES))
|
|
invalid_scene_types = sorted(set(requested_scene_types) - set(SCENE_IDEA_TYPES))
|
|
if invalid_scene_types:
|
|
raise ValueError("unsupported scene types: " + ", ".join(invalid_scene_types))
|
|
detail_level = detail_level.strip().lower()
|
|
if detail_level not in {"full", "compact"}:
|
|
raise ValueError("detail_level must be full or compact")
|
|
authorities = authorities or [DocumentAuthority.CANON, DocumentAuthority.PLANNING]
|
|
if book_state:
|
|
if book_state.work_id != work.id:
|
|
raise ValueError("book state must belong to the ideation work")
|
|
if book_state.status != "approved":
|
|
raise ValueError("book state must be approved before ideation")
|
|
focus = focus.strip()
|
|
query = f"{target_book} {focus}".strip() or (
|
|
f"{work.title} unresolved choices open threads relationship pressure "
|
|
"character agency promises boundaries consequences"
|
|
)
|
|
pack, _ranked = build_scene_context_pack(
|
|
work=work,
|
|
query=query,
|
|
authorities=authorities,
|
|
pinned_document_keys=pinned_document_keys,
|
|
governing_document_keys=governing_document_keys,
|
|
limit=32,
|
|
max_chars=150000,
|
|
)
|
|
if book_state:
|
|
book_packet = {
|
|
"book_state_id": str(book_state.id),
|
|
"version": book_state.version,
|
|
"sha256": book_state.sha256,
|
|
"content": book_state.content,
|
|
}
|
|
pack["rendered_context"] = (
|
|
"APPROVED BOOK STATE PLANNING CONTRACT\n"
|
|
+ json.dumps(book_packet, ensure_ascii=False, indent=2)
|
|
+ "\n\n"
|
|
+ pack["rendered_context"]
|
|
)
|
|
pack["book_state_id"] = str(book_state.id)
|
|
pack["book_state_sha256"] = book_state.sha256
|
|
pack["sha256"] = json_sha256(
|
|
{key: value for key, value in pack.items() if key != "sha256"}
|
|
)
|
|
prompt, prompt_version_id = render_authoring_prompt(
|
|
"SCENE_IDEATION",
|
|
SCENE_IDEATION_SYSTEM,
|
|
(
|
|
SCENE_IDEATION_COMPACT_TEMPLATE
|
|
if detail_level == "compact"
|
|
else SCENE_IDEATION_TEMPLATE
|
|
),
|
|
candidate_count=candidate_count,
|
|
work_title=work.title,
|
|
target_book=target_book,
|
|
focus=focus or "Open scan for unspent, evidence-backed scene opportunities.",
|
|
context=pack["rendered_context"],
|
|
scene_types=json.dumps(
|
|
{
|
|
key: SCENE_IDEA_TYPES[key]
|
|
for key in requested_scene_types
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
),
|
|
)
|
|
story = getattr(work, "story_project", None)
|
|
response = self.router.complete(
|
|
ModelRequestContract(
|
|
purpose=ModelCapability.STORY_PLANNING,
|
|
prompt=prompt,
|
|
model_hint=model_hint,
|
|
token_budget=(
|
|
max(3000, min(8000, candidate_count * 700))
|
|
if detail_level == "compact"
|
|
else max(5000, min(12000, candidate_count * 1200))
|
|
),
|
|
project=story.project if story else None,
|
|
)
|
|
)
|
|
candidates = self._normalize_candidates(
|
|
extract_json_object(response.content),
|
|
candidate_count=candidate_count,
|
|
citation_ids={item["id"] for item in pack["citations"]},
|
|
allowed_scene_types=set(requested_scene_types),
|
|
compact=detail_level == "compact",
|
|
)
|
|
return SceneIdeation.objects.create(
|
|
work=work,
|
|
book_state=book_state,
|
|
target_book=target_book,
|
|
focus=focus,
|
|
requested_scene_types=requested_scene_types,
|
|
candidate_count=candidate_count,
|
|
authorities=authorities,
|
|
pinned_document_keys=pinned_document_keys or [],
|
|
context_pack=pack,
|
|
context_pack_sha256=pack["sha256"],
|
|
candidates=candidates,
|
|
generation_metadata={
|
|
"model": response.model,
|
|
"prompt_version_id": prompt_version_id,
|
|
"prompt_sha256": text_sha256(prompt),
|
|
"response_sha256": text_sha256(response.content),
|
|
"context_pack_sha256": pack["sha256"],
|
|
"detail_level": detail_level,
|
|
},
|
|
)
|
|
|
|
@transaction.atomic
|
|
def select_candidate(
|
|
self,
|
|
ideation: SceneIdeation,
|
|
*,
|
|
candidate_id: str,
|
|
target_words: int | None = None,
|
|
book_chapter_key: str | None = None,
|
|
) -> tuple[StandaloneScene, bool]:
|
|
ideation = SceneIdeation.objects.select_for_update().select_related("work").get(
|
|
id=ideation.id
|
|
)
|
|
candidates = list(ideation.candidates or [])
|
|
candidate = next(
|
|
(item for item in candidates if item.get("candidate_id") == candidate_id),
|
|
None,
|
|
)
|
|
if candidate is None:
|
|
raise ValueError("idea candidate not found")
|
|
selected_scene_id = candidate.get("selected_scene_id")
|
|
if selected_scene_id:
|
|
return StandaloneScene.objects.get(id=selected_scene_id), False
|
|
if ideation.book_state_id and not str(book_chapter_key or "").strip():
|
|
raise ValueError("book_chapter_key is required for book-state ideation selection")
|
|
constraints = list(candidate.get("constraints") or [])
|
|
scope_fit = str(candidate.get("scope_fit") or "").strip()
|
|
if ideation.target_book and scope_fit:
|
|
constraints.append(f"Placement scope: {ideation.target_book}. {scope_fit}")
|
|
scene = StandaloneSceneService(self.router).create(
|
|
work=ideation.work,
|
|
title=candidate["title"],
|
|
brief=candidate["brief"],
|
|
target_words=int(target_words or candidate.get("target_words") or 1800),
|
|
constraints=constraints,
|
|
forbidden_events=candidate.get("forbidden_events") or [],
|
|
boundary_constraints=candidate.get("boundary_constraints") or [],
|
|
book_state=ideation.book_state,
|
|
book_chapter_key=book_chapter_key,
|
|
)
|
|
candidate["selected_scene_id"] = str(scene.id)
|
|
ideation.candidates = candidates
|
|
ideation.save(update_fields=["candidates", "updated_at"])
|
|
return scene, True
|
|
|
|
@staticmethod
|
|
def _normalize_candidates(
|
|
data: dict[str, Any],
|
|
*,
|
|
candidate_count: int,
|
|
citation_ids: set[str],
|
|
allowed_scene_types: set[str] | None = None,
|
|
compact: bool = False,
|
|
) -> list[dict[str, Any]]:
|
|
allowed_scene_types = allowed_scene_types or set(SCENE_IDEA_TYPES)
|
|
raw_candidates = data.get("candidates")
|
|
if not isinstance(raw_candidates, list) or len(raw_candidates) != candidate_count:
|
|
raise ValueError(f"ideation must return exactly {candidate_count} candidates")
|
|
normalized = []
|
|
for index, raw in enumerate(raw_candidates, start=1):
|
|
if not isinstance(raw, dict):
|
|
raise ValueError("each idea candidate must be an object")
|
|
title = str(raw.get("title") or "").strip()
|
|
brief = str(raw.get("brief") or "").strip()
|
|
purpose = str(raw.get("purpose") or "").strip()
|
|
opportunity = str(raw.get("opportunity") or "").strip()
|
|
scene_type = str(raw.get("scene_type") or "").strip()
|
|
scope_fit = str(raw.get("scope_fit") or "").strip()
|
|
type_fit = str(raw.get("type_fit") or "").strip()
|
|
raw_citations = raw.get("citations")
|
|
if not isinstance(raw_citations, list):
|
|
raise ValueError("each idea candidate citations value must be a list")
|
|
citations = list(
|
|
dict.fromkeys(str(value).strip() for value in raw_citations if str(value).strip())
|
|
)
|
|
invalid_citations = sorted(set(citations) - citation_ids)
|
|
if not title or not brief or not opportunity or (not compact and not purpose):
|
|
raise ValueError(
|
|
"each idea candidate requires title, brief, opportunity, "
|
|
"and full-detail purpose"
|
|
)
|
|
if scene_type not in allowed_scene_types:
|
|
raise ValueError(
|
|
"each idea candidate requires a requested scene_type: "
|
|
+ ", ".join(sorted(allowed_scene_types))
|
|
)
|
|
if not compact and (not scope_fit or not type_fit):
|
|
raise ValueError("each idea candidate requires scope_fit and type_fit")
|
|
if not citations or invalid_citations:
|
|
raise ValueError("each idea candidate must use only supplied citation IDs")
|
|
future_opportunities = _string_list(raw.get("future_opportunities"))
|
|
if not future_opportunities:
|
|
raise ValueError("each idea candidate requires future_opportunities")
|
|
candidate = {
|
|
"candidate_id": f"idea-{index:02d}",
|
|
"title": title,
|
|
"brief": brief,
|
|
"scene_type": scene_type,
|
|
"citations": citations,
|
|
"opportunity": opportunity,
|
|
"future_opportunities": future_opportunities,
|
|
}
|
|
if not compact:
|
|
proposed_words = int(raw.get("target_words") or 1800)
|
|
candidate.update(
|
|
{
|
|
"purpose": purpose,
|
|
"placement": str(raw.get("placement") or "").strip(),
|
|
"pov_character": str(raw.get("pov_character") or "").strip(),
|
|
"type_fit": type_fit,
|
|
"scope_fit": scope_fit,
|
|
"prerequisites": _string_list(raw.get("prerequisites")),
|
|
"target_words": min(10000, max(300, proposed_words)),
|
|
"constraints": _string_list(raw.get("constraints")),
|
|
"forbidden_events": _string_list(raw.get("forbidden_events")),
|
|
"boundary_constraints": _string_list(raw.get("boundary_constraints")),
|
|
"continuity_questions": _string_list(raw.get("continuity_questions")),
|
|
"risks": _string_list(raw.get("risks")),
|
|
}
|
|
)
|
|
normalized.append(candidate)
|
|
required_distinct = min(candidate_count, len(allowed_scene_types))
|
|
distinct_types = {candidate["scene_type"] for candidate in normalized}
|
|
if len(distinct_types) < required_distinct:
|
|
raise ValueError(
|
|
f"ideation requires {required_distinct} distinct scene types; "
|
|
f"received {len(distinct_types)}"
|
|
)
|
|
return normalized
|
|
|
|
|
|
def _string_list(value: Any) -> list[str]:
|
|
if not isinstance(value, list):
|
|
return []
|
|
return [str(item).strip() for item in value if str(item).strip()]
|
|
|
|
|
|
def render_scene_ideation_markdown(ideation: SceneIdeation) -> str:
|
|
context_pack = ideation.context_pack or {}
|
|
candidates = ideation.candidates or []
|
|
metadata = ideation.generation_metadata or {}
|
|
selected = [
|
|
(candidate.get("candidate_id"), candidate.get("selected_scene_id"))
|
|
for candidate in candidates
|
|
if candidate.get("selected_scene_id")
|
|
]
|
|
lines = [
|
|
f"# Scene Ideas: {_single_line(ideation.work.title)}",
|
|
"",
|
|
"## Review Status",
|
|
"",
|
|
]
|
|
if selected:
|
|
for candidate_id, scene_id in selected:
|
|
lines.append(f"- `{candidate_id}` selected as scene `{scene_id}`.")
|
|
else:
|
|
lines.extend(
|
|
[
|
|
"- No candidate has been selected.",
|
|
"- No scene was created by this proposal.",
|
|
]
|
|
)
|
|
lines.extend(
|
|
[
|
|
"",
|
|
"## Proposal Record",
|
|
"",
|
|
f"- Ideation ID: `{ideation.id}`",
|
|
f"- Created: `{ideation.created_at.isoformat()}`",
|
|
f"- Series: `{ideation.work.series.slug}`",
|
|
f"- Work: `{ideation.work.slug}`",
|
|
f"- Target book: `{ideation.target_book or 'not recorded'}`",
|
|
f"- Candidate count: `{ideation.candidate_count}`",
|
|
f"- Requested scene types: {_code_list(ideation.requested_scene_types)}",
|
|
f"- Governing documents: {_code_list(context_pack.get('governing_document_keys') or [])}",
|
|
f"- Authorities: {_code_list(ideation.authorities)}",
|
|
f"- Focus: {_single_line(ideation.focus) or '(open scan)'}",
|
|
f"- Context pack SHA-256: `{ideation.context_pack_sha256}`",
|
|
f"- Model: `{_single_line(metadata.get('model')) or 'unknown'}`",
|
|
f"- Prompt version ID: `{metadata.get('prompt_version_id') or 'default'}`",
|
|
f"- Prompt SHA-256: `{metadata.get('prompt_sha256') or ''}`",
|
|
f"- Response SHA-256: `{metadata.get('response_sha256') or ''}`",
|
|
]
|
|
)
|
|
if ideation.pinned_document_keys:
|
|
lines.extend(["", "### Pinned Documents", ""])
|
|
_append_markdown_list(lines, ideation.pinned_document_keys)
|
|
|
|
for candidate in candidates:
|
|
candidate_id = _single_line(candidate.get("candidate_id"))
|
|
title = _single_line(candidate.get("title"))
|
|
lines.extend(
|
|
[
|
|
"",
|
|
"---",
|
|
"",
|
|
f"## {candidate_id}: {title}",
|
|
"",
|
|
f"**POV:** {_single_line(candidate.get('pov_character')) or 'Unspecified'} ",
|
|
f"**Scene type:** `{candidate.get('scene_type') or 'not recorded'}` ",
|
|
f"**Target:** {candidate.get('target_words') or 1800:,} words ",
|
|
f"**Selected scene:** `{candidate.get('selected_scene_id') or 'not selected'}`",
|
|
]
|
|
)
|
|
_append_markdown_paragraph(lines, "Brief", candidate.get("brief"))
|
|
_append_markdown_paragraph(lines, "Purpose", candidate.get("purpose"))
|
|
_append_markdown_paragraph(lines, "Placement", candidate.get("placement"))
|
|
_append_markdown_paragraph(lines, "Book Scope Fit", candidate.get("scope_fit"))
|
|
_append_markdown_section(lines, "Prerequisites", candidate.get("prerequisites"))
|
|
_append_markdown_paragraph(lines, "Scene Type Fit", candidate.get("type_fit"))
|
|
_append_markdown_paragraph(lines, "Opportunity Spent", candidate.get("opportunity"))
|
|
_append_markdown_section(
|
|
lines,
|
|
"Future Opportunities Created",
|
|
candidate.get("future_opportunities"),
|
|
empty="Not recorded; this proposal predates future-opportunity tracking.",
|
|
)
|
|
_append_markdown_section(lines, "Constraints", candidate.get("constraints"))
|
|
_append_markdown_section(lines, "Forbidden Events", candidate.get("forbidden_events"))
|
|
_append_markdown_section(
|
|
lines, "Boundary Constraints", candidate.get("boundary_constraints")
|
|
)
|
|
_append_markdown_section(
|
|
lines, "Continuity Questions", candidate.get("continuity_questions")
|
|
)
|
|
_append_markdown_section(lines, "Risks", candidate.get("risks"))
|
|
_append_markdown_section(lines, "Citations", candidate.get("citations"), code=True)
|
|
|
|
lines.extend(["", "---", "", "## Frozen Citation Index", ""])
|
|
for citation in context_pack.get("citations") or []:
|
|
citation_id = _single_line(citation.get("id"))
|
|
document_key = _single_line(citation.get("document_key"))
|
|
start_line = citation.get("start_line")
|
|
end_line = citation.get("end_line")
|
|
line_range = str(start_line) if start_line == end_line else f"{start_line}-{end_line}"
|
|
lines.extend(
|
|
[
|
|
f"### {citation_id}",
|
|
"",
|
|
f"- Document: `{document_key}`",
|
|
f"- Title: {_single_line(citation.get('document_title'))}",
|
|
f"- Version: `{citation.get('document_version')}`",
|
|
f"- Authority: `{citation.get('authority')}`",
|
|
f"- Lines: `{line_range}`",
|
|
f"- Passage SHA-256: `{citation.get('sha256')}`",
|
|
f"- Retrieval score: `{citation.get('score')}`",
|
|
f"- Retrieval reason: {_single_line(citation.get('reason'))}",
|
|
"",
|
|
]
|
|
)
|
|
return "\n".join(lines).rstrip() + "\n"
|
|
|
|
|
|
def render_scene_ideation_compact_markdown(ideation: SceneIdeation) -> str:
|
|
lines = [
|
|
f"# Scene Ideas: {_single_line(ideation.work.title)}",
|
|
"",
|
|
f"- Ideation ID: `{ideation.id}`",
|
|
f"- Target book: `{ideation.target_book or 'not recorded'}`",
|
|
f"- Focus: {_single_line(ideation.focus) or '(open scan)'}",
|
|
]
|
|
for candidate in ideation.candidates or []:
|
|
lines.extend(
|
|
[
|
|
"",
|
|
"---",
|
|
"",
|
|
(
|
|
f"## {_single_line(candidate.get('candidate_id'))}: "
|
|
f"{_single_line(candidate.get('title'))}"
|
|
),
|
|
]
|
|
)
|
|
_append_markdown_paragraph(lines, "Brief", candidate.get("brief"))
|
|
_append_markdown_paragraph(lines, "Opportunity Spent", candidate.get("opportunity"))
|
|
_append_markdown_section(
|
|
lines,
|
|
"Future Opportunities Created",
|
|
candidate.get("future_opportunities"),
|
|
)
|
|
lines.extend(
|
|
[
|
|
"",
|
|
f"### Evaluation: {_single_line(candidate.get('evaluation'))}".rstrip(),
|
|
"",
|
|
f"### Feedback: {_single_line(candidate.get('feedback'))}".rstrip(),
|
|
]
|
|
)
|
|
return "\n".join(lines).rstrip() + "\n"
|
|
|
|
|
|
def export_scene_ideation_markdown(
|
|
ideation: SceneIdeation, output: Path, *, compact: bool = False
|
|
) -> Path:
|
|
content = (
|
|
render_scene_ideation_compact_markdown(ideation)
|
|
if compact
|
|
else render_scene_ideation_markdown(ideation)
|
|
)
|
|
atomic_write_text(output, content)
|
|
return output
|
|
|
|
|
|
def _single_line(value: Any) -> str:
|
|
return " ".join(str(value or "").splitlines()).strip()
|
|
|
|
|
|
def _code_list(values: list[Any]) -> str:
|
|
return ", ".join(f"`{_single_line(value)}`" for value in values) or "(none)"
|
|
|
|
|
|
def _append_markdown_paragraph(lines: list[str], heading: str, value: Any) -> None:
|
|
text = str(value or "").strip()
|
|
if text:
|
|
lines.extend(["", f"### {heading}", "", text])
|
|
|
|
|
|
def _append_markdown_section(
|
|
lines: list[str],
|
|
heading: str,
|
|
values: Any,
|
|
*,
|
|
code: bool = False,
|
|
empty: str = "",
|
|
) -> None:
|
|
items = _string_list(values)
|
|
if not items and not empty:
|
|
return
|
|
lines.extend(["", f"### {heading}", ""])
|
|
_append_markdown_list(lines, items or [empty], code=code)
|
|
|
|
|
|
def _append_markdown_list(lines: list[str], values: list[Any], *, code: bool = False) -> None:
|
|
for value in values:
|
|
text = str(value).strip()
|
|
if not text:
|
|
continue
|
|
if code:
|
|
text = f"`{_single_line(text)}`"
|
|
else:
|
|
text = text.replace("\n", "\n ")
|
|
lines.append(f"- {text}")
|
|
|
|
|
|
class StandaloneSceneService:
|
|
def __init__(self, router: ModelRouter) -> None:
|
|
self.router = router
|
|
self.writer = ResumableDraftWriter(router)
|
|
|
|
@transaction.atomic
|
|
def create(
|
|
self,
|
|
*,
|
|
work: Work,
|
|
title: str,
|
|
brief: str,
|
|
target_words: int = 1800,
|
|
constraints: list[str] | None = None,
|
|
forbidden_events: list[str] | None = None,
|
|
boundary_constraints: list[str] | None = None,
|
|
book_state=None,
|
|
book_chapter_key: str | None = None,
|
|
) -> StandaloneScene:
|
|
work = Work.objects.select_for_update().get(pk=work.pk)
|
|
if not 300 <= int(target_words) <= 10000:
|
|
raise ValueError("target_words must be between 300 and 10000")
|
|
chapter_key = str(book_chapter_key or "").strip()
|
|
if bool(book_state) != bool(chapter_key):
|
|
raise ValueError("book_state and book_chapter_key must be provided together")
|
|
if book_state:
|
|
if book_state.work_id != work.id:
|
|
raise ValueError("book state must belong to the scene work")
|
|
if str(book_state.status).upper() != "APPROVED":
|
|
raise ValueError("book state must be approved before binding a scene")
|
|
latest_run = book_state.runs.order_by("-created_at").first()
|
|
if latest_run and latest_run.status == "complete":
|
|
raise ValueError("start a new book run before revising a completed chapter")
|
|
self._book_chapter(book_state.content or {}, chapter_key)
|
|
scene_key = slugify(title)[:200] or "scene"
|
|
lineage = StandaloneScene.objects.filter(work=work, scene_key=scene_key)
|
|
if book_state:
|
|
chapter_scenes = StandaloneScene.objects.filter(
|
|
work=work,
|
|
book_state=book_state,
|
|
book_chapter_key=chapter_key,
|
|
)
|
|
other_lineage = chapter_scenes.exclude(scene_key=scene_key).exists()
|
|
if other_lineage:
|
|
raise ValueError("book state chapter already has a different scene lineage")
|
|
lineage = lineage.filter(book_state=book_state, book_chapter_key=chapter_key)
|
|
else:
|
|
lineage = lineage.filter(book_state__isnull=True)
|
|
latest = lineage.order_by("-revision").first()
|
|
story = getattr(work, "story_project", None)
|
|
return StandaloneScene.objects.create(
|
|
work=work,
|
|
story=story,
|
|
parent=latest,
|
|
scene_key=scene_key,
|
|
revision=(latest.revision + 1) if latest else 1,
|
|
title=title.strip(),
|
|
brief=brief.strip(),
|
|
target_words=int(target_words),
|
|
constraints=constraints or [],
|
|
forbidden_events=forbidden_events or [],
|
|
boundary_constraints=boundary_constraints or [],
|
|
book_state=book_state,
|
|
book_chapter_key=chapter_key,
|
|
)
|
|
|
|
def plan(
|
|
self,
|
|
scene: StandaloneScene,
|
|
*,
|
|
authorities: list[str] | None = None,
|
|
pinned_document_keys: list[str] | None = None,
|
|
model_hint: str | None = None,
|
|
) -> StandaloneScene:
|
|
if scene.status not in {SceneDraftStatus.PLANNING, SceneDraftStatus.PLAN_REVIEW}:
|
|
raise ValueError(f"scene cannot be planned from status {scene.status}")
|
|
self.prepare_context(
|
|
scene,
|
|
authorities=authorities,
|
|
pinned_document_keys=pinned_document_keys,
|
|
)
|
|
pack = scene.context_pack
|
|
prompt, prompt_version_id = render_authoring_prompt(
|
|
"STANDALONE_SCENE_PLAN",
|
|
STANDALONE_SCENE_PLAN_SYSTEM,
|
|
STANDALONE_SCENE_PLAN_TEMPLATE,
|
|
title=scene.title,
|
|
brief=scene.brief,
|
|
context=pack["rendered_context"],
|
|
constraints=json.dumps(scene.constraints, ensure_ascii=False, indent=2),
|
|
forbidden_events=json.dumps(scene.forbidden_events, ensure_ascii=False, indent=2),
|
|
boundary_constraints=json.dumps(
|
|
scene.boundary_constraints, ensure_ascii=False, indent=2
|
|
),
|
|
target_words=scene.target_words,
|
|
)
|
|
response = self.router.complete(
|
|
ModelRequestContract(
|
|
purpose=ModelCapability.STORY_PLANNING,
|
|
prompt=prompt,
|
|
model_hint=model_hint,
|
|
token_budget=5000,
|
|
project=scene.story.project if scene.story else None,
|
|
)
|
|
)
|
|
plan = self._normalize_plan(extract_json_object(response.content), scene)
|
|
discarded_exact_values = []
|
|
discarded_proposed_beats = []
|
|
if scene.book_state_id:
|
|
plan, discarded_proposed_beats = self._apply_book_chapter_scope(scene, plan)
|
|
plan["exact_values"], discarded_exact_values = self._scoped_exact_values(
|
|
scene, plan["exact_values"]
|
|
)
|
|
requirements = self._requirements(plan)
|
|
scene.plan = plan
|
|
scene.contract_requirements = requirements
|
|
scene.status = SceneDraftStatus.PLAN_REVIEW
|
|
scene.failure_reason = ""
|
|
metadata = dict(scene.generation_metadata or {})
|
|
metadata["planning"] = {
|
|
"model": response.model,
|
|
"prompt_version_id": prompt_version_id,
|
|
"prompt_sha256": text_sha256(prompt),
|
|
"response_sha256": text_sha256(response.content),
|
|
"context_pack_sha256": pack["sha256"],
|
|
"discarded_out_of_scope_exact_values": discarded_exact_values,
|
|
"discarded_model_proposed_beats": discarded_proposed_beats,
|
|
**self._book_state_metadata(scene),
|
|
}
|
|
scene.generation_metadata = metadata
|
|
scene.save()
|
|
return scene
|
|
|
|
def _apply_book_chapter_scope(
|
|
self, scene: StandaloneScene, plan: dict[str, Any]
|
|
) -> tuple[dict[str, Any], list[dict[str, Any]]]:
|
|
chapter, _chapters, _index = self._book_chapter(
|
|
scene.book_state.content or {}, scene.book_chapter_key
|
|
)
|
|
approved_beats = [
|
|
{
|
|
"text": str(item.get("text") or "").strip(),
|
|
"required": bool(item.get("required")),
|
|
}
|
|
for item in chapter.get("beats") or []
|
|
if isinstance(item, dict) and str(item.get("text") or "").strip()
|
|
]
|
|
approved_text = {item["text"].casefold() for item in approved_beats}
|
|
discarded = [
|
|
item
|
|
for item in plan["beats"]
|
|
if str(item.get("text") or "").strip().casefold() not in approved_text
|
|
]
|
|
plan["beats"] = approved_beats
|
|
plan["purpose"] = str(chapter.get("purpose") or "").strip()
|
|
plan["ending_state"] = str(chapter.get("ending_state") or "").strip()
|
|
return plan, discarded
|
|
|
|
def _scoped_exact_values(
|
|
self, scene: StandaloneScene, exact_values: list[Any]
|
|
) -> tuple[list[Any], list[Any]]:
|
|
scope = "\n".join(
|
|
[
|
|
self._book_planning_packet(scene),
|
|
scene.brief,
|
|
json.dumps(scene.constraints, ensure_ascii=False),
|
|
json.dumps(scene.forbidden_events, ensure_ascii=False),
|
|
json.dumps(scene.boundary_constraints, ensure_ascii=False),
|
|
]
|
|
).casefold()
|
|
allowed = []
|
|
discarded = []
|
|
for item in exact_values:
|
|
value = item.get("value") if isinstance(item, dict) else item
|
|
candidate = "" if value is None else str(value).strip().casefold()
|
|
if candidate and candidate in scope:
|
|
allowed.append(item)
|
|
else:
|
|
discarded.append(item)
|
|
return allowed, discarded
|
|
|
|
def prepare_context(
|
|
self,
|
|
scene: StandaloneScene,
|
|
*,
|
|
authorities: list[str] | None = None,
|
|
pinned_document_keys: list[str] | None = None,
|
|
) -> StandaloneScene:
|
|
if scene.status not in {SceneDraftStatus.PLANNING, SceneDraftStatus.PLAN_REVIEW}:
|
|
raise ValueError(f"scene context cannot be changed from status {scene.status}")
|
|
query = "\n".join(
|
|
[scene.title, scene.brief, *scene.constraints, *scene.boundary_constraints]
|
|
)
|
|
pack, ranked = build_scene_context_pack(
|
|
work=scene.work,
|
|
query=query,
|
|
authorities=authorities,
|
|
pinned_document_keys=pinned_document_keys,
|
|
)
|
|
if scene.book_state_id:
|
|
packet = self._book_planning_packet(scene)
|
|
pack["rendered_context"] = f"{packet}\n\n{pack['rendered_context']}"
|
|
pack.update(self._book_state_metadata(scene))
|
|
pack["sha256"] = json_sha256(
|
|
{key: value for key, value in pack.items() if key != "sha256"}
|
|
)
|
|
with transaction.atomic():
|
|
scene.context_query = query
|
|
scene.context_pack = pack
|
|
scene.context_pack_sha256 = pack["sha256"]
|
|
scene.save(
|
|
update_fields=[
|
|
"context_query",
|
|
"context_pack",
|
|
"context_pack_sha256",
|
|
"updated_at",
|
|
]
|
|
)
|
|
scene.context_citations.all().delete()
|
|
SceneContextCitation.objects.bulk_create(
|
|
[
|
|
SceneContextCitation(
|
|
scene=scene,
|
|
passage=item.passage,
|
|
rank=index,
|
|
score=item.score,
|
|
reason=item.reason,
|
|
)
|
|
for index, item in enumerate(ranked, start=1)
|
|
]
|
|
)
|
|
return scene
|
|
|
|
def approve_plan(self, scene: StandaloneScene) -> StandaloneScene:
|
|
if scene.status != SceneDraftStatus.PLAN_REVIEW or not scene.plan:
|
|
raise ValueError("scene has no plan awaiting approval")
|
|
scene.status = SceneDraftStatus.READY
|
|
scene.plan_approved_at = timezone.now()
|
|
scene.save(update_fields=["status", "plan_approved_at", "updated_at"])
|
|
return scene
|
|
|
|
def write(
|
|
self,
|
|
scene: StandaloneScene,
|
|
*,
|
|
model_hint: str | None = None,
|
|
max_attempts: int = 2,
|
|
) -> StandaloneScene:
|
|
if scene.status not in {SceneDraftStatus.READY, SceneDraftStatus.FAILED}:
|
|
raise ValueError(f"scene cannot be written from status {scene.status}")
|
|
if not scene.plan_approved_at:
|
|
raise ValueError("scene plan must be approved before writing")
|
|
prompt, prompt_version_id = render_authoring_prompt(
|
|
"STANDALONE_SCENE_PROSE",
|
|
STANDALONE_SCENE_PROSE_SYSTEM,
|
|
STANDALONE_SCENE_PROSE_TEMPLATE,
|
|
title=scene.title,
|
|
brief=scene.brief,
|
|
context=(scene.context_pack or {}).get("rendered_context")
|
|
or "(No approved source context.)",
|
|
plan=json.dumps(scene.plan, ensure_ascii=False, indent=2),
|
|
requirements=json.dumps(scene.contract_requirements, ensure_ascii=False, indent=2),
|
|
target_words=scene.target_words,
|
|
)
|
|
partial_path, artifact_path, _review_path = self._artifact_paths(scene)
|
|
scene.status = SceneDraftStatus.DRAFTING
|
|
scene.partial_artifact_uri = str(partial_path)
|
|
scene.failure_reason = ""
|
|
scene.save(update_fields=["status", "partial_artifact_uri", "failure_reason", "updated_at"])
|
|
try:
|
|
result = self.writer.generate(
|
|
request=ModelRequestContract(
|
|
purpose=ModelCapability.STORY_PROSE,
|
|
prompt=prompt,
|
|
model_hint=model_hint,
|
|
token_budget=max(4000, int(scene.target_words * 2.2)),
|
|
project=scene.story.project if scene.story else None,
|
|
),
|
|
partial_path=partial_path,
|
|
minimum_words=max(250, int(scene.target_words * 0.6)),
|
|
maximum_words=max(600, int(scene.target_words * 1.8)),
|
|
completion_marker="[[END_OF_SCENE]]",
|
|
max_attempts=max(1, int(max_attempts)),
|
|
)
|
|
except Exception as exc:
|
|
scene.status = SceneDraftStatus.FAILED
|
|
scene.failure_reason = str(exc)
|
|
scene.save(update_fields=["status", "failure_reason", "updated_at"])
|
|
raise
|
|
prose = self._scene_document(scene.title, result.text)
|
|
atomic_write_text(artifact_path, prose + "\n")
|
|
scene.prose = prose
|
|
scene.artifact_uri = str(artifact_path)
|
|
scene.status = SceneDraftStatus.DRAFT_REVIEW
|
|
metadata = dict(scene.generation_metadata or {})
|
|
metadata["prose"] = {
|
|
"model_hint": model_hint,
|
|
"routed_model": model_hint or self.router.route(ModelCapability.STORY_PROSE),
|
|
"prompt_version_id": prompt_version_id,
|
|
"prompt_sha256": text_sha256(prompt),
|
|
"attempts": result.attempts,
|
|
"resumed": result.resumed,
|
|
"word_count": word_count(prose),
|
|
**self._book_state_metadata(scene),
|
|
}
|
|
scene.generation_metadata = metadata
|
|
scene.save()
|
|
return scene
|
|
|
|
def review(self, scene: StandaloneScene, *, model_hint: str | None = None) -> StandaloneScene:
|
|
if scene.status != SceneDraftStatus.DRAFT_REVIEW or not scene.prose:
|
|
raise ValueError("scene has no completed draft to review")
|
|
prose = self._scene_document(scene.title, scene.prose)
|
|
if prose != scene.prose:
|
|
_partial_path, artifact_path, _review_path = self._artifact_paths(scene)
|
|
atomic_write_text(artifact_path, prose + "\n")
|
|
scene.prose = prose
|
|
scene.review = {}
|
|
scene.save(
|
|
update_fields=[
|
|
"prose",
|
|
"word_count",
|
|
"sha256",
|
|
"review",
|
|
"updated_at",
|
|
]
|
|
)
|
|
prompt, prompt_version_id = render_authoring_prompt(
|
|
"STANDALONE_SCENE_REVIEW",
|
|
STANDALONE_SCENE_REVIEW_SYSTEM,
|
|
STANDALONE_SCENE_REVIEW_TEMPLATE,
|
|
context=(scene.context_pack or {}).get("rendered_context")
|
|
or "(No approved source context.)",
|
|
plan=json.dumps(scene.plan, ensure_ascii=False, indent=2),
|
|
requirements=json.dumps(scene.contract_requirements, ensure_ascii=False, indent=2),
|
|
prose=scene.prose,
|
|
)
|
|
response = self.router.complete(
|
|
ModelRequestContract(
|
|
purpose=ModelCapability.STORY_REVIEW,
|
|
prompt=prompt,
|
|
model_hint=model_hint,
|
|
token_budget=5000,
|
|
project=scene.story.project if scene.story else None,
|
|
)
|
|
)
|
|
review = self._normalize_review(
|
|
extract_json_object(response.content), scene.contract_requirements, scene.prose
|
|
)
|
|
_partial_path, _artifact_path, review_path = self._artifact_paths(scene)
|
|
atomic_write_text(review_path, json.dumps(review, ensure_ascii=False, indent=2) + "\n")
|
|
scene.review = review
|
|
scene.review_artifact_uri = str(review_path)
|
|
metadata = dict(scene.generation_metadata or {})
|
|
metadata["review"] = {
|
|
"model": response.model,
|
|
"prompt_version_id": prompt_version_id,
|
|
"prompt_sha256": text_sha256(prompt),
|
|
"response_sha256": text_sha256(response.content),
|
|
**self._book_state_metadata(scene),
|
|
}
|
|
scene.generation_metadata = metadata
|
|
scene.save(
|
|
update_fields=[
|
|
"review",
|
|
"review_artifact_uri",
|
|
"generation_metadata",
|
|
"updated_at",
|
|
]
|
|
)
|
|
return scene
|
|
|
|
@staticmethod
|
|
def _book_chapters(content: dict[str, Any]) -> list[dict[str, Any]]:
|
|
chapters: list[dict[str, Any]] = []
|
|
|
|
def append_chapters(value: Any) -> None:
|
|
if isinstance(value, dict):
|
|
for key, chapter in value.items():
|
|
if isinstance(chapter, dict):
|
|
chapters.append({"key": str(chapter.get("key") or key), **chapter})
|
|
elif isinstance(value, list):
|
|
chapters.extend(chapter for chapter in value if isinstance(chapter, dict))
|
|
|
|
append_chapters(content.get("chapters"))
|
|
acts = content.get("acts") or []
|
|
if isinstance(acts, dict):
|
|
acts = acts.values()
|
|
for act in acts:
|
|
if isinstance(act, dict):
|
|
append_chapters(act.get("chapters"))
|
|
return chapters
|
|
|
|
@classmethod
|
|
def _book_chapter(
|
|
cls, content: dict[str, Any], chapter_key: str
|
|
) -> tuple[dict[str, Any], list[dict[str, Any]], int]:
|
|
chapters = cls._book_chapters(content)
|
|
for index, chapter in enumerate(chapters):
|
|
key = str(
|
|
chapter.get("key")
|
|
or chapter.get("chapter_key")
|
|
or chapter.get("id")
|
|
or ""
|
|
)
|
|
if key == chapter_key:
|
|
return chapter, chapters, index
|
|
raise ValueError(f"book state chapter not found: {chapter_key}")
|
|
|
|
@classmethod
|
|
def _book_planning_packet(cls, scene: StandaloneScene) -> str:
|
|
state = scene.book_state
|
|
content = state.content or {}
|
|
chapter, chapters, index = cls._book_chapter(content, scene.book_chapter_key)
|
|
prior = chapters[index - 1] if index else {}
|
|
following = chapters[index + 1] if index + 1 < len(chapters) else {}
|
|
packet = {
|
|
"book_state": {
|
|
"id": str(state.id),
|
|
"version": state.version,
|
|
"sha256": state.sha256,
|
|
},
|
|
"book_constraints": content.get("constraints")
|
|
or content.get("book_constraints")
|
|
or content.get("global_constraints")
|
|
or [],
|
|
"book_forbidden_events": content.get("forbidden_events")
|
|
or content.get("book_forbidden_events")
|
|
or [],
|
|
"current_chapter": chapter,
|
|
"prior_chapter_ending": chapter.get("prior_ending")
|
|
or prior.get("ending_state")
|
|
or prior.get("ending")
|
|
or prior.get("final_state")
|
|
or "",
|
|
"next_chapter_purpose": chapter.get("next_purpose")
|
|
or following.get("purpose")
|
|
or "",
|
|
"relevant_arcs": cls._relevant_book_items(
|
|
content.get("arcs"),
|
|
chapter.get("arcs") or chapter.get("arc_keys") or chapter.get("arc_ids"),
|
|
scene.book_chapter_key,
|
|
),
|
|
"relevant_threads": cls._relevant_book_items(
|
|
content.get("threads") or content.get("plot_threads"),
|
|
chapter.get("threads")
|
|
or chapter.get("thread_keys")
|
|
or chapter.get("thread_ids"),
|
|
scene.book_chapter_key,
|
|
),
|
|
"continuity_facts": chapter.get("continuity_facts")
|
|
or cls._relevant_continuity(content, index),
|
|
}
|
|
return "PLANNING BOOK STATE PACKET (approved constraints)\n" + json.dumps(
|
|
packet, ensure_ascii=False, indent=2
|
|
)
|
|
|
|
@staticmethod
|
|
def _relevant_book_items(items: Any, references: Any, chapter_key: str) -> list[Any]:
|
|
if isinstance(references, dict):
|
|
references = [references]
|
|
elif isinstance(references, str):
|
|
references = [references]
|
|
references = references or []
|
|
if references and all(isinstance(item, dict) for item in references):
|
|
return list(references)
|
|
reference_keys = {str(value) for value in references}
|
|
if isinstance(items, dict):
|
|
values = [
|
|
{"key": key, **value}
|
|
for key, value in items.items()
|
|
if isinstance(value, dict)
|
|
]
|
|
elif isinstance(items, list):
|
|
values = items
|
|
else:
|
|
values = []
|
|
if not values:
|
|
return list(references)
|
|
if not reference_keys:
|
|
relevant = [
|
|
item
|
|
for item in values
|
|
if chapter_key
|
|
in {
|
|
str(value)
|
|
for value in (
|
|
item.get("chapters") or item.get("chapter_keys") or []
|
|
)
|
|
}
|
|
]
|
|
return relevant or values
|
|
return [
|
|
item
|
|
for item in values
|
|
if str(
|
|
item.get("key")
|
|
or item.get("id")
|
|
or item.get("arc_id")
|
|
or item.get("thread_id")
|
|
)
|
|
in reference_keys
|
|
]
|
|
|
|
@classmethod
|
|
def _relevant_continuity(
|
|
cls, content: dict[str, Any], chapter_index: int
|
|
) -> list[dict[str, Any]]:
|
|
chapter_positions = {
|
|
str(chapter.get("chapter_key") or chapter.get("key") or ""): index
|
|
for index, chapter in enumerate(cls._book_chapters(content))
|
|
}
|
|
relevant = []
|
|
for fact in content.get("continuity") or content.get("continuity_facts") or []:
|
|
if not isinstance(fact, dict):
|
|
continue
|
|
established = chapter_positions.get(str(fact.get("established_in") or ""))
|
|
resolved = chapter_positions.get(str(fact.get("resolved_in") or ""))
|
|
if established is None or established > chapter_index:
|
|
continue
|
|
if fact.get("resolved_in") and (resolved is None or resolved < chapter_index):
|
|
continue
|
|
relevant.append(fact)
|
|
return relevant
|
|
|
|
@staticmethod
|
|
def _book_state_metadata(scene: StandaloneScene) -> dict[str, Any]:
|
|
if not scene.book_state_id:
|
|
return {}
|
|
state = scene.book_state
|
|
return {
|
|
"book_state_id": str(state.id),
|
|
"book_state_version": state.version,
|
|
"book_state_sha256": state.sha256,
|
|
"book_chapter_key": scene.book_chapter_key,
|
|
}
|
|
|
|
@staticmethod
|
|
def _scene_document(title: str, prose: str) -> str:
|
|
body = prose.strip()
|
|
first_line = body.partition("\n")[0].lstrip("# ").strip()
|
|
if first_line == title.strip():
|
|
return body
|
|
return f"# {title.strip()}\n\n{body}"
|
|
|
|
def approve(
|
|
self,
|
|
scene: StandaloneScene,
|
|
*,
|
|
actor: str,
|
|
force: bool = False,
|
|
) -> StandaloneScene:
|
|
if scene.status != SceneDraftStatus.DRAFT_REVIEW or not scene.prose:
|
|
raise ValueError("scene has no completed draft awaiting approval")
|
|
if not scene.review:
|
|
raise ValueError("scene must be reviewed before approval")
|
|
if not scene.review.get("passed") and not force:
|
|
raise ValueError("scene review has blocking issues")
|
|
artifact_path = Path(scene.artifact_uri)
|
|
root = self._artifact_root(scene)
|
|
source = register_source(
|
|
work=scene.work,
|
|
path=artifact_path,
|
|
root=root,
|
|
authority=DocumentAuthority.PROVISIONAL,
|
|
document_type=DocumentType.SCENE,
|
|
)
|
|
source_document = scene.work.source_documents.get(logical_key=source.logical_key)
|
|
source_version = source_document.versions.get(version=source.version)
|
|
scene.status = SceneDraftStatus.APPROVED
|
|
scene.approved_at = timezone.now()
|
|
scene.approved_by = actor.strip()
|
|
scene.source_version = source_version
|
|
metadata = dict(scene.generation_metadata or {})
|
|
metadata["approval"] = {
|
|
"actor": scene.approved_by,
|
|
"forced": bool(force),
|
|
"source_version_id": str(source_version.id),
|
|
}
|
|
scene.generation_metadata = metadata
|
|
scene.save(
|
|
update_fields=[
|
|
"status",
|
|
"approved_at",
|
|
"approved_by",
|
|
"source_version",
|
|
"generation_metadata",
|
|
"updated_at",
|
|
]
|
|
)
|
|
return scene
|
|
|
|
def reject(self, scene: StandaloneScene, *, actor: str) -> StandaloneScene:
|
|
if scene.status == SceneDraftStatus.APPROVED:
|
|
raise ValueError(
|
|
"approved scenes cannot be rejected in place; create a superseding revision"
|
|
)
|
|
scene.status = SceneDraftStatus.REJECTED
|
|
scene.approved_by = actor.strip()
|
|
scene.save(update_fields=["status", "approved_by", "updated_at"])
|
|
return scene
|
|
|
|
def _normalize_plan(self, data: dict[str, Any], scene: StandaloneScene) -> dict[str, Any]:
|
|
beats = []
|
|
for raw in data.get("beats") or []:
|
|
if isinstance(raw, str):
|
|
raw = {"text": raw, "required": False}
|
|
text = str(raw.get("text") or "").strip()
|
|
if text:
|
|
beats.append({"text": text, "required": bool(raw.get("required"))})
|
|
if not beats:
|
|
raise ValueError("scene plan must contain at least one beat")
|
|
plan = {
|
|
"purpose": str(data.get("purpose") or "").strip(),
|
|
"pov_character": str(data.get("pov_character") or "").strip(),
|
|
"tense": str(data.get("tense") or "past").strip(),
|
|
"location": str(data.get("location") or "").strip(),
|
|
"time_context": str(data.get("time_context") or "").strip(),
|
|
"present": list(data.get("present") or []),
|
|
"target_words": scene.target_words,
|
|
"beats": beats[:12],
|
|
"exact_values": list(data.get("exact_values") or []),
|
|
"constraints": list(
|
|
dict.fromkeys([*scene.constraints, *(data.get("constraints") or [])])
|
|
),
|
|
"forbidden_events": list(
|
|
dict.fromkeys([*scene.forbidden_events, *(data.get("forbidden_events") or [])])
|
|
),
|
|
"ending_state": str(data.get("ending_state") or "").strip(),
|
|
"final_image": str(data.get("final_image") or "").strip(),
|
|
"boundary_constraints": list(
|
|
dict.fromkeys(
|
|
[*scene.boundary_constraints, *(data.get("boundary_constraints") or [])]
|
|
)
|
|
),
|
|
"continuity_questions": list(data.get("continuity_questions") or []),
|
|
}
|
|
if not plan["ending_state"]:
|
|
raise ValueError("scene plan must define ending_state")
|
|
return plan
|
|
|
|
def _requirements(self, plan: dict[str, Any]) -> list[dict[str, Any]]:
|
|
contract = {
|
|
"scenes": [
|
|
{
|
|
"number": 1,
|
|
"beats": plan["beats"],
|
|
"ending_state": plan["ending_state"],
|
|
}
|
|
],
|
|
"exact_values": plan["exact_values"],
|
|
"forbidden_events": plan["forbidden_events"],
|
|
"chapter_constraints": plan["constraints"],
|
|
"boundary_constraints": plan["boundary_constraints"],
|
|
"final_image": plan["final_image"],
|
|
}
|
|
return build_contract_requirements(contract, max_required_per_scene=None)
|
|
|
|
def _normalize_review(
|
|
self,
|
|
data: dict[str, Any],
|
|
requirements: list[dict[str, Any]],
|
|
prose: str,
|
|
) -> dict[str, Any]:
|
|
expected = {item["id"]: item for item in requirements}
|
|
results_by_id = {
|
|
str(item.get("requirement_id") or ""): item
|
|
for item in data.get("requirement_results") or []
|
|
if isinstance(item, dict)
|
|
}
|
|
requirement_results = []
|
|
blocking_failure = False
|
|
for requirement_id, requirement in expected.items():
|
|
raw = results_by_id.get(requirement_id) or {
|
|
"requirement_id": requirement_id,
|
|
"status": "UNVERIFIABLE",
|
|
"evidence_quote": "",
|
|
"details": "Reviewer omitted this frozen requirement.",
|
|
}
|
|
status = str(raw.get("status") or "UNVERIFIABLE").upper()
|
|
evidence = str(raw.get("evidence_quote") or "")
|
|
evidence_valid = bool(evidence and evidence_is_present(prose, evidence))
|
|
if requirement.get("blocking") and status != "HIT":
|
|
blocking_failure = True
|
|
requirement_results.append(
|
|
{
|
|
"requirement_id": requirement_id,
|
|
"status": status,
|
|
"evidence_quote": evidence,
|
|
"evidence_valid": evidence_valid,
|
|
"details": str(raw.get("details") or ""),
|
|
}
|
|
)
|
|
findings = []
|
|
for raw in data.get("findings") or []:
|
|
if not isinstance(raw, dict):
|
|
continue
|
|
finding = dict(raw)
|
|
evidence = str(finding.get("evidence_quote") or "")
|
|
finding["evidence_valid"] = bool(evidence and evidence_is_present(prose, evidence))
|
|
severity = str(finding.get("severity") or "MEDIUM").upper()
|
|
finding["severity"] = severity
|
|
if severity in {"HIGH", "CRITICAL"}:
|
|
blocking_failure = True
|
|
findings.append(finding)
|
|
proposed_changes = []
|
|
for raw in data.get("proposed_changes") or []:
|
|
if not isinstance(raw, dict):
|
|
continue
|
|
change = dict(raw)
|
|
evidence = str(change.get("evidence_quote") or "")
|
|
change["evidence_valid"] = bool(evidence and evidence_is_present(prose, evidence))
|
|
proposed_changes.append(change)
|
|
return {
|
|
"schema_version": 1,
|
|
"passed": bool(data.get("passed")) and not blocking_failure,
|
|
"requirement_results": requirement_results,
|
|
"findings": findings,
|
|
"observed_state": data.get("observed_state") or {},
|
|
"proposed_changes": proposed_changes,
|
|
}
|
|
|
|
def _artifact_root(self, scene: StandaloneScene) -> Path:
|
|
if scene.story and scene.story.artifact_root.strip():
|
|
return Path(scene.story.artifact_root)
|
|
return (
|
|
Path(settings.BASE_DIR)
|
|
/ "artifacts"
|
|
/ "stories"
|
|
/ scene.work.series.slug
|
|
/ scene.work.slug
|
|
)
|
|
|
|
def _artifact_paths(self, scene: StandaloneScene) -> tuple[Path, Path, Path]:
|
|
directory = self._artifact_root(scene) / "scenes"
|
|
if scene.book_state_id:
|
|
directory = (
|
|
directory
|
|
/ f"book-state-v{scene.book_state.version:04d}-{scene.book_state_id}"
|
|
/ scene.book_chapter_key
|
|
)
|
|
directory /= scene.scene_key
|
|
stem = f"r{scene.revision:02d}"
|
|
return (
|
|
directory / f"{stem}.partial.md",
|
|
directory / f"{stem}.md",
|
|
directory / f"{stem}.review.json",
|
|
)
|