import json from pathlib import Path import pytest from recoil.execution.execution_store import ExecutionStore from recoil.execution.pass_store import PassStore from recoil.pipeline.core.persistence import save_scene from recoil.pipeline.core.take import Beat, Scene from recoil.workspace import board as board_module from recoil.workspace import helpers as workspace_helpers from recoil.workspace.board import build_episode_board, normalize_episode @pytest.fixture def board_project(tmp_path, monkeypatch): projects_root = tmp_path / "projects" projects_root.mkdir() (projects_root / ".recoil-data-root").write_text("recoil-data-root\n") project = "fixture" project_root = projects_root / project project_root.mkdir() (project_root / "project_config.json").write_text("{}") monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(projects_root)) workspace_helpers._stores.clear() scenes_dir = project_root / "_pipeline" / "state" / "orchestration" / "scenes" storyboards_dir = project_root / "prep" / "ep_001" / "storyboards" scenes_dir.mkdir(parents=True) storyboards_dir.mkdir(parents=True) _write_scene( scenes_dir / "ep_001_BATCH_001.json", "BATCH_001", [ {"segment_id": "EP001_SH01", "setting": "kitchen", "duration_s": 1.5}, {"segment_id": "EP001_SH02", "setting": "kitchen", "duration_s": 2.0}, ], board_artifact="prep/ep_001/storyboards/BATCH_001_v06.png", board_status="proposed", ) _write_scene( scenes_dir / "ep_001_BATCH_002.json", "BATCH_002", [ {"segment_id": "EP001_SH03", "setting": "hall", "duration_s": 3}, {"segment_id": "EP001_SH04", "setting": "hall", "duration_s": 4.25}, ], ) (scenes_dir / "ep_001_BATCH_dryrun_polluted_backup_001.json").write_text("{}") _write_sidecar(storyboards_dir / "BATCH_001_v05.png.json", {"status": "approved"}) _write_sidecar(storyboards_dir / "BATCH_001_v06.png.json", {"status": "proposed"}) (storyboards_dir / "BATCH_001_v06.png").write_bytes(b"png") store = ExecutionStore(project) for shot_id in ("EP001_SH01", "EP001_SH02", "EP001_SH03", "EP001_SH04"): store.update_shot(shot_id, episode_id="EP001", status="previs_pending") store.close() passes = PassStore(project) passes.create_pass("EP001_PASS_001_SH01_02_TEST", ["EP001_SH01", "EP001_SH02"]) passes.close() yield project, storyboards_dir workspace_helpers._stores.clear() def test_normalize_episode_accepts_known_forms(): assert normalize_episode("1") == ("ep_001", "ep_001", "EP001") assert normalize_episode("ep_001") == ("ep_001", "ep_001", "EP001") assert normalize_episode("EP001") == ("ep_001", "ep_001", "EP001") assert normalize_episode("ep_1") == ("ep_001", "ep_001", "EP001") assert normalize_episode("EP_001") == ("ep_001", "ep_001", "EP001") def test_normalize_episode_rejects_junk(): with pytest.raises(ValueError, match="nope"): normalize_episode("nope") def test_build_episode_board_assembles_batches_and_coverage(board_project): project, _storyboards_dir = board_project payload = build_episode_board(project, "EP001") assert payload["episode_id"] == "EP001" assert [batch["batch_id"] for batch in payload["batches"]] == [ "BATCH_001", "BATCH_002", ] first, second = payload["batches"] assert first["scene_order"] == 1 assert first["board_artifact"] == "prep/ep_001/storyboards/BATCH_001_v06.png" assert first["version"] == 6 assert first["photoreal_artifact"] is None assert first["duration_s"] == 3.5 assert first["shared_location_id"] == "loc_batch_001" assert first["shared_characters"] == ["ada", "ben"] assert len(first["panels"]) == 2 assert first["coverage_summary"] == {"covered": 2, "total": 2, "awaiting": 0} assert second["board_artifact"] is None assert second["version"] is None assert second["duration_s"] == 7.25 assert len(second["panels"]) == 2 assert second["coverage_summary"] == {"covered": 0, "total": 2, "awaiting": 2} assert payload["summary"] == { "covered": 2, "total": 4, "awaiting": 2, "review_count": 2, } def test_build_episode_board_detects_photoreal_artifact(board_project): project, storyboards_dir = board_project (storyboards_dir / "BATCH_001_v06_photoreal.png").write_bytes(b"png") payload = build_episode_board(project, "1") assert ( payload["batches"][0]["photoreal_artifact"] == "prep/ep_001/storyboards/BATCH_001_v06_photoreal.png" ) def test_build_episode_board_version_matches_resolved_sidecar_artifact(board_project): project, storyboards_dir = board_project (storyboards_dir / "BATCH_001_v06.png.json").write_text( json.dumps({"artifact": "prep/ep_001/storyboards/BATCH_001_v07.png"}), encoding="utf-8", ) payload = build_episode_board(project, "1") first = payload["batches"][0] assert first["board_artifact"] == "prep/ep_001/storyboards/BATCH_001_v07.png" assert first["version"] == 7 def test_build_episode_board_unsafe_board_artifact_fails_closed( board_project, tmp_path, monkeypatch ): project, _storyboards_dir = board_project outside_dir = tmp_path / "outside" outside_dir.mkdir() unsafe_artifact = outside_dir / "BATCH_001_v06.png" unsafe_artifact.write_bytes(b"png") (outside_dir / "BATCH_001_v06_photoreal.png").write_bytes(b"png") original_load_scene_active = board_module.load_scene_active def _load_scene_with_unsafe_board(project_arg, episode_arg, batch_id): scene = original_load_scene_active(project_arg, episode_arg, batch_id) if batch_id == "BATCH_001": scene.beats[0].board["artifact"] = str(unsafe_artifact) return scene monkeypatch.setattr(board_module, "load_scene_active", _load_scene_with_unsafe_board) payload = build_episode_board(project, "1") first = payload["batches"][0] assert first["board_artifact"] is None assert first["version"] is None assert first["status"] == "proposed" assert first["photoreal_artifact"] is None def _write_scene( path: Path, batch_id: str, batch_shots: list[dict], board_artifact: str | None = None, board_status: str | None = None, ) -> None: beat = Beat( beat_id=batch_id, beat_metadata={ "batch_shots": batch_shots, "shared_location_id": f"loc_{batch_id.lower()}", "shared_characters": ["ada", "ben"], "total_duration_s": sum(shot["duration_s"] for shot in batch_shots), }, ) if board_artifact is not None: # REC-231 Phase 4: the wall dereferences the active body's pointed board # (beat.board.artifact), not the newest sidecar on disk. beat.set_board_proposed( artifact=board_artifact, source_sha256="0" * 64, fingerprint_version=1 ) if board_status == "approved": beat.approve_board("JT") save_scene(Scene(scene_id=batch_id, beats=[beat]), path) def _write_sidecar(path: Path, data: dict) -> None: path.write_text(json.dumps(data), encoding="utf-8")