from __future__ import annotations import dataclasses import json import sys from pathlib import Path from types import SimpleNamespace import pytest from recoil.core.paths import ProjectPaths from recoil.pipeline._lib import board_builder as bb from recoil.pipeline._lib import derivation_manifest from recoil.pipeline._lib import dispatch_payload as dp from recoil.pipeline._lib.board_builder import compute_source_sha256 from recoil.pipeline._lib.derivation_sha import shotset_hash from recoil.pipeline._lib.plan_loader import CanonicalShot from recoil.pipeline.cli import generate from recoil.pipeline.core.persistence import ( SceneVersionConflictError, load_scene, save_scene, scene_path, ) from recoil.pipeline.core.receipts import GenerationReceipt, utc_now_iso8601 from recoil.pipeline.core.registry import MODALITY_STORYBOARD, RunResult from recoil.pipeline.core.scene_version_store import SceneVersionStore from recoil.pipeline.core.take import Beat, Scene from recoil.pipeline.orchestrator.episode_runner import EpisodeRunner, RerollPreflightError PROJECT = "fixture" BATCH_ID = "EP001_CONT_004" SCENE_ID = "BATCH_004" @pytest.fixture() def project_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: root = tmp_path / "projects" root.mkdir() (root / ".recoil-data-root").touch() project = root / PROJECT project.mkdir() monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(root)) return project def _shot(shot_id: str = "EP001_SH10") -> dict: return { "shot_id": shot_id, "scene_index": 1, "sequence_id": "SEQ001", "pipeline": "video", "previs_model": None, "video_model": "seeddance-2.0", "location_id": None, "characters": [], "shot_type": "MS", "duration_s": 1.0, "is_env_only": False, "has_dialogue": False, "aspect_ratio": "9:16", "intent": "Jade studies the pod seal.", "asset_data": {"characters": [], "location_id": None}, "spatial_data": {}, "raw": {"source_text": "Jade studies the pod seal."}, } def _source_sha( *, intent: str = "Jade studies the pod seal.", duration_s: float = 1.0, version: int = 1, ) -> str: return compute_source_sha256( [ { "shot_id": "EP001_SH10", "start_s": 0.0, "end_s": duration_s, "duration_s": duration_s, "intent": intent, "sublocation": None, } ], version=version, ) def _write_png(path: Path, size: int = 2048) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(b"\x89PNG\r\n" + (b"0" * size)) def _write_ref(project_root: Path, rel: str) -> str: path = project_root / rel _write_png(path) return rel def _current_source_from_beat(beat: Beat, *, version: int = 2) -> str: primitive = bb._primitive_from_beat(PROJECT, BATCH_ID, beat) segments = list(getattr(primitive, "timing_segments", []) or []) return compute_source_sha256(segments, version=version) def _stamp_finish_record( beat: Beat, *, status: str = "approved", artifact: str, source_sha256: str | None = None, fingerprint_version: int = 2, photoreal_artifact: str | None = None, needs_revalidation: bool | None = None, ) -> dict: grouping = (beat.beat_metadata or {}).get("grouping") or {} h = grouping.get("shotset_hash") or shotset_hash(grouping["shot_ids"]) record = { "status": status, "artifact": artifact, "photoreal_artifact": photoreal_artifact, "source_sha256": source_sha256 or _current_source_from_beat(beat, version=fingerprint_version), "fingerprint_version": fingerprint_version, "covered_shot_ids": list(grouping.get("shot_ids") or []), "approved_by": "JT", "updated_at": utc_now_iso8601(), } if needs_revalidation is not None: record["needs_revalidation"] = needs_revalidation derivation_manifest.stamp_board(PROJECT, 1, h, record) return record def _save_board_scene( project_root: Path, *, status: str = "proposed", photoreal_artifact: str | None = None, sidecar_refs: dict | None = None, ) -> Path: artifact_rel = "prep/ep_001/storyboards/EP001_CONT_004_v03.png" artifact_path = project_root / artifact_rel _write_png(artifact_path) sidecar = { "kind": "storyboard", "batch_id": BATCH_ID, "version": 3, "source_sha256": _source_sha(), "identity_refs": sidecar_refs.get("identity_refs", []) if sidecar_refs else [], "sublocation_refs": sidecar_refs.get("sublocation_refs", []) if sidecar_refs else [], "prop_refs": sidecar_refs.get("prop_refs", []) if sidecar_refs else [], } Path(f"{artifact_path}.json").write_text(json.dumps(sidecar), encoding="utf-8") shot = _shot() board = { "status": status, "artifact": artifact_rel, "source_sha256": _source_sha(), "approved_by": "JT" if status == "approved" else None, "updated_at": utc_now_iso8601(), } if photoreal_artifact: board["photoreal_artifact"] = photoreal_artifact beat = Beat( beat_id=SCENE_ID, beat_metadata={ "scene_id": SCENE_ID, "modality": "r2v_multi", "shot": shot, "batch_shots": [shot], "batch_summary": {"shared_characters": ["JADE"]}, "grouping": { "strategy": "continuity", "ordinal": 4, "shot_ids": ["EP001_SH10"], "source_pass_id": None, "shotset_hash": shotset_hash(["EP001_SH10"]), }, }, board=board, ) scene = Scene(scene_id=SCENE_ID, beats=[beat]) path = scene_path(PROJECT, "ep_001", SCENE_ID) save_scene(scene, path) return path def _save_born_v2_finish_scene( project_root: Path, *, stored_intent: str = "Jade studies the pod seal.", current_intent: str = "Jade studies the pod seal.", stored_duration_s: float = 1.0, current_duration_s: float = 1.0, stamp_ssot: bool = True, record_status: str = "approved", needs_revalidation: bool | None = None, ) -> Path: artifact_rel = "prep/ep_001/storyboards/EP001_CONT_004_v03.png" artifact_path = project_root / artifact_rel _write_png(artifact_path) source_sha256 = _source_sha( intent=stored_intent, duration_s=stored_duration_s, version=2, ) sidecar = { "kind": "storyboard", "batch_id": BATCH_ID, "version": 3, "source_sha256": source_sha256, "fingerprint_version": 2, "identity_refs": [], "sublocation_refs": [], "prop_refs": [], } Path(f"{artifact_path}.json").write_text(json.dumps(sidecar), encoding="utf-8") shot = _shot() shot["intent"] = current_intent shot["duration_s"] = current_duration_s shot["raw"]["source_text"] = current_intent board = { "status": "approved", "artifact": artifact_rel, "source_sha256": source_sha256, "fingerprint_version": 2, "approved_by": "JT", "updated_at": utc_now_iso8601(), } beat = Beat( beat_id=SCENE_ID, beat_metadata={ "scene_id": SCENE_ID, "modality": "r2v_multi", "shot": shot, "batch_shots": [shot], "batch_summary": {"shared_characters": ["JADE"]}, "grouping": { "strategy": "continuity", "ordinal": 4, "shot_ids": ["EP001_SH10"], "source_pass_id": None, "shotset_hash": shotset_hash(["EP001_SH10"]), }, }, board=board, ) scene = Scene(scene_id=SCENE_ID, beats=[beat]) path = scene_path(PROJECT, "ep_001", SCENE_ID) save_scene(scene, path) if stamp_ssot: _stamp_finish_record( beat, status=record_status, artifact=artifact_rel, source_sha256=source_sha256, fingerprint_version=2, needs_revalidation=needs_revalidation, ) return path def _settings_passthrough(segments, **_kwargs): return [dict(segment, setting=f"Setting {i}") for i, segment in enumerate(segments, start=1)] def _receipt(success: bool = True, error: str | None = None) -> GenerationReceipt: return GenerationReceipt( receipt_id="rcpt_finish", modality=MODALITY_STORYBOARD, caller_id="board_builder", project=PROJECT, episode=1, shot_id=BATCH_ID, timestamp_utc="2026-06-13T00:00:00Z", run_result=RunResult( id="run_finish", modality=MODALITY_STORYBOARD, output_path="/tmp/finish.png" if success else None, metadata={}, success=success, error=error, ), ) def _capture_finish_dispatch(monkeypatch: pytest.MonkeyPatch, *, success: bool = True): calls: list[dict] = [] def fake_dispatch(modality, payload, *, context): calls.append({"modality": modality, "payload": payload, "context": context}) if success: out = Path(payload["save_dir"]) / f"{payload['filename_stem']}.png" _write_png(out) Path(f"{out}.json").write_text( json.dumps(payload["sidecar_extra"]), encoding="utf-8", ) return _receipt(success=success, error=None if success else "finish boom") monkeypatch.setattr(bb, "dispatch", fake_dispatch) monkeypatch.setattr(bb, "derive_settings", _settings_passthrough) monkeypatch.setattr(generate, "_build_step_runner_for_episode", lambda *a, **k: object()) return calls def _run_cli(monkeypatch: pytest.MonkeyPatch, capsys, *args: str) -> tuple[int, dict, list[str]]: monkeypatch.setattr(sys, "argv", ["generate.py", *args]) code = generate.main() lines = capsys.readouterr().out.strip().splitlines() return code, json.loads(lines[-1]), lines def test_approve_dispatches_finish_with_copied_sidecar_refs_and_persists( project_root: Path, monkeypatch: pytest.MonkeyPatch, capsys, ) -> None: sidecar_refs = { "identity_refs": [ _write_ref(project_root, "assets/sidecar/jade_front.png"), _write_ref(project_root, "assets/sidecar/jade_profile.png"), ], "sublocation_refs": [ {"slug": "pod_platform", "ref": _write_ref(project_root, "assets/sidecar/pod_platform.png")} ], "prop_refs": [ {"slug": "cryo_pod", "ref": _write_ref(project_root, "assets/sidecar/cryo_pod.png")} ], } scene_file = _save_board_scene(project_root, sidecar_refs=sidecar_refs) calls = _capture_finish_dispatch(monkeypatch) code, payload, lines = _run_cli( monkeypatch, capsys, "--project", PROJECT, "--episode", "1", "--approve-board", BATCH_ID, ) assert code == 0 assert "photoreal finish: 1 × ~$0.41" in lines assert len(calls) == 1 finish_payload = calls[0]["payload"] assert calls[0]["modality"] == MODALITY_STORYBOARD assert finish_payload["quality"] == "high" assert finish_payload["size_override"] == "1728x3072" assert finish_payload["reference_images"][0].endswith("EP001_CONT_004_v03.png") assert finish_payload["reference_images"][1:] == [ str(project_root / "assets/sidecar/jade_front.png"), str(project_root / "assets/sidecar/jade_profile.png"), str(project_root / "assets/sidecar/pod_platform.png"), str(project_root / "assets/sidecar/cryo_pod.png"), ] sidecar = finish_payload["sidecar_extra"] assert sidecar["kind"] == "storyboard_finish" assert sidecar["finish_of"] == "prep/ep_001/storyboards/EP001_CONT_004_v03.png" assert sidecar["source_sha256"] == _source_sha() assert sidecar["iteration_tier"] is None assert payload["photoreal_artifact"].endswith("EP001_CONT_004_v03_photoreal.png") assert load_scene(scene_file).beats[0].board["photoreal_artifact"] == payload["photoreal_artifact"] def test_render_board_finish_born_v2_prose_reroll_is_not_stale( project_root: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: _save_born_v2_finish_scene( project_root, stored_intent="Jade studies the pod seal.", current_intent="Jade reconsiders the pod seal.", ) calls = _capture_finish_dispatch(monkeypatch) result = bb.render_board_finish( PROJECT, 1, BATCH_ID, step_runner=object(), expected_version=1, ) assert result["success"] is True assert len(calls) == 1 assert result["source_sha256"] == _source_sha( intent="Jade studies the pod seal.", version=2, ) def test_render_board_finish_born_v2_structural_change_is_blocked_by_resolver( project_root: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: _save_born_v2_finish_scene( project_root, stored_duration_s=1.0, current_duration_s=1.25, ) calls = _capture_finish_dispatch(monkeypatch) with pytest.raises(bb.BoardBuilderError, match="requires an approved storyboard"): bb.render_board_finish( PROJECT, 1, BATCH_ID, step_runner=object(), expected_version=1, ) assert calls == [] @pytest.mark.parametrize( "record_kwargs", [ {"record_status": "rejected"}, {"needs_revalidation": True}, ], ) def test_render_board_finish_blocks_ssot_smuggled_cache_approval( project_root: Path, monkeypatch: pytest.MonkeyPatch, record_kwargs: dict, ) -> None: _save_born_v2_finish_scene(project_root, **record_kwargs) calls = _capture_finish_dispatch(monkeypatch) with pytest.raises(bb.BoardBuilderError, match="requires an approved storyboard"): bb.render_board_finish( PROJECT, 1, BATCH_ID, step_runner=object(), expected_version=1, ) assert calls == [] def test_render_board_finish_active_version_move_rejects_before_dispatch( project_root: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: scene_file = _save_born_v2_finish_scene(project_root) store = SceneVersionStore(PROJECT, "ep_001") store.write_scene_candidate(SCENE_ID, load_scene(scene_file)) calls = _capture_finish_dispatch(monkeypatch) def _derive_then_move(segments, **kwargs): store.conform(SCENE_ID, 2) return _settings_passthrough(segments, **kwargs) monkeypatch.setattr(bb, "derive_settings", _derive_then_move) # REC-231: caller-captured active version must be rechecked before paid finish dispatch. with pytest.raises(SceneVersionConflictError): bb.render_board_finish( PROJECT, 1, BATCH_ID, step_runner=object(), expected_version=1, ) assert calls == [] def test_render_board_finish_carries_approved_board_from_ssot( project_root: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: scene_file = _save_born_v2_finish_scene(project_root) scene = load_scene(scene_file) scene.beats[0].board = None save_scene(scene, scene_file) calls = _capture_finish_dispatch(monkeypatch) result = bb.render_board_finish( PROJECT, 1, BATCH_ID, step_runner=object(), expected_version=1, ) assert result["success"] is True assert len(calls) == 1 assert calls[0]["payload"]["reference_images"][0].endswith( "EP001_CONT_004_v03.png" ) def test_render_board_finish_blocks_grouped_cache_without_ssot_record( project_root: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: _save_born_v2_finish_scene(project_root, stamp_ssot=False) calls = _capture_finish_dispatch(monkeypatch) with pytest.raises(bb.BoardBuilderError, match="requires an approved storyboard"): bb.render_board_finish( PROJECT, 1, BATCH_ID, step_runner=object(), expected_version=1, ) assert calls == [] def test_render_board_finish_stale_check_remains_after_resolver( project_root: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: _save_born_v2_finish_scene( project_root, stored_duration_s=1.0, current_duration_s=1.25, stamp_ssot=False, ) def approve_cached_board(_project, _episode, beat): return True, beat.board monkeypatch.setattr(bb, "resolve_board_for_spend", approve_cached_board) calls = _capture_finish_dispatch(monkeypatch) with pytest.raises(bb.BoardBuilderError, match="source_sha256 is stale"): bb.render_board_finish( PROJECT, 1, BATCH_ID, step_runner=object(), expected_version=1, ) assert calls == [] def test_sidecar_listed_ref_missing_aborts_finish_but_approval_stands( project_root: Path, monkeypatch: pytest.MonkeyPatch, capsys, ) -> None: scene_file = _save_board_scene( project_root, sidecar_refs={"prop_refs": [{"slug": "cryo_pod", "ref": "assets/missing/pod.png"}]}, ) calls = _capture_finish_dispatch(monkeypatch) code, payload, _lines = _run_cli( monkeypatch, capsys, "--project", PROJECT, "--episode", "1", "--approve-board", BATCH_ID, ) assert code != 0 assert calls == [] assert "assets/missing/pod.png" in payload["message"] assert payload["hint"] == "re-run --approve-board to retry the finish" board = load_scene(scene_file).beats[0].board assert board["status"] == "approved" assert "photoreal_artifact" not in board def test_approve_again_skips_existing_finish( project_root: Path, monkeypatch: pytest.MonkeyPatch, capsys, ) -> None: finish_rel = "prep/ep_001/storyboards/EP001_CONT_004_v03_photoreal.png" _write_png(project_root / finish_rel) scene_file = _save_board_scene(project_root, status="approved", photoreal_artifact=finish_rel) monkeypatch.setattr(generate, "render_board_finish", lambda *a, **k: pytest.fail("no second dispatch")) code, payload, lines = _run_cli( monkeypatch, capsys, "--project", PROJECT, "--episode", "1", "--approve-board", BATCH_ID, ) assert code == 0 assert "photoreal finish: already exists; skipping" in lines assert payload["photoreal_artifact"] == finish_rel assert load_scene(scene_file).beats[0].board["photoreal_artifact"] == finish_rel def test_finish_dispatch_failure_leaves_approval_standing_without_photoreal( project_root: Path, monkeypatch: pytest.MonkeyPatch, capsys, ) -> None: scene_file = _save_board_scene(project_root) calls = _capture_finish_dispatch(monkeypatch, success=False) code, payload, _lines = _run_cli( monkeypatch, capsys, "--project", PROJECT, "--episode", "1", "--approve-board", BATCH_ID, ) assert code != 0 assert len(calls) == 1 assert payload["message"] == "finish boom" assert payload["hint"] == "re-run --approve-board to retry the finish" board = load_scene(scene_file).beats[0].board assert board["status"] == "approved" assert "photoreal_artifact" not in board def test_preferred_board_artifact_prefers_existing_photoreal_and_warns_on_missing( tmp_path: Path, caplog: pytest.LogCaptureFixture, ) -> None: pencil = tmp_path / "pencil.png" finish = tmp_path / "finish.png" _write_png(pencil) _write_png(finish) assert bb.preferred_board_artifact( {"artifact": "pencil.png", "photoreal_artifact": "finish.png"}, project_root=tmp_path, ) == "finish.png" assert bb.preferred_board_artifact( {"artifact": "pencil.png"}, project_root=tmp_path, ) == "pencil.png" caplog.clear() assert bb.preferred_board_artifact( {"artifact": "pencil.png", "photoreal_artifact": "missing.png"}, project_root=tmp_path, ) == "pencil.png" assert "board_finish_missing" in caplog.text assert bb.preferred_board_artifact(None, project_root=tmp_path) is None def test_live_runner_uses_photoreal_for_dispatch_and_drift_fingerprint( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: paths = ProjectPaths(project_root=tmp_path / "fixture_project") paths.project_root.mkdir() _write_png(paths.project_root / "prep/ep_001/storyboards/pencil.png") _write_png(paths.project_root / "prep/ep_001/storyboards/finish.png") monkeypatch.setattr( ProjectPaths, "for_project", classmethod(lambda cls, project=None: paths), ) captured: list[str | None] = [] def fake_build_dispatch_payload(**kwargs): captured.append(kwargs.get("board_ref_path")) return {"reference_images": [kwargs.get("board_ref_path")]} monkeypatch.setattr(dp, "build_dispatch_payload", fake_build_dispatch_payload) shot = CanonicalShot( shot_id="EP001_SH01", scene_index=1, sequence_id="SEQ001", pipeline="video", previs_model=None, video_model="seeddance-2.0", location_id=None, characters=[], shot_type="MS", duration_s=2.0, is_env_only=False, has_dialogue=False, aspect_ratio="9:16", raw={}, ) beat = Beat( beat_id=SCENE_ID, beat_metadata={ "modality": "r2v_multi", "scene_id": SCENE_ID, "shot": dataclasses.asdict(shot), "batch_shots": [dataclasses.asdict(shot)], "grouping": {"strategy": "continuity", "ordinal": 4, "shot_ids": []}, "inputs_fingerprint": "stale", }, board={ "status": "approved", "artifact": "prep/ep_001/storyboards/pencil.png", "photoreal_artifact": "prep/ep_001/storyboards/finish.png", "source_sha256": "sha", "approved_by": "JT", "updated_at": utc_now_iso8601(), }, ) fake_take = SimpleNamespace( take_id="t1", take_index=1, status="succeeded", take_metadata={}, ) beat.takes.append(fake_take) beat.primary_take_id = "t1" scene = Scene(scene_id=SCENE_ID, beats=[beat], scene_metadata={}) runner = EpisodeRunner.__new__(EpisodeRunner) runner.project = PROJECT runner.episode = "ep_001" runner.model_override = None runner.tier_override = None runner.generate_audio = None workflow = runner._build_workflow_for_beat(beat, take_index=2, beat_index=0) assert workflow.steps[0].payload["reference_images"] == [ "prep/ep_001/storyboards/finish.png" ] fake_take.status = "succeeded" beat.beat_metadata["inputs_fingerprint"] = "stale" with pytest.raises(RerollPreflightError): runner.revalidate_succeeded_fingerprints(scene, mutate=False) assert captured == [ "prep/ep_001/storyboards/finish.png", "prep/ep_001/storyboards/finish.png", ]