from __future__ import annotations import json from types import SimpleNamespace import pytest from recoil.core.paths import ProjectPaths from recoil.pipeline._lib import derivation_manifest from recoil.pipeline._lib.derivation_sha import ( board_content_freshness_sha, plan_structural_sha, shotset_hash, ) from recoil.pipeline.cli import generate from recoil.pipeline.core.persistence import save_scene, scene_path from recoil.pipeline.core.take import Beat, Scene PROJECT = "fixture" EPISODE = 1 BATCH = "EP001_CONT_004" SCENE_ID = "BATCH_004" SHOT_ID = "EP001_SH10" @pytest.fixture def project_paths(tmp_path, monkeypatch): 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)) return ProjectPaths.for_project(PROJECT) def _raw_shot(shot_id: str, source_text_hash: str | None) -> dict: return { "shot_id": shot_id, "scene_index": 1, "source_text_hash": source_text_hash, "routing_data": { "is_env_only": False, "target_editorial_duration_s": 1.0, "has_dialogue": False, }, "asset_data": {"location_id": "LOC_A", "characters": []}, "prompt_data": {"shot_type": "MS"}, "spatial_data": {}, } def _write_plan(paths: ProjectPaths, spans: dict[str, str | None]) -> dict: raw = { "episode_id": "EP001", "project": PROJECT, "total_shots": len(spans), "shots": [_raw_shot(shot_id, span) for shot_id, span in spans.items()], } paths.plans_dir.mkdir(parents=True, exist_ok=True) (paths.plans_dir / "ep_001_plan.json").write_text(json.dumps(raw), encoding="utf-8") return raw def _stamp_fresh_scenes(project_paths: ProjectPaths, spans: dict[str, str | None]) -> None: raw = _write_plan(project_paths, spans) derivation_manifest.stamp_stage( PROJECT, EPISODE, "camera_tested", kind="derived", content_sha="sha256:CT", source={"script_sha": "script-sha"}, builder="test", ) derivation_manifest.stamp_bible( PROJECT, content_sha="sha256:BIBLE", builder="test", built_at="2026-06-20T00:00:00+00:00", ) derivation_manifest.stamp_stage( PROJECT, EPISODE, "plan", kind="derived", structural_sha=plan_structural_sha(raw), content_sha="sha256:PLAN", source={ "camera_tested_content_sha": "sha256:CT", "bible_content_sha": "sha256:BIBLE", }, builder="test", ) derivation_manifest.stamp_stage( PROJECT, EPISODE, "scenes", kind="derived", content_sha="sha256:SCENES", source={"plan_structural_sha": plan_structural_sha(raw)}, builder="test", extra={"shot_script_spans": {"SC_001": dict(spans)}}, ) def _beat(*, covered_shot_ids: list[str] | None = None, source_sha256: str = "source-v2"): shot_ids = list(covered_shot_ids or [SHOT_ID]) return SimpleNamespace( beat_id=SCENE_ID, beat_metadata={ "modality": "r2v_multi", "grouping": { "strategy": "continuity", "ordinal": 4, "shot_ids": shot_ids, "shotset_hash": shotset_hash(shot_ids), }, }, board={ "status": "approved", "artifact": "prep/ep_001/storyboards/board.png", "source_sha256": source_sha256, "fingerprint_version": 2, "approved_by": "JT", "updated_at": "2026-06-20T00:00:00Z", }, ) def _save_revalidate_scene() -> None: beat = Beat( beat_id=SCENE_ID, beat_metadata={ "modality": "r2v_multi", "grouping": { "strategy": "continuity", "ordinal": 4, "shot_ids": [SHOT_ID], "shotset_hash": shotset_hash([SHOT_ID]), }, }, ) save_scene(Scene(scene_id=SCENE_ID, beats=[beat]), scene_path(PROJECT, "ep_001", SCENE_ID)) def test_stamp_board_ssot_writes_content_freshness(project_paths, monkeypatch): spans = {SHOT_ID: "span-a"} _write_plan(project_paths, spans) derivation_manifest.stamp_stage( PROJECT, EPISODE, "plan", kind="derived", structural_sha="sha256:PLAN0", content_sha="sha256:PLAN-CONTENT", source={}, builder="test", ) generate._stamp_board_ssot(PROJECT, EPISODE, _beat()) key = shotset_hash([SHOT_ID]) record = derivation_manifest.get_board(PROJECT, EPISODE, key) assert record["covered_shot_ids"] == [SHOT_ID] assert record["shot_script_spans"] == spans assert record["content_freshness_sha"] == board_content_freshness_sha(spans) assert record["source_sha256"] == "source-v2" assert record["fingerprint_version"] == 2 missing_plan = project_paths.plans_dir / "ep_001_plan.json" missing_plan.unlink() missing_key = shotset_hash(["EP001_SH99"]) generate._stamp_board_ssot( PROJECT, EPISODE, _beat(covered_shot_ids=["EP001_SH99"], source_sha256="source-missing"), ) degraded = derivation_manifest.get_board(PROJECT, EPISODE, missing_key) assert degraded["shot_script_spans"] == {"EP001_SH99": None} assert degraded["content_freshness_sha"] is None assert degraded["source_sha256"] == "source-missing" assert derivation_manifest.board_freshness(PROJECT, EPISODE) == [ (key, False, "camera_tested"), (missing_key, False, "camera_tested"), ] def test_revalidate_board_writes_content_freshness(project_paths, monkeypatch): spans = {SHOT_ID: "span-a"} _stamp_fresh_scenes(project_paths, spans) monkeypatch.setattr( derivation_manifest.episode_script, "episode_script_sha", lambda project, episode: "script-sha", ) monkeypatch.setattr( generate, "_compute_board_source_sha256", lambda project, batch, beat: "source-v2-new", ) _save_revalidate_scene() key = shotset_hash([SHOT_ID]) derivation_manifest.stamp_board( PROJECT, EPISODE, key, { "status": "approved", "artifact": "prep/ep_001/storyboards/board.png", "source_sha256": "source-v2-old", "fingerprint_version": 2, "covered_shot_ids": [SHOT_ID], "approved_by": "JT", "updated_at": "2026-06-20T00:00:00Z", "needs_revalidation": True, }, ) code, payload = generate._run_revalidate_board( project=PROJECT, episode=EPISODE, batch=BATCH, ) assert code == generate.EXIT_OK assert payload == {"revalidated": True, "spend": 0} record = derivation_manifest.get_board(PROJECT, EPISODE, key) assert record["source_sha256"] == "source-v2-new" assert record["fingerprint_version"] == 2 assert "needs_revalidation" not in record assert record["shot_script_spans"] == spans assert record["content_freshness_sha"] == board_content_freshness_sha(spans) assert derivation_manifest.board_freshness(PROJECT, EPISODE) == [(key, True, None)] (project_paths.plans_dir / "ep_001_plan.json").unlink() derivation_manifest.stamp_board( PROJECT, EPISODE, key, { **record, "source_sha256": "source-v2-stale", "needs_revalidation": True, }, ) code, payload = generate._run_revalidate_board( project=PROJECT, episode=EPISODE, batch=BATCH, ) assert code == generate.EXIT_OK degraded = derivation_manifest.get_board(PROJECT, EPISODE, key) assert degraded["shot_script_spans"] == {SHOT_ID: None} assert degraded["content_freshness_sha"] is None assert derivation_manifest.board_freshness(PROJECT, EPISODE) == [ (key, False, "scenes") ]