Artifex/control_plane/authoring/book_state.py

from __future__ import annotations

import hashlib
import json
import re
from copy import deepcopy
from pathlib import Path
from typing import Any

from django.conf import settings
from django.db import transaction
from django.db.models import Max
from django.utils import timezone

from control_plane.authoring.models import (
    BookRun,
    BookRunStatus,
    BookStateStatus,
    BookStateVersion,
    SceneDraftStatus,
    StandaloneScene,
    Work,
)
from control_plane.authoring.prompts import (
    BOOK_CONTINUITY_REVIEW_SYSTEM,
    BOOK_CONTINUITY_REVIEW_TEMPLATE,
    BOOK_STRUCTURE_REVIEW_SYSTEM,
    BOOK_STRUCTURE_REVIEW_TEMPLATE,
)
from control_plane.authoring.standalone_scenes import render_authoring_prompt
from control_plane.authoring.streaming import atomic_write_text
from model_router.providers import extract_json_object
from model_router.router import ModelCapability, ModelRequestContract, ModelRouter

CHAPTER_KEY_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
CONTINUITY_CATEGORIES = {
    "character",
    "reveal",
    "object",
    "injury",
    "route",
    "promise",
    "relationship",
    "location",
    "money",
    "other",
}
REVIEW_CATEGORIES = {
    "structure",
    "continuity",
    "chronology",
    "character",
    "plot",
    "pacing",
    "contract",
    "canon",
    "logic",
    "relationship",
    "location",
    "object",
    "injury",
    "route",
    "promise",
    "money",
    "other",
}
ROOT_KEYS = {
    "schema_version",
    "title",
    "premise",
    "target_words",
    "constraints",
    "forbidden_events",
    "acts",
    "arcs",
    "threads",
    "continuity",
    "chapters",
    "ending",
    "open_questions",
}


def _canonical(value: Any, *, pretty: bool = False) -> str:
    options: dict[str, Any] = {"ensure_ascii": False, "sort_keys": True}
    if pretty:
        options["indent"] = 2
    else:
        options["separators"] = (",", ":")
    return json.dumps(value, **options)


def _sha256(value: str) -> str:
    return hashlib.sha256(value.encode("utf-8")).hexdigest()


def _text(value: Any) -> str:
    return value.strip() if isinstance(value, str) else ""


def _integer(value: Any) -> int | None:
    return value if isinstance(value, int) and not isinstance(value, bool) else None


def _range_value(item: dict[str, Any], side: str) -> Any:
    return item.get(f"{side}_chapter", item.get(f"chapter_{side}"))


def validate_book_state_content(
    content: Any, *, work: Work, for_approval: bool = False
) -> dict[str, Any]:
    """Validate schema v1 without changing or normalizing the supplied document."""
    errors: list[str] = []
    warnings: list[str] = []

    def error(path: str, message: str) -> None:
        errors.append(f"{path}: {message}")

    def warning(path: str, message: str) -> None:
        warnings.append(f"{path}: {message}")

    if not isinstance(content, dict):
        return {
            "schema_version": 1,
            "valid": False,
            "errors": ["$: must be an object"],
            "warnings": [],
        }

    unknown = sorted(set(content) - ROOT_KEYS)
    for key in unknown:
        error(f"$.{key}", "unknown root property")
    if content.get("schema_version") != 1:
        error("$.schema_version", "must equal 1")
    for key in ("title", "premise"):
        if not _text(content.get(key)):
            error(f"$.{key}", "must be a nonempty string")
    target_words = _integer(content.get("target_words"))
    if target_words is None or target_words < 300:
        error("$.target_words", "must be an integer of at least 300")
    for key in ("constraints", "forbidden_events", "open_questions"):
        value = content.get(key, [])
        if not isinstance(value, list) or any(not _text(item) for item in value):
            error(f"$.{key}", "must be a list of nonempty strings")

    chapters = content.get("chapters")
    if not isinstance(chapters, list) or not chapters:
        error("$.chapters", "must be a nonempty list")
        chapters = []
    chapter_keys: list[str] = []
    seen_chapter_keys: set[str] = set()
    chapter_numbers: dict[str, int] = {}
    beat_ids: set[str] = set()
    scene_placements: set[str] = set()
    scene_lookups: list[tuple[str, int, dict[str, Any]]] = []
    for index, raw in enumerate(chapters):
        path = f"$.chapters[{index}]"
        if not isinstance(raw, dict):
            error(path, "must be an object")
            continue
        key = _text(raw.get("chapter_key"))
        if not key or not CHAPTER_KEY_RE.fullmatch(key):
            error(f"{path}.chapter_key", "must be lowercase hyphen-separated text")
        elif key in chapter_numbers:
            error(f"{path}.chapter_key", "must be unique")
        else:
            chapter_keys.append(key)
        number = _integer(raw.get("number"))
        order = _integer(raw.get("order"))
        expected = index + 1
        if number != expected:
            error(f"{path}.number", f"must equal {expected}")
        if order != expected:
            error(f"{path}.order", f"must equal {expected}")
        for field in ("title", "purpose"):
            if not _text(raw.get(field)):
                error(f"{path}.{field}", "must be a nonempty string")
        words = _integer(raw.get("target_words"))
        if words is None or not 300 <= words <= 20000:
            error(f"{path}.target_words", "must be an integer from 300 through 20000")
        dependencies = raw.get("depends_on", [])
        if not isinstance(dependencies, list):
            error(f"{path}.depends_on", "must be a list")
        else:
            seen_dependencies: set[str] = set()
            for dependency in dependencies:
                if not isinstance(dependency, str) or dependency not in seen_chapter_keys:
                    error(f"{path}.depends_on", f"{dependency!r} must name an earlier chapter")
                elif dependency in seen_dependencies:
                    error(f"{path}.depends_on", f"{dependency!r} is duplicated")
                seen_dependencies.add(str(dependency))
        if key:
            chapter_numbers[key] = number or expected
            seen_chapter_keys.add(key)
        for field in ("act_id",):
            if not _text(raw.get(field)):
                error(f"{path}.{field}", "must be a nonempty string")
        for field in ("arc_ids", "thread_ids"):
            values = raw.get(field, [])
            if not isinstance(values, list) or any(not _text(item) for item in values):
                error(f"{path}.{field}", "must be a list of nonempty IDs")
            elif len(values) != len(set(values)):
                error(f"{path}.{field}", "must not contain duplicate IDs")
        beats = raw.get("beats")
        if not isinstance(beats, list) or not beats:
            error(f"{path}.beats", "must be a nonempty list")
        else:
            for beat_index, beat in enumerate(beats):
                beat_path = f"{path}.beats[{beat_index}]"
                if not isinstance(beat, dict):
                    error(beat_path, "must be an object")
                    continue
                beat_id = _text(beat.get("beat_id"))
                if not beat_id:
                    error(f"{beat_path}.beat_id", "must be a nonempty string")
                elif beat_id in beat_ids:
                    error(f"{beat_path}.beat_id", "must be globally unique")
                beat_ids.add(beat_id)
                if not _text(beat.get("text")):
                    error(f"{beat_path}.text", "must be a nonempty string")
                if not isinstance(beat.get("required"), bool):
                    error(f"{beat_path}.required", "must be boolean")
        if not _text(raw.get("ending_state")):
            (error if for_approval else warning)(
                f"{path}.ending_state", "must be a nonempty string before approval"
            )
        refs = raw.get("scene_refs", [])
        if not isinstance(refs, list):
            error(f"{path}.scene_refs", "must be a list")
        else:
            for ref_index, ref in enumerate(refs):
                ref_path = f"{path}.scene_refs[{ref_index}]"
                if not isinstance(ref, dict):
                    error(ref_path, "must be an object")
                    continue
                scene_id = _text(ref.get("scene_id"))
                raw_placement = ref.get("placement", ref.get("order"))
                placement = f"{key}:{raw_placement}" if raw_placement is not None else scene_id
                if not scene_id:
                    error(f"{ref_path}.scene_id", "must be a nonempty string")
                    continue
                if scene_id in scene_placements or placement in scene_placements:
                    error(f"{ref_path}.placement", "must be unique across the book")
                scene_placements.add(scene_id)
                scene_placements.add(placement)
                scene_lookups.append((ref_path, index, ref))

    chapter_word_total = sum(
        _integer(chapter.get("target_words")) or 0
        for chapter in chapters
        if isinstance(chapter, dict)
    )
    if target_words is not None and target_words != chapter_word_total:
        error("$.target_words", "must equal the sum of chapter target_words")
    if not _text(content.get("ending")):
        error("$.ending", "must be a nonempty string")

    chapter_key_set = set(chapter_keys)
    acts = content.get("acts")
    if not isinstance(acts, list) or not acts:
        error("$.acts", "must be a nonempty list")
        acts = []
    act_ids: set[str] = set()
    act_ranges: dict[str, tuple[int, int]] = {}
    previous_end = 0
    for index, raw in enumerate(acts):
        path = f"$.acts[{index}]"
        if not isinstance(raw, dict):
            error(path, "must be an object")
            continue
        act_id = _text(raw.get("act_id"))
        if not act_id:
            error(f"{path}.act_id", "must be a nonempty string")
        elif act_id in act_ids:
            error(f"{path}.act_id", "must be unique")
        act_ids.add(act_id)
        if _integer(raw.get("number")) != index + 1:
            error(f"{path}.number", f"must equal {index + 1}")
        start = _integer(_range_value(raw, "start"))
        end = _integer(_range_value(raw, "end"))
        if start is None or end is None or start > end:
            error(path, "must define a valid inclusive chapter range")
        else:
            if start != previous_end + 1:
                error(path, f"range must start at chapter {previous_end + 1}")
            previous_end = end
            if act_id:
                act_ranges[act_id] = (start, end)
    if chapters and previous_end != len(chapters):
        error("$.acts", "ranges must cover every chapter exactly once")

    id_sets: dict[str, set[str]] = {"arc_ids": set(), "thread_ids": set()}
    item_ranges: dict[str, dict[str, tuple[int, int]]] = {
        "arc_ids": {},
        "thread_ids": {},
    }
    for collection, id_name, chapter_field in (
        ("arcs", "arc_id", "arc_ids"),
        ("threads", "thread_id", "thread_ids"),
    ):
        values = content.get(collection)
        if not isinstance(values, list):
            error(f"$.{collection}", "must be a list")
            values = []
        for index, raw in enumerate(values):
            path = f"$.{collection}[{index}]"
            if not isinstance(raw, dict):
                error(path, "must be an object")
                continue
            item_id = _text(raw.get(id_name))
            if not item_id:
                error(f"{path}.{id_name}", "must be a nonempty string")
            elif item_id in id_sets[chapter_field]:
                error(f"{path}.{id_name}", "must be unique")
            id_sets[chapter_field].add(item_id)
            start_value = _range_value(raw, "start")
            end_value = _range_value(raw, "end")
            start = chapter_numbers.get(str(start_value), _integer(start_value))
            end = chapter_numbers.get(str(end_value), _integer(end_value))
            if (
                start is None
                or end is None
                or not 1 <= start <= len(chapters)
                or not 1 <= end <= len(chapters)
            ):
                error(path, "chapter range must reference existing chapters")
            elif start > end:
                error(path, "chapter range cannot end before it starts")
            elif item_id:
                item_ranges[chapter_field][item_id] = (start, end)

    for index, raw in enumerate(chapters):
        if not isinstance(raw, dict):
            continue
        path = f"$.chapters[{index}]"
        act_id = raw.get("act_id")
        if act_id not in act_ids:
            error(f"{path}.act_id", "must reference an existing act")
        elif act_id in act_ranges and not (
            act_ranges[act_id][0] <= index + 1 <= act_ranges[act_id][1]
        ):
            error(f"{path}.act_id", "does not match the act range containing this chapter")
        for field in ("arc_ids", "thread_ids"):
            values = raw.get(field, [])
            if isinstance(values, list):
                for item_id in values:
                    if item_id not in id_sets[field]:
                        error(f"{path}.{field}", f"{item_id!r} does not exist")
                    elif item_id in item_ranges[field] and not (
                        item_ranges[field][item_id][0]
                        <= index + 1
                        <= item_ranges[field][item_id][1]
                    ):
                        error(
                            f"{path}.{field}",
                            f"{item_id!r} is assigned outside its chapter range",
                        )

    continuity = content.get("continuity")
    if not isinstance(continuity, list):
        error("$.continuity", "must be a list")
        continuity = []
    fact_ids: set[str] = set()
    for index, raw in enumerate(continuity):
        path = f"$.continuity[{index}]"
        if not isinstance(raw, dict):
            error(path, "must be an object")
            continue
        fact_id = _text(raw.get("fact_id"))
        if not fact_id:
            error(f"{path}.fact_id", "must be a nonempty string")
        elif fact_id in fact_ids:
            error(f"{path}.fact_id", "must be unique")
        fact_ids.add(fact_id)
        if not _text(raw.get("entity_key")):
            error(f"{path}.entity_key", "must be a nonempty string")
        if raw.get("category") not in CONTINUITY_CATEGORIES:
            error(f"{path}.category", f"must be one of {', '.join(sorted(CONTINUITY_CATEGORIES))}")
        if not _text(raw.get("description")):
            error(f"{path}.description", "must be a nonempty string")
        established = raw.get("established_in")
        resolved = raw.get("resolved_in")
        if established not in chapter_key_set:
            error(f"{path}.established_in", "must reference an existing chapter")
        if resolved is not None:
            if resolved not in chapter_key_set:
                error(f"{path}.resolved_in", "must reference an existing chapter")
            elif (
                established in chapter_numbers
                and chapter_numbers[str(resolved)] < chapter_numbers[str(established)]
            ):
                error(f"{path}.resolved_in", "cannot precede establishment")

    for path, _chapter_index, ref in scene_lookups:
        try:
            scene = StandaloneScene.objects.get(pk=ref["scene_id"])
        except (StandaloneScene.DoesNotExist, ValueError, TypeError):
            error(f"{path}.scene_id", "does not resolve to a standalone scene")
            continue
        if scene.work_id != work.pk:
            error(f"{path}.scene_id", "scene belongs to a different work")
        if scene.book_state_id:
            error(f"{path}.scene_id", "scene_refs must reference unbound assembly scenes")
        revision = _integer(ref.get("revision"))
        if revision is None or scene.revision != revision:
            error(f"{path}.revision", "does not match the scene revision")
        expected_hash = _text(ref.get("sha256"))
        if not expected_hash or scene.sha256 != expected_hash:
            error(f"{path}.sha256", "does not match the scene content hash")
        if for_approval and scene.status != SceneDraftStatus.APPROVED:
            error(f"{path}.scene_id", "scene must be approved")

    return {
        "schema_version": 1,
        "valid": not errors,
        "errors": sorted(set(errors)),
        "warnings": sorted(set(warnings)),
    }


def _field_names(instance_or_model: Any) -> set[str]:
    model = instance_or_model if isinstance(instance_or_model, type) else type(instance_or_model)
    return {name for field in model._meta.concrete_fields for name in (field.name, field.attname)}


def _value(instance: Any, *names: str, default: Any = None) -> Any:
    fields = _field_names(instance)
    for name in names:
        if name in fields:
            return getattr(instance, name)
    return default


def _assign(instance: Any, value: Any, *names: str) -> str | None:
    fields = _field_names(instance)
    for name in names:
        if name in fields:
            setattr(instance, name, value)
            return name
    return None


def _status(enum: Any, name: str, fallback: str) -> str:
    return str(getattr(enum, name, fallback))


class BookStateService:
    def __init__(self, router: ModelRouter) -> None:
        self.router = router

    @transaction.atomic
    def create(
        self,
        *,
        work: Work,
        content: dict[str, Any],
        actor: str = "",
        context_pack: dict[str, Any] | None = None,
        generation_metadata: dict[str, Any] | None = None,
    ) -> BookStateVersion:
        self._validate_metadata(context_pack, "context_pack")
        self._validate_metadata(generation_metadata, "generation_metadata")
        locked_work = Work.objects.select_for_update().get(pk=work.pk)
        version = (
            BookStateVersion.objects.filter(work=locked_work).aggregate(value=Max("version"))[
                "value"
            ]
            or 0
        ) + 1
        validation = validate_book_state_content(content, work=locked_work)
        if not validation["valid"]:
            raise ValueError("invalid book state: " + "; ".join(validation["errors"]))
        summary = {
            "type": "create",
            "chapter_count": len(content["chapters"]),
            "changed_from_chapter": 1,
        }
        state = self._new_state(
            locked_work,
            version,
            content,
            None,
            summary,
            validation,
            actor,
            context_pack,
            generation_metadata,
        )
        self._write_artifacts(state)
        return state

    @transaction.atomic
    def revise(
        self,
        state: BookStateVersion,
        *,
        content: dict[str, Any],
        actor: str = "",
        context_pack: dict[str, Any] | None = None,
        generation_metadata: dict[str, Any] | None = None,
    ) -> BookStateVersion:
        self._validate_metadata(context_pack, "context_pack")
        self._validate_metadata(generation_metadata, "generation_metadata")
        locked_work = Work.objects.select_for_update().get(pk=state.work_id)
        parent = (
            BookStateVersion.objects.select_for_update().select_related("work").get(pk=state.pk)
        )
        version = (
            BookStateVersion.objects.filter(work=locked_work).aggregate(value=Max("version"))[
                "value"
            ]
            or 0
        ) + 1
        validation = validate_book_state_content(content, work=locked_work)
        if not validation["valid"]:
            raise ValueError("invalid book state: " + "; ".join(validation["errors"]))
        summary = self._change_summary(_value(parent, "content", default={}), content)
        child = self._new_state(
            locked_work,
            version,
            content,
            parent,
            summary,
            validation,
            actor,
            context_pack,
            generation_metadata,
        )
        self._write_artifacts(child)
        return child

    def validate(self, state: BookStateVersion, *, for_approval: bool = False) -> dict[str, Any]:
        result = validate_book_state_content(
            _value(state, "content", default={}), work=state.work, for_approval=for_approval
        )
        field = _assign(state, result, "validation", "validation_result")
        if field:
            state.save(update_fields=[field, "updated_at"])
        return result

    def review(
        self,
        state: BookStateVersion,
        level: str | None = None,
        *,
        review_level: str | None = None,
        model_hint: str | None = None,
    ) -> dict[str, Any]:
        level = _text(level or review_level)
        content = _value(state, "content", default={})
        if _value(state, "status") not in {
            _status(BookStateStatus, "DRAFT", "draft"),
            _status(BookStateStatus, "REVIEW", "review"),
        }:
            raise ValueError("only draft or review book states may be reviewed")
        valid_levels = {
            "manuscript",
            "continuity",
            *{f"act:{item.get('act_id')}" for item in content.get("acts", [])},
        }
        if level not in valid_levels:
            raise ValueError(f"invalid review level: {level}")
        if level == "continuity":
            scene_packets = self._scene_packets(state)
            prompt, prompt_version_id = render_authoring_prompt(
                "BOOK_CONTINUITY_REVIEW",
                BOOK_CONTINUITY_REVIEW_SYSTEM,
                BOOK_CONTINUITY_REVIEW_TEMPLATE,
                state=_canonical(self._compact_state(content), pretty=True),
                scene_packets=_canonical(scene_packets, pretty=True),
            )
            purpose = ModelCapability.STORY_CONTINUITY
        else:
            review_state = (
                content
                if level == "manuscript"
                else self._act_slice(content, level.partition(":")[2])
            )
            prompt, prompt_version_id = render_authoring_prompt(
                "BOOK_STRUCTURE_REVIEW",
                BOOK_STRUCTURE_REVIEW_SYSTEM,
                BOOK_STRUCTURE_REVIEW_TEMPLATE,
                review_level=level,
                state=_canonical(self._compact_state(review_state), pretty=True),
            )
            purpose = ModelCapability.STORY_REVIEW
        story = getattr(state.work, "story_project", None)
        response = self.router.complete(
            ModelRequestContract(
                purpose=purpose,
                prompt=prompt,
                model_hint=model_hint,
                token_budget=6000,
                project=story.project if story else None,
            )
        )
        review = self._normalize_review(extract_json_object(response.content), content)
        review.update(
            {
                "level": level,
                "state_sha256": self._state_hash(state),
                "model": response.model,
                "prompt_version_id": prompt_version_id,
                "prompt_sha256": _sha256(prompt),
                "response_sha256": _sha256(response.content),
                "reviewed_at": timezone.now().isoformat(),
            }
        )
        with transaction.atomic():
            locked = BookStateVersion.objects.select_for_update().get(pk=state.pk)
            if _value(locked, "status") not in {
                _status(BookStateStatus, "DRAFT", "draft"),
                _status(BookStateStatus, "REVIEW", "review"),
            }:
                raise ValueError("book state changed while review was running")
            reviews = deepcopy(_value(locked, "reviews", "review", default={}) or {})
            reviews[level] = review
            fields = []
            field = _assign(locked, reviews, "reviews", "review")
            if field:
                fields.append(field)
            field = _assign(locked, _status(BookStateStatus, "REVIEW", "review"), "status")
            if field:
                fields.append(field)
            if not fields:
                raise ValueError("BookStateVersion must provide a reviews or review JSON field")
            locked.save(update_fields=[*fields, "updated_at"])
        self._write_artifacts(locked)
        return review

    @transaction.atomic
    def approve(
        self,
        state: BookStateVersion,
        *,
        actor: str = "",
        force: bool = False,
        notes: str = "",
    ) -> BookStateVersion:
        work = Work.objects.select_for_update().get(pk=state.work_id)
        locked = (
            BookStateVersion.objects.select_for_update().select_related("work").get(pk=state.pk)
        )
        if _value(locked, "status") != _status(BookStateStatus, "REVIEW", "review"):
            raise ValueError("book state must be in review before approval")
        validation = validate_book_state_content(
            _value(locked, "content", default={}), work=work, for_approval=True
        )
        if not validation["valid"]:
            raise ValueError("book state is not approval-valid: " + "; ".join(validation["errors"]))
        current_id = _value(work, "current_book_state_id")
        approved_base_id = self._approved_base_id(locked)
        if approved_base_id != current_id:
            raise ValueError("book state parent is stale relative to the work's approved state")
        if force and not notes.strip():
            raise ValueError("forced approval requires notes")
        content = _value(locked, "content", default={})
        required = ["manuscript", *[f"act:{act['act_id']}" for act in content.get("acts", [])]]
        if any(chapter.get("scene_refs") for chapter in content.get("chapters", [])):
            required.append("continuity")
        reviews = _value(locked, "reviews", "review", default={}) or {}
        state_hash = self._state_hash(locked)
        for level in required:
            review = reviews.get(level) or {}
            if (not review.get("passed") or review.get("state_sha256") != state_hash) and not force:
                raise ValueError(f"passing current {level} review is required")
        fields = []
        for value, names in (
            (_status(BookStateStatus, "APPROVED", "approved"), ("status",)),
            (timezone.now(), ("approved_at",)),
            (actor.strip(), ("approved_by",)),
            (notes.strip(), ("approval_notes",)),
            (force, ("approval_forced",)),
            (validation, ("validation", "validation_result")),
        ):
            field = _assign(locked, value, *names)
            if field:
                fields.append(field)
        locked.save(update_fields=[*fields, "updated_at"])
        current_field = _assign(work, locked, "current_book_state")
        if not current_field:
            raise ValueError("Work must provide current_book_state")
        work.save(update_fields=[current_field, "updated_at"])
        self._write_artifacts(locked)
        return locked

    @staticmethod
    def _approved_base_id(state: BookStateVersion) -> Any:
        ancestor_id = state.parent_id
        while ancestor_id:
            ancestor = BookStateVersion.objects.select_for_update().only(
                "id", "parent_id", "status"
            ).get(pk=ancestor_id)
            if ancestor.status == BookStateStatus.APPROVED:
                return ancestor.id
            ancestor_id = ancestor.parent_id
        return None

    @transaction.atomic
    def reject(
        self, state: BookStateVersion, *, actor: str = "", notes: str = ""
    ) -> BookStateVersion:
        locked = BookStateVersion.objects.select_for_update().get(pk=state.pk)
        if _value(locked, "status") not in {
            _status(BookStateStatus, "DRAFT", "draft"),
            _status(BookStateStatus, "REVIEW", "review"),
        }:
            raise ValueError("only draft or review book states may be rejected")
        fields = []
        for value, names in (
            (_status(BookStateStatus, "REJECTED", "rejected"), ("status",)),
            (timezone.now(), ("rejected_at",)),
            (actor.strip(), ("rejected_by",)),
            (notes.strip(), ("rejection_notes",)),
        ):
            field = _assign(locked, value, *names)
            if field:
                fields.append(field)
        locked.save(update_fields=[*fields, "updated_at"])
        self._write_artifacts(locked)
        return locked

    def impact(self, state: BookStateVersion) -> dict[str, Any]:
        parent = _value(state, "parent")
        if parent is None:
            chapter_keys = [
                item.get("chapter_key")
                for item in _value(state, "content", default={}).get("chapters", [])
            ]
            return {
                "schema_version": 1,
                "from_chapter": 1 if chapter_keys else None,
                "replan_required": chapter_keys,
                "continuity_review_required": chapter_keys[1:],
                "reasons": ["initial_state"],
            }
        return self._impact_content(
            _value(parent, "content", default={}), _value(state, "content", default={})
        )

    @transaction.atomic
    def start_run(
        self, state: BookStateVersion, *, policy: dict[str, Any] | None = None
    ) -> BookRun:
        self._validate_metadata(policy, "policy")
        locked = BookStateVersion.objects.select_for_update().get(pk=state.pk)
        if _value(locked, "status") != _status(BookStateStatus, "APPROVED", "approved"):
            raise ValueError("book runs require an approved book state")
        chapters = _value(locked, "content", default={}).get("chapters", [])
        progress = {item["chapter_key"]: "needs_scene" for item in chapters}
        run_policy = deepcopy(policy or {})
        run_policy["required_reviews"] = self._required_reviews(
            _value(locked, "content", default={})
        )
        run_policy["run_required_reviews"] = ["continuity"]
        values: dict[str, Any] = {}
        fields = _field_names(BookRun)
        for names, value in (
            (("state", "book_state"), locked),
            (("work",), locked.work),
            (("status",), _status(BookRunStatus, "RUNNING", "running")),
            (("progress",), progress),
            (
                ("current_chapter_key", "cursor_chapter_key", "cursor"),
                chapters[0]["chapter_key"] if chapters else "",
            ),
            (
                ("policy",),
                run_policy,
            ),
            (("started_at",), timezone.now()),
        ):
            for name in names:
                if name in fields:
                    values[name] = value
                    break
        return BookRun.objects.create(**values)

    @transaction.atomic
    def sync_run(self, run: BookRun) -> BookRun:
        run_scope = BookRun.objects.select_related("book_state").get(pk=run.pk)
        Work.objects.select_for_update().get(pk=run_scope.book_state.work_id)
        locked = BookRun.objects.select_for_update().select_related("book_state").get(pk=run.pk)
        if locked.status == BookRunStatus.COMPLETE:
            return locked
        if locked.status in {
            BookRunStatus.FAILED,
            BookRunStatus.CANCELLED,
        }:
            raise ValueError(f"cannot synchronize a {locked.status} book run")
        state = _value(locked, "state", "book_state")
        chapters = _value(state, "content", default={}).get("chapters", [])
        progress: dict[str, str] = {}
        for chapter in chapters:
            key = chapter["chapter_key"]
            progress[key] = self._chapter_progress(state, chapter)
        first_incomplete = next(
            (key for key, status in progress.items() if status != "complete"), ""
        )
        if first_incomplete:
            status = _status(BookRunStatus, "PAUSED", "paused")
        elif self._run_reviews_pass(locked, state):
            status = _status(BookRunStatus, "COMPLETE", "complete")
        else:
            status = _status(BookRunStatus, "REVIEW", "review")
        fields = []
        finished_at = (
            timezone.now() if status == _status(BookRunStatus, "COMPLETE", "complete") else None
        )
        for value, names in (
            (progress, ("progress",)),
            (first_incomplete, ("current_chapter_key", "cursor_chapter_key", "cursor")),
            (status, ("status",)),
            (finished_at, ("finished_at",)),
        ):
            field = _assign(locked, value, *names)
            if field:
                fields.append(field)
        locked.save(update_fields=[*fields, "updated_at"])
        return locked

    def review_run(self, run: BookRun, *, model_hint: str | None = None) -> dict[str, Any]:
        run = BookRun.objects.select_related("book_state__work").get(pk=run.pk)
        if run.status != BookRunStatus.REVIEW:
            raise ValueError("book run must have all chapters complete before continuity review")
        state = run.book_state
        content = state.content or {}
        scene_packets = self._run_scene_packets(state)
        scene_set_sha256 = _sha256(_canonical(scene_packets))
        prompt, prompt_version_id = render_authoring_prompt(
            "BOOK_CONTINUITY_REVIEW",
            BOOK_CONTINUITY_REVIEW_SYSTEM,
            BOOK_CONTINUITY_REVIEW_TEMPLATE,
            state=_canonical(self._compact_state(content), pretty=True),
            scene_packets=_canonical(scene_packets, pretty=True),
        )
        story = getattr(state.work, "story_project", None)
        response = self.router.complete(
            ModelRequestContract(
                purpose=ModelCapability.STORY_CONTINUITY,
                prompt=prompt,
                model_hint=model_hint,
                token_budget=6000,
                project=story.project if story else None,
            )
        )
        review = self._normalize_review(extract_json_object(response.content), content)
        review.update(
            {
                "level": "continuity",
                "state_sha256": self._state_hash(state),
                "scene_set_sha256": scene_set_sha256,
                "scene_manifest": self._scene_manifest(scene_packets),
                "model": response.model,
                "prompt_version_id": prompt_version_id,
                "prompt_sha256": _sha256(prompt),
                "response_sha256": _sha256(response.content),
                "reviewed_at": timezone.now().isoformat(),
            }
        )
        with transaction.atomic():
            locked = BookRun.objects.select_for_update().select_related("book_state").get(pk=run.pk)
            if locked.status != BookRunStatus.REVIEW:
                raise ValueError("book run changed while continuity review was running")
            current_packets = self._run_scene_packets(locked.book_state)
            if _sha256(_canonical(current_packets)) != scene_set_sha256:
                raise ValueError("book run scenes changed while continuity review was running")
            reviews = deepcopy(locked.reviews or {})
            reviews["continuity"] = review
            locked.reviews = reviews
            locked.save(update_fields=["reviews", "updated_at"])
        self.sync_run(locked)
        return review

    def _new_state(
        self,
        work: Work,
        version: int,
        content: dict[str, Any],
        parent: BookStateVersion | None,
        summary: dict[str, Any],
        validation: dict[str, Any],
        actor: str,
        context_pack: dict[str, Any] | None,
        generation_metadata: dict[str, Any] | None,
    ) -> BookStateVersion:
        fields = _field_names(BookStateVersion)
        values: dict[str, Any] = {"work": work, "version": version, "content": deepcopy(content)}
        candidates = (
            ("parent", parent),
            ("status", _status(BookStateStatus, "DRAFT", "draft")),
            ("change_summary", summary),
            ("validation", validation),
            ("validation_result", validation),
            ("sha256", _sha256(_canonical(content))),
            ("created_by", actor.strip()),
            ("context_pack", deepcopy(context_pack or {})),
            ("generation_metadata", deepcopy(generation_metadata or {})),
        )
        for name, value in candidates:
            if name in fields and name not in values:
                values[name] = value
        return BookStateVersion.objects.create(**values)

    @staticmethod
    def _validate_metadata(value: Any, name: str) -> None:
        if value is not None and not isinstance(value, dict):
            raise ValueError(f"{name} must be an object")

    @staticmethod
    def _change_summary(parent: dict[str, Any], child: dict[str, Any]) -> dict[str, Any]:
        changed = sorted(key for key in ROOT_KEYS if parent.get(key) != child.get(key))
        impact = BookStateService._impact_content(parent, child)
        return {
            "type": "revision",
            "changed_sections": changed,
            "changed_from_chapter": impact["from_chapter"],
            "replan_required": impact["replan_required"],
            "continuity_review_required": impact["continuity_review_required"],
        }

    @staticmethod
    def _impact_content(parent: dict[str, Any], child: dict[str, Any]) -> dict[str, Any]:
        old_chapters = parent.get("chapters", [])
        new_chapters = child.get("chapters", [])
        new_keys = [item.get("chapter_key") for item in new_chapters]
        reasons: list[str] = []
        earliest: int | None = None
        global_fields = (
            "premise",
            "target_words",
            "constraints",
            "forbidden_events",
            "acts",
            "arcs",
            "threads",
            "continuity",
            "ending",
        )
        for field in global_fields:
            if parent.get(field) != child.get(field):
                earliest = 1
                reasons.append(field)
        old_keys = [item.get("chapter_key") for item in old_chapters]
        directly_changed: list[str] = []
        limit = max(len(old_chapters), len(new_chapters))
        for index in range(limit):
            old = old_chapters[index] if index < len(old_chapters) else None
            new = new_chapters[index] if index < len(new_chapters) else None
            if old != new:
                candidate = index + 1
                earliest = candidate if earliest is None else min(earliest, candidate)
                if old is None:
                    reasons.append("chapter_inserted")
                elif new is None:
                    reasons.append("chapter_deleted")
                elif old.get("chapter_key") != new.get("chapter_key"):
                    reasons.append("chapter_reordered")
                else:
                    reasons.append(f"chapter_contract:{new.get('chapter_key')}")
                    directly_changed.append(new.get("chapter_key"))
        if (
            old_keys != new_keys
            and "chapter_reordered" not in reasons
            and len(old_keys) == len(new_keys)
        ):
            reasons.append("chapter_reordered")
        structural = any(
            reason in {"chapter_inserted", "chapter_deleted", "chapter_reordered"}
            for reason in reasons
        )
        global_change = any(field in reasons for field in global_fields)
        if global_change or structural:
            replan = new_keys[earliest - 1 :] if earliest else []
        else:
            affected = set(directly_changed)
            changed = True
            while changed:
                changed = False
                for chapter in new_chapters:
                    key = chapter.get("chapter_key")
                    if key not in affected and affected.intersection(chapter.get("depends_on", [])):
                        affected.add(key)
                        changed = True
            replan = [key for key in new_keys if key in affected]
        continuity = new_keys[earliest:] if earliest else []
        return {
            "schema_version": 1,
            "from_chapter": earliest,
            "replan_required": replan,
            "continuity_review_required": continuity,
            "reasons": sorted(set(reasons)),
        }

    @staticmethod
    def _compact_state(content: dict[str, Any]) -> dict[str, Any]:
        return deepcopy(content)

    @staticmethod
    def _act_slice(content: dict[str, Any], act_id: str) -> dict[str, Any]:
        act = next(item for item in content.get("acts", []) if item.get("act_id") == act_id)
        start = int(_range_value(act, "start"))
        end = int(_range_value(act, "end"))
        sliced = {key: deepcopy(value) for key, value in content.items() if key != "chapters"}
        sliced["acts"] = [deepcopy(act)]
        sliced["chapters"] = deepcopy(content.get("chapters", [])[start - 1 : end])
        keys = {item["chapter_key"] for item in sliced["chapters"]}
        sliced["continuity"] = [
            deepcopy(item)
            for item in content.get("continuity", [])
            if item.get("established_in") in keys or item.get("resolved_in") in keys
        ]
        return sliced

    @staticmethod
    def _normalize_review(data: dict[str, Any], content: dict[str, Any]) -> dict[str, Any]:
        raw_findings = data.get("findings")
        if not isinstance(raw_findings, list):
            raise ValueError("review response must contain a findings list")
        chapter_keys = {item.get("chapter_key") for item in content.get("chapters", [])}
        findings = []
        for raw in raw_findings:
            if not isinstance(raw, dict):
                raise ValueError("each review finding must be an object")
            severity = str(raw.get("severity") or "").upper()
            if severity not in {"INFO", "LOW", "MEDIUM", "HIGH", "CRITICAL"}:
                raise ValueError("each review finding must have a valid severity")
            category = str(raw.get("category") or "").lower()
            if category not in REVIEW_CATEGORIES:
                raise ValueError("each review finding must have a valid category")
            chapter_key = _text(raw.get("chapter_key"))
            if chapter_key not in chapter_keys:
                chapter_key = ""
            description = _text(raw.get("description"))
            if not description:
                raise ValueError("each review finding must have a description")
            findings.append(
                {
                    "severity": severity,
                    "category": category,
                    "chapter_key": chapter_key,
                    "description": description,
                    "suggested_revision": _text(raw.get("suggested_revision")),
                }
            )
        findings.sort(
            key=lambda item: (
                item["chapter_key"],
                item["severity"],
                item["category"],
                item["description"],
            )
        )
        blocking = sum(item["severity"] in {"HIGH", "CRITICAL"} for item in findings)
        return {
            "schema_version": 1,
            "passed": blocking == 0,
            "blocking_count": blocking,
            "findings": findings,
        }

    def _scene_packets(self, state: BookStateVersion) -> list[dict[str, Any]]:
        packets = []
        for chapter in _value(state, "content", default={}).get("chapters", []):
            for placement, ref in enumerate(chapter.get("scene_refs", []), start=1):
                scene = StandaloneScene.objects.get(pk=ref["scene_id"], work=state.work)
                packets.append(
                    {
                        "chapter_key": chapter["chapter_key"],
                        "placement": ref.get("placement", placement),
                        "scene_key": scene.scene_key,
                        "revision": scene.revision,
                        "sha256": scene.sha256,
                        "title": scene.title,
                        "prose": scene.prose,
                    }
                )
        return packets

    @staticmethod
    def _referenced_scenes(
        state: BookStateVersion, chapter: dict[str, Any]
    ) -> list[StandaloneScene]:
        references = chapter.get("scene_refs") or []
        if not references:
            return []
        scenes = {
            str(scene.id): scene
            for scene in StandaloneScene.objects.filter(
                work=state.work,
                id__in=[reference["scene_id"] for reference in references],
            )
        }
        missing = [
            str(reference["scene_id"])
            for reference in references
            if str(reference["scene_id"]) not in scenes
        ]
        if missing:
            raise ValueError("referenced scenes no longer exist: " + ", ".join(missing))
        return [scenes[str(reference["scene_id"])] for reference in references]

    @staticmethod
    def _latest_bound_scenes(state: BookStateVersion, chapter_key: str) -> list[StandaloneScene]:
        scenes = StandaloneScene.objects.filter(
            work=state.work,
            book_state=state,
            book_chapter_key=chapter_key,
        ).order_by("scene_key", "-revision", "-created_at")
        latest: dict[str, StandaloneScene] = {}
        for scene in scenes:
            latest.setdefault(scene.scene_key, scene)
        return list(latest.values())

    def _chapter_progress(self, state: BookStateVersion, chapter: dict[str, Any]) -> str:
        scenes = self._referenced_scenes(state, chapter)
        if not scenes:
            scenes = self._latest_bound_scenes(state, chapter["chapter_key"])
        if not scenes:
            return "needs_scene"
        if len(scenes) > 1 and not chapter.get("scene_refs"):
            return "ambiguous_scenes"
        if all(scene.status == SceneDraftStatus.APPROVED for scene in scenes):
            return "complete"
        statuses = sorted({str(scene.status) for scene in scenes})
        return statuses[0] if len(statuses) == 1 else "mixed_scene_status"

    def _run_scene_packets(self, state: BookStateVersion) -> list[dict[str, Any]]:
        packets = []
        for chapter in (state.content or {}).get("chapters", []):
            scenes = self._referenced_scenes(state, chapter)
            if not scenes:
                scenes = self._latest_bound_scenes(state, chapter["chapter_key"])
            for placement, scene in enumerate(scenes, start=1):
                packets.append(
                    {
                        "chapter_key": chapter["chapter_key"],
                        "placement": placement,
                        "scene_id": str(scene.id),
                        "scene_key": scene.scene_key,
                        "revision": scene.revision,
                        "sha256": scene.sha256,
                        "title": scene.title,
                        "prose": scene.prose,
                    }
                )
        return packets

    def _run_reviews_pass(self, run: BookRun, state: BookStateVersion) -> bool:
        review = (run.reviews or {}).get("continuity") or {}
        scene_set_sha256 = _sha256(_canonical(self._run_scene_packets(state)))
        return bool(
            review.get("passed")
            and review.get("state_sha256") == self._state_hash(state)
            and review.get("scene_set_sha256") == scene_set_sha256
        )

    @staticmethod
    def _scene_manifest(scene_packets: list[dict[str, Any]]) -> list[dict[str, Any]]:
        return [
            {
                "chapter_key": packet["chapter_key"],
                "placement": packet["placement"],
                "scene_id": packet["scene_id"],
                "scene_key": packet["scene_key"],
                "revision": packet["revision"],
                "sha256": packet["sha256"],
            }
            for packet in scene_packets
        ]

    @staticmethod
    def _required_reviews(content: dict[str, Any]) -> list[str]:
        levels = [*[f"act:{item['act_id']}" for item in content.get("acts", [])], "manuscript"]
        if any(item.get("scene_refs") for item in content.get("chapters", [])):
            levels.append("continuity")
        return levels

    def _reviews_pass(self, state: BookStateVersion, levels: list[str]) -> bool:
        reviews = _value(state, "reviews", "review", default={}) or {}
        state_hash = self._state_hash(state)
        return all(
            reviews.get(level, {}).get("passed")
            and reviews[level].get("state_sha256") == state_hash
            for level in levels
        )

    @staticmethod
    def _state_hash(state: BookStateVersion) -> str:
        return _value(state, "sha256", default="") or _sha256(
            _canonical(_value(state, "content", default={}))
        )

    def _artifact_root(self, state: BookStateVersion) -> Path:
        story = getattr(state.work, "story_project", None)
        if story and story.artifact_root.strip():
            return Path(story.artifact_root) / "book-states"
        return (
            Path(settings.BASE_DIR)
            / "artifacts"
            / "stories"
            / state.work.series.slug
            / state.work.slug
            / "book-states"
        )

    def _write_artifacts(self, state: BookStateVersion) -> None:
        stem = f"v{state.version:04d}"
        content = _value(state, "content", default={})
        payload = {
            "schema_version": 1,
            "work_id": str(state.work_id),
            "version": state.version,
            "status": str(_value(state, "status", default="")),
            "sha256": self._state_hash(state),
            "parent_id": str(_value(state, "parent_id", default="") or ""),
            "change_summary": _value(state, "change_summary", default={}) or {},
            "context_pack": _value(state, "context_pack", default={}) or {},
            "context_pack_sha256": _value(state, "context_pack_sha256", default=""),
            "generation_metadata": _value(state, "generation_metadata", default={}) or {},
            "created_by": _value(state, "created_by", default=""),
            "validation": _value(state, "validation", "validation_result", default={}) or {},
            "reviews": _value(state, "reviews", "review", default={}) or {},
            "approved_at": str(_value(state, "approved_at", default="") or ""),
            "approved_by": _value(state, "approved_by", default=""),
            "approval_notes": _value(state, "approval_notes", default=""),
            "approval_forced": bool(_value(state, "approval_forced", default=False)),
            "rejected_at": str(_value(state, "rejected_at", default="") or ""),
            "rejected_by": _value(state, "rejected_by", default=""),
            "rejection_notes": _value(state, "rejection_notes", default=""),
            "content": content,
        }
        json_path = (
            Path(state.json_artifact_uri)
            if state.json_artifact_uri
            else self._artifact_root(state) / f"{stem}.json"
        )
        markdown_path = (
            Path(state.markdown_artifact_uri)
            if state.markdown_artifact_uri
            else self._artifact_root(state) / f"{stem}.md"
        )
        atomic_write_text(json_path, _canonical(payload, pretty=True) + "\n")
        lines = [
            f"# {content.get('title') or state.work.title}",
            "",
            f"Book state version: {state.version}",
            f"Status: {payload['status']}",
            f"SHA-256: `{payload['sha256']}`",
            "",
            "## Premise",
            "",
            str(content.get("premise") or ""),
            "",
            "## Chapters",
            "",
        ]
        for chapter in content.get("chapters", []):
            lines.extend(
                [
                    f"### {chapter.get('number')}. {chapter.get('title')}",
                    "",
                    str(chapter.get("purpose") or ""),
                    "",
                ]
            )
        atomic_write_text(markdown_path, "\n".join(lines).rstrip() + "\n")
        fields = []
        for value, names in (
            (str(json_path), ("json_artifact_uri", "artifact_uri")),
            (str(markdown_path), ("markdown_artifact_uri",)),
        ):
            field = _assign(state, value, *names)
            if field:
                fields.append(field)
        if fields:
            state.save(update_fields=[*fields, "updated_at"])