from __future__ import annotations import dataclasses from pathlib import Path import pytest from recoil.pipeline._lib import derivation_manifest from recoil.pipeline._lib.plan_loader import CanonicalPlan, CanonicalShot from recoil.pipeline.core.persistence import save_scene, scene_path from recoil.pipeline.core.take import Beat, Scene from recoil.pipeline.orchestrator.batch_selector import parse_batch_selector from recoil.pipeline.orchestrator.from_script_target import resolve_from_script_target PROJECT = "fixture" EPISODE = 1 SELECTOR_TOKEN = "EP001_CONT_001" @pytest.fixture(autouse=True) def _projects_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: root = tmp_path / "projects" root.mkdir() (root / ".recoil-data-root").touch() project_root = root / PROJECT project_root.mkdir() monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(root)) monkeypatch.setattr( "recoil.pipeline.orchestrator.from_script_target.derivation_manifest.freshness", lambda *args, **kwargs: (False, "scenes"), ) return project_root def _selector(): selector = parse_batch_selector(SELECTOR_TOKEN) assert selector is not None return selector def _shot(shot_id: str, source_text_hash: str, *, scene_index: int = 1): return CanonicalShot( shot_id=shot_id, scene_index=scene_index, sequence_id=None, pipeline="video", previs_model=None, video_model="seeddance-2.0", location_id="L1", characters=[], shot_type="MS", duration_s=2.0, is_env_only=True, has_dialogue=False, aspect_ratio="9:16", raw={"shot_id": shot_id, "source_text_hash": source_text_hash}, ) def _plan(shots: list[CanonicalShot]) -> CanonicalPlan: return CanonicalPlan( episode_id="ep_001", project=PROJECT, shots=list(shots), source_path=Path("ep_001_plan.json"), raw={"shots": [dataclasses.asdict(shot) for shot in shots]}, ) def _spans(shots: list[CanonicalShot]) -> dict[str, str | None]: return { shot.shot_id: (shot.raw or {}).get("source_text_hash") for shot in shots } def _write_manifest(*, shot_script_spans) -> None: manifest = derivation_manifest.load(PROJECT, EPISODE) manifest["stages"]["scenes"] = { "kind": "derived", "content_sha": "content", "structural_sha": None, "source": {"plan_structural_sha": "plan"}, "builder": "test", "model": None, "via": None, "built_at": "2026-06-21T00:00:00Z", } if shot_script_spans is not None: manifest["stages"]["scenes"]["shot_script_spans"] = shot_script_spans derivation_manifest.save(PROJECT, EPISODE, manifest) def _write_target_scene(shots: list[CanonicalShot], *, scene_id: str = "BATCH_001"): grouping = { "strategy": "continuity", "ordinal": 1, "shot_ids": [shot.shot_id for shot in shots], "source_pass_id": None, } beat = Beat( beat_id=scene_id, beat_metadata={ "scene_id": scene_id, "modality": "r2v_multi", "shot": dataclasses.asdict(shots[0]), "batch_shots": [dataclasses.asdict(shot) for shot in shots], "grouping": dict(grouping), }, ) scene = Scene( scene_id=scene_id, beats=[beat], scene_metadata={ "episode": "ep_001", "project": PROJECT, "batch": True, "grouping": dict(grouping), }, ) save_scene(scene, scene_path(PROJECT, "ep_001", scene_id)) def test_resolve_from_script_target_proceed_for_stale_target() -> None: stored = [_shot("EP001_SH01", "old-a"), _shot("EP001_SH02", "old-b"), _shot("EP001_SH03", "old-c")] live = [_shot("EP001_SH01", "new-a"), _shot("EP001_SH02", "new-b"), _shot("EP001_SH03", "new-c")] _write_manifest(shot_script_spans={"BATCH_001": _spans(stored)}) _write_target_scene(stored) verdict = resolve_from_script_target( PROJECT, EPISODE, _selector(), live_plan=_plan(live), ) assert verdict.status == "PROCEED" assert verdict.reason == "stale" assert verdict.scene_id == "BATCH_001" assert verdict.live_spans == _spans(live) assert verdict.persisted_spans == _spans(stored) def test_resolve_from_script_target_not_stale_from_persisted_scene() -> None: stored_backlink = [_shot("EP001_SH01", "old-a"), _shot("EP001_SH02", "old-b"), _shot("EP001_SH03", "old-c")] current = [_shot("EP001_SH01", "new-a"), _shot("EP001_SH02", "new-b"), _shot("EP001_SH03", "new-c")] _write_manifest(shot_script_spans={"BATCH_001": _spans(stored_backlink)}) _write_target_scene(current) verdict = resolve_from_script_target( PROJECT, EPISODE, _selector(), live_plan=_plan(current), ) assert verdict.status == "REFUSE" assert verdict.reason == "not_stale" assert verdict.error == "not_stale" assert verdict.message == "this scene file already reflects the current script" def test_resolve_from_script_target_legacy_no_backlink() -> None: _write_manifest(shot_script_spans=None) verdict = resolve_from_script_target(PROJECT, EPISODE, _selector()) assert verdict.status == "REFUSE" assert verdict.reason == "legacy_no_backlink" assert verdict.error == "legacy_no_backlink" assert "run `rederive --episode` once" in verdict.message def test_resolve_from_script_target_shot_set_change() -> None: stored = [_shot("EP001_SH01", "old-a"), _shot("EP001_SH02", "old-b"), _shot("EP001_SH03", "old-c")] live = [_shot("EP001_SH01", "new-a"), _shot("EP001_SH02", "new-b"), _shot("EP001_SH99", "new-z")] _write_manifest(shot_script_spans={"BATCH_001": _spans(stored)}) _write_target_scene(stored) verdict = resolve_from_script_target( PROJECT, EPISODE, _selector(), live_plan=_plan(live), ) assert verdict.status == "ESCALATE" assert verdict.reason == "shot_set_change" assert verdict.error == "shot_set_change" assert "run a full `rederive --episode`" in verdict.message def test_resolve_from_script_target_scene_missing() -> None: stored = [_shot("EP001_SH01", "old-a"), _shot("EP001_SH02", "old-b"), _shot("EP001_SH03", "old-c")] _write_manifest(shot_script_spans={"BATCH_001": _spans(stored)}) verdict = resolve_from_script_target(PROJECT, EPISODE, _selector()) assert verdict.status == "REFUSE" assert verdict.reason == "scene_missing" assert verdict.error == "batch_scene_missing" assert verdict.path == scene_path(PROJECT, "ep_001", "BATCH_001")