"""REC-235 Phase 0 — atom-version index read-model acceptance tests.

``readmodel.resolve_atom_version`` / ``readmodel.get_atom`` are the BEAT-grain atom
resolver (the ``atom_read_model`` capability): address
``atom:{episode}/beat/{beat_id}@tN`` → artifact + continuity facets, resolved against the
ACTIVE persisted scene version (the REC-231 canon pointer), batch-board-INDEPENDENT. Pure
projection — never writes, never moves the pointer. Mirrors ``get_board``'s no-data shape.

Reuses the LIVE ``_make_project`` fixture from ``test_readmodel_board`` (creates the
``.recoil-data-root`` / ``project_config.json`` / ``RECOIL_PROJECTS_ROOT`` setup so
``ProjectPaths.for_project`` resolves), and builds the shot the way EpisodeRunner really
does — ``beat_metadata["shot"] = dataclasses.asdict(CanonicalShot(...))`` — so the facet
assertions are NON-vacuous (a flat-key fixture would pass while real persisted scenes
return ``None``).
"""

from __future__ import annotations

import dataclasses

import pytest

from recoil.api.schemas.atom import AtomModel, AtomVersionModel
from recoil.core.paths import ProjectPaths
from recoil.pipeline._lib.plan_loader import CanonicalShot, CharacterEntry
from recoil.pipeline.core.persistence import save_scene, scene_path
from recoil.pipeline.core.receipts import GenerationReceipt
from recoil.pipeline.core.registry import RunResult
from recoil.pipeline.core.scene_version_store import SceneVersionStore
from recoil.pipeline.core.take import Beat, Scene, Take
from recoil.pipeline.core.workflow import Workflow, WorkflowStep
from recoil.workspace import readmodel as rm
from recoil.workspace.tests.test_readmodel_board import _make_project

# A valid Beat.board pointer (the BATCH board). The atom resolvers MUST NOT read it — its
# distinct artifact lets every artifact assertion prove the take receipt is the source.
_BOARD = {
    "status": "proposed",
    "artifact": "prep/ep_001/storyboards/EP001_SH02_v01.png",
    "source_sha256": "deadbeef",
    "approved_by": None,
    "updated_at": "2026-06-23T00:00:00Z",
}


def _shot(shot_id: str = "EP001_SH02", location_id: str = "loc_bridge", char: str = "JADE") -> dict:
    """The EpisodeRunner shot shape: ``dataclasses.asdict(CanonicalShot(...))``."""
    return dataclasses.asdict(
        CanonicalShot(
            shot_id=shot_id,
            scene_index=0,
            sequence_id="BATCH_001",
            pipeline="video",
            previs_model=None,
            video_model=None,
            location_id=location_id,
            characters=[CharacterEntry(char_id=char, wardrobe_phase_id="jade_phase_1")],
            shot_type="WIDE",
            duration_s=None,
            is_env_only=False,
            has_dialogue=False,
            aspect_ratio=None,
            # REAL EpisodeRunner shape: action/props live nested in the original
            # plan-shot (raw), NOT flat keys. A flat-key fixture would pass vacuously
            # while production returns None (the bug this test now guards).
            raw={
                "prompt_data": {"kinetic_action": "grips the railing", "shot_type": "WIDE"},
                "asset_data": {
                    "props": ["railing"],
                    "prop_interaction": "grips the railing",
                    "location_id": location_id,
                    "characters": [{"char_id": char, "wardrobe_phase_id": "jade_phase_1"}],
                },
            },
        )
    )


def _receipt(output_path: str) -> GenerationReceipt:
    rr = RunResult(id="rr", modality="video_i2v", output_path=output_path, success=True)
    return GenerationReceipt(
        receipt_id="rcpt",
        modality="video_i2v",
        caller_id="test",
        project=None,
        episode=None,
        shot_id=None,
        timestamp_utc="2026-06-23T00:00:00Z",
        run_result=rr,
    )


def _take(beat_id: str, idx: int, output_path: str | None) -> Take:
    """A Take whose terminal workflow step carries a receipt output (or none)."""
    step = WorkflowStep(step_id="video", modality="video_i2v", payload={})
    if output_path is not None:
        step.receipt = _receipt(output_path)
        step.status = "succeeded"
    wf = Workflow(workflow_id=f"{beat_id}_wf{idx}", steps=[step])
    return Take(
        take_id=f"{beat_id}_take_{idx}",
        take_index=idx,
        workflow=wf,
        status="succeeded" if output_path is not None else "pending",
    )


def _main_scene(project: str) -> Scene:
    """One scene (BATCH_001) with beat EP001_SH02 carrying four takes that exercise every
    artifact-path branch + a board pointer the resolver must ignore. primary = take 12."""
    root = ProjectPaths.for_project(project).project_root
    beat = Beat(
        beat_id="EP001_SH02",
        takes=[
            _take("EP001_SH02", 12, "renders/ep_001/EP001_SH02/t12.mp4"),  # relative
            _take("EP001_SH02", 13, str(root / "renders/ep_001/EP001_SH02/t13.mp4")),  # abs-under-root
            _take("EP001_SH02", 14, str(root.parent / "outside_t14.mp4")),  # abs-outside-root
            _take("EP001_SH02", 15, "renders/../../escape.mp4"),  # relative '..' escape
        ],
        primary_take_id="EP001_SH02_take_12",
        beat_metadata={"shot": _shot(), "scene_id": "BATCH_001"},
        board=dict(_BOARD),
    )
    return Scene(scene_id="BATCH_001", beats=[beat])


def _save_main(project: str) -> None:
    save_scene(_main_scene(project), scene_path(project, 1, "BATCH_001"))


# ── _persistence_episode normalization (R4 CRITICAL helper) ─────────────────────────────


def test_persistence_episode_normalization():
    assert rm._persistence_episode("EP001") == 1
    assert rm._persistence_episode("ep001") == 1  # case-insensitive EP prefix
    assert rm._persistence_episode("EP12") == 12
    assert rm._persistence_episode("FOO") is None
    assert rm._persistence_episode("EPISODE1") is None


# ── resolve_atom_version: artifact + facets + canon + token normalization ────────────────


def test_resolve_atom_version_artifact_facets_canon_and_token_preservation(tmp_path, monkeypatch):
    project, *_ = _make_project(tmp_path, monkeypatch)
    _save_main(project)  # persisted on disk under the CANONICAL token ep_001_BATCH_001.json

    m = rm.resolve_atom_version("atom:EP001/beat/EP001_SH02@t12", project)

    assert isinstance(m, AtomVersionModel)
    assert m.reason is None
    # Episode-token normalization: EP001 URN resolves the ep_001-on-disk scene, AND the
    # ORIGINAL EP001 token is preserved verbatim (NOT rewritten to ep_001).
    assert m.atom_version_urn == "atom:EP001/beat/EP001_SH02@t12"
    assert m.beat_urn == "atom:EP001/beat/EP001_SH02"
    assert m.take_index == 12
    # Artifact is the take's receipt output (relative preserved), NOT Beat.board.
    assert m.artifact == "renders/ep_001/EP001_SH02/t12.mp4"
    assert m.artifact != _BOARD["artifact"]
    # is_canon True iff take 12 is the primary_take_id.
    assert m.is_canon is True
    # NON-VACUOUS facets extracted from the real CanonicalShot shape.
    assert m.facets["location"] == "loc_bridge"
    assert m.facets["identity"] == ["JADE"]
    assert m.facets["spine_anchor"]["shot_id"] == "EP001_SH02"
    assert m.facets["spine_anchor"]["scene_id"] == "BATCH_001"
    assert m.facets["spine_anchor"]["beat_index"] == 0
    # action ← prompt_data.kinetic_action (NOT the shot_type code "WIDE")
    assert m.facets["action"] == "grips the railing"
    # wardrobe ← promoted characters[].wardrobe_phase_id (char_id -> phase)
    assert m.facets["wardrobe"] == {"JADE": "jade_phase_1"}
    # props ← asset_data.props (a list)
    assert m.facets["props"] == ["railing"]


def test_resolve_non_canon_take(tmp_path, monkeypatch):
    project, *_ = _make_project(tmp_path, monkeypatch)
    _save_main(project)

    m = rm.resolve_atom_version("atom:EP001/beat/EP001_SH02@t13", project)

    assert m.reason is None
    assert m.take_index == 13
    assert m.is_canon is False  # primary is take 12, not 13


# ── Artifact path handling (R4 MAJOR) ────────────────────────────────────────────────────


def test_resolve_artifact_path_handling(tmp_path, monkeypatch):
    project, *_ = _make_project(tmp_path, monkeypatch)
    _save_main(project)

    rel = rm.resolve_atom_version("atom:EP001/beat/EP001_SH02@t12", project)
    under = rm.resolve_atom_version("atom:EP001/beat/EP001_SH02@t13", project)
    outside = rm.resolve_atom_version("atom:EP001/beat/EP001_SH02@t14", project)
    dotdot = rm.resolve_atom_version("atom:EP001/beat/EP001_SH02@t15", project)

    # Relative rel_path preserved UNCHANGED (the R4 MAJOR — never dropped to None).
    assert rel.artifact == "renders/ep_001/EP001_SH02/t12.mp4"
    # Absolute-under-root returned made-relative.
    assert under.artifact == "renders/ep_001/EP001_SH02/t13.mp4"
    # Absolute outside root, and relative '..' escape, both reject to None (never leak).
    assert outside.artifact is None
    assert dotdot.artifact is None


# ── get_atom: ordered versions, canon, beat-grain newer_unpointed ────────────────────────


def test_get_atom_ordered_versions_canon_and_newer_unpointed(tmp_path, monkeypatch):
    project, *_ = _make_project(tmp_path, monkeypatch)
    _save_main(project)

    a = rm.get_atom("atom:EP001/beat/EP001_SH02", project)

    assert isinstance(a, AtomModel)
    assert a.reason is None
    assert a.beat_urn == "atom:EP001/beat/EP001_SH02"
    assert a.take_indices == [12, 13, 14, 15]
    assert a.atom_version_urns == [
        "atom:EP001/beat/EP001_SH02@t12",
        "atom:EP001/beat/EP001_SH02@t13",
        "atom:EP001/beat/EP001_SH02@t14",
        "atom:EP001/beat/EP001_SH02@t15",
    ]
    assert a.canon_take_index == 12
    assert a.canon_urn == "atom:EP001/beat/EP001_SH02@t12"
    # beat-grain newer_unpointed = takes after the canon index (13, 14, 15 > 12).
    assert a.newer_unpointed_versions == 3


def test_get_atom_newer_unpointed_zero_when_no_canon(tmp_path, monkeypatch):
    project, *_ = _make_project(tmp_path, monkeypatch)
    beat = Beat(
        beat_id="EP001_SH02",
        takes=[_take("EP001_SH02", 1, "renders/a.mp4"), _take("EP001_SH02", 2, "renders/b.mp4")],
        beat_metadata={"shot": _shot()},
    )  # no primary_take_id set
    save_scene(Scene(scene_id="BATCH_001", beats=[beat]), scene_path(project, 1, "BATCH_001"))

    a = rm.get_atom("atom:EP001/beat/EP001_SH02", project)

    assert a.canon_take_index is None
    assert a.canon_urn is None
    assert a.newer_unpointed_versions == 0


# ── Reads the ACTIVE scene version, not a superseded one ─────────────────────────────────


def test_reads_active_scene_version_not_superseded(tmp_path, monkeypatch):
    project, *_ = _make_project(tmp_path, monkeypatch)
    # v1 (flat) — superseded body: beat EP001_SH09 take 1 → v1 artifact.
    v1_beat = Beat(
        beat_id="EP001_SH09",
        takes=[_take("EP001_SH09", 1, "renders/v1.mp4")],
        beat_metadata={"shot": _shot(shot_id="EP001_SH09")},
    )
    save_scene(Scene(scene_id="BATCH_009", beats=[v1_beat]), scene_path(project, 1, "BATCH_009"))
    # v2 (candidate) — conform active → v2: same beat take 1 → v2 artifact.
    v2_beat = Beat(
        beat_id="EP001_SH09",
        takes=[_take("EP001_SH09", 1, "renders/v2.mp4")],
        beat_metadata={"shot": _shot(shot_id="EP001_SH09")},
    )
    store = SceneVersionStore(project, 1)
    v2 = store.write_scene_candidate("BATCH_009", Scene(scene_id="BATCH_009", beats=[v2_beat]))
    store.conform("BATCH_009", v2)

    m = rm.resolve_atom_version("atom:EP001/beat/EP001_SH09@t1", project)

    assert m.reason is None
    assert m.artifact == "renders/v2.mp4"  # the ACTIVE (v2) body, not the superseded v1


# ── Resolution does not depend on the batch BOARD (independence gate) ────────────────────


def test_resolution_independent_of_batch_board(tmp_path, monkeypatch):
    project, *_ = _make_project(tmp_path, monkeypatch)
    _save_main(project)

    def _boom(*args, **kwargs):
        raise AssertionError("the batch BOARD must not be read by the atom resolvers")

    # Monkeypatch get_board to RAISE; do NOT touch load_manifest / load_scene_active_with_version,
    # which the active-version load legitimately needs.
    monkeypatch.setattr(rm, "get_board", _boom)

    m = rm.resolve_atom_version("atom:EP001/beat/EP001_SH02@t12", project)
    a = rm.get_atom("atom:EP001/beat/EP001_SH02", project)

    assert m.reason is None
    assert m.artifact == "renders/ep_001/EP001_SH02/t12.mp4"
    assert m.artifact != _BOARD["artifact"]  # the take output, NOT Beat.board's artifact
    assert a.reason is None
    assert a.take_indices == [12, 13, 14, 15]


# ── Ambiguous beat_id fails loud (episode-uniqueness contract) ───────────────────────────


def test_ambiguous_beat_id_fails_loud(tmp_path, monkeypatch):
    project, *_ = _make_project(tmp_path, monkeypatch)
    # The SAME beat_id active in TWO scenes of the episode → integrity violation.
    save_scene(
        Scene(scene_id="BATCH_001", beats=[Beat(beat_id="EP001_SH02", beat_metadata={"shot": _shot()})]),
        scene_path(project, 1, "BATCH_001"),
    )
    save_scene(
        Scene(scene_id="BATCH_002", beats=[Beat(beat_id="EP001_SH02", beat_metadata={"shot": _shot()})]),
        scene_path(project, 1, "BATCH_002"),
    )

    with pytest.raises(ValueError):
        rm.get_atom("atom:EP001/beat/EP001_SH02", project)
    with pytest.raises(ValueError):
        rm.resolve_atom_version("atom:EP001/beat/EP001_SH02@t1", project)


# ── Malformed URN raises; no-data returns a non-empty reason ─────────────────────────────


def test_malformed_urn_raises(tmp_path, monkeypatch):
    project, *_ = _make_project(tmp_path, monkeypatch)
    _save_main(project)

    with pytest.raises(ValueError):
        rm.resolve_atom_version("atom:EP001/beat/EP001_SH02", project)  # missing @tN
    with pytest.raises(ValueError):
        rm.get_atom("atom:EP001/beat/EP001_SH02@t12", project)  # @tN on a beat URN
    with pytest.raises(ValueError):
        rm.resolve_atom_version("not-an-atom-urn", project)


def test_no_data_returns_reason(tmp_path, monkeypatch):
    project, *_ = _make_project(tmp_path, monkeypatch)
    _save_main(project)

    missing_beat = rm.resolve_atom_version("atom:EP001/beat/EP001_SH99@t1", project)
    assert missing_beat.reason and missing_beat.artifact is None

    missing_take = rm.resolve_atom_version("atom:EP001/beat/EP001_SH02@t999", project)
    assert missing_take.reason and missing_take.artifact is None

    missing_atom = rm.get_atom("atom:EP001/beat/EP001_SH99", project)
    assert missing_atom.reason and missing_atom.take_indices == []

    # A well-formed URN whose episode token is not EP<NNN> can't address a persisted
    # episode → no-data reason (NOT a ValueError).
    bad_episode = rm.resolve_atom_version("atom:FOO/beat/EP001_SH02@t1", project)
    assert bad_episode.reason and bad_episode.artifact is None