import asyncio from pathlib import Path import pytest from recoil.pipeline._lib.plan_loader import CanonicalPlan, CanonicalShot from recoil.pipeline.core.persistence import ( load_manifest, load_scene, save_scene, scene_path, scene_version_path, ) from recoil.pipeline.core.take import Beat, Scene from recoil.pipeline.orchestrator.episode_runner import EpisodeRunner def _shot( batch_index: int, shot_index: int, *, shots_per_batch: int = 3, ) -> CanonicalShot: ordinal = (batch_index - 1) * shots_per_batch + shot_index return CanonicalShot( shot_id=f"EP001_SH{ordinal:02d}", scene_index=batch_index, sequence_id=None, pipeline="video", previs_model=None, video_model="seeddance-2.0", location_id=f"LOC_{batch_index}", characters=[], shot_type="MS", duration_s=2.0, is_env_only=False, has_dialogue=False, aspect_ratio="9:16", quality=None, cinematography=None, raw={}, ) def _canonical_plan() -> CanonicalPlan: shots = [ _shot(batch_index, shot_index) for batch_index in range(1, 5) for shot_index in range(1, 4) ] return CanonicalPlan( episode_id="ep_001", project="fixture", shots=shots, source_path=Path("fixture_plan.json"), raw={ "episode_id": "ep_001", "project": "fixture", "shots": [{"shot_id": shot.shot_id} for shot in shots], }, ) def _empty_plan() -> CanonicalPlan: return CanonicalPlan( episode_id="ep_001", project="fixture", shots=[], source_path=Path("fixture_empty_plan.json"), raw={ "episode_id": "ep_001", "project": "fixture", "shots": [], }, ) def _stale_scene(scene_id: str, *, locked: bool = False) -> Scene: return Scene( scene_id=scene_id, beats=[ Beat( beat_id=f"OLD_{scene_id}", beat_metadata={"marker": "stale"}, ) ], scene_metadata={"marker": "stale"}, locked=locked, lock_reason="staged" if locked else None, ) def _seed_stale_scenes() -> dict[str, bytes]: original: dict[str, bytes] = {} for index in range(1, 5): scene_id = f"BATCH_{index:03d}" path = scene_path("fixture", "ep_001", scene_id) save_scene(_stale_scene(scene_id, locked=scene_id == "BATCH_004"), path) original[scene_id] = path.read_bytes() return original def _runner(plan: CanonicalPlan) -> EpisodeRunner: return EpisodeRunner( project="fixture", plan=plan.raw, casting={}, episode="ep_001", concurrency=1, ) def _configure_projects_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: (tmp_path / ".recoil-data-root").write_text("recoil-data-root\n", encoding="utf-8") (tmp_path / "fixture").mkdir() monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(tmp_path)) def _trap_run_scene(calls: list[str]): async def _run_scene(scene, **kwargs): # noqa: ANN001, ANN003 calls.append(scene.scene_id) raise AssertionError("run_scene must not be called in derive_only mode") return _run_scene def test_derive_only_appends_all_including_locked_and_skips_dispatch( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: """REC-231 Phase 7: derive-only re-derivation appends a candidate for EVERY batch — including the (now inert-)locked BATCH_004. ``scene.locked`` no longer skips the write (the lock-as-clobber-guard is DELETED), so nothing is reported skipped; every flat body is materialized IN PLACE as v1 (byte-unchanged) with the fresh structure appended as a v2 candidate, and the pointer never moves.""" _configure_projects_root(tmp_path, monkeypatch) original = _seed_stale_scenes() plan = _canonical_plan() runner = _runner(plan) run_scene_calls: list[str] = [] monkeypatch.setattr(runner, "run_scene", _trap_run_scene(run_scene_calls)) result = asyncio.run(runner.run_episode_batches(plan, derive_only=True)) assert isinstance(result, dict) assert result["derive_only"] is True assert result["dry_run"] is False assert "BATCH_001" in result["written"] assert "BATCH_004" in result["written"] # the locked batch is appended, not skipped assert result["skipped"] == [] assert run_scene_calls == [] # Every batch — unlocked BATCH_001 AND the inert-locked BATCH_004 — materializes its # flat body IN PLACE as v1 (byte-unchanged) and appends the fresh structure as a v2 # candidate; the pointer never moves. for batch_id in ("BATCH_001", "BATCH_004"): flat_path = scene_path("fixture", "ep_001", batch_id) assert flat_path.exists() assert flat_path.read_bytes() == original[batch_id] # v1 preserved (no clobber) assert load_scene(flat_path).scene_id == batch_id manifest = load_manifest("fixture", "ep_001", batch_id) assert manifest["active_version"] == 1 assert manifest["versions"][0]["source"] == "legacy_flat" assert manifest["versions"][1]["state"] == "candidate" assert scene_version_path("fixture", "ep_001", batch_id, 2).exists() def test_derive_only_no_shots_returns_summary_dict( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: _configure_projects_root(tmp_path, monkeypatch) plan = _empty_plan() runner = _runner(plan) run_scene_calls: list[str] = [] monkeypatch.setattr(runner, "run_scene", _trap_run_scene(run_scene_calls)) result = asyncio.run(runner.run_episode_batches(plan, derive_only=True)) assert result == { "derive_only": True, "written": [], "skipped": [], "dry_run": False, } assert run_scene_calls == [] def test_derive_only_dry_run_is_read_only_with_inert_lock( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: """REC-231 Phase 7: a dry-run derive stays fully read-only (REC-100) and the lock is inert — the locked BATCH_004 is a would-write candidate (no longer reported skipped), yet NOTHING is persisted: no manifest, no version body, and the locked flat body is byte-untouched.""" _configure_projects_root(tmp_path, monkeypatch) original = _seed_stale_scenes() plan = _canonical_plan() runner = _runner(plan) run_scene_calls: list[str] = [] monkeypatch.setattr(runner, "run_scene", _trap_run_scene(run_scene_calls)) result = asyncio.run( runner.run_episode_batches(plan, derive_only=True, dry_run=True) ) assert isinstance(result, dict) assert result["derive_only"] is True assert result["dry_run"] is True assert result["skipped"] == [] assert "BATCH_004" in result["written"] # would-write (lock inert), not skipped assert run_scene_calls == [] # Dry-run persists NOTHING: locked body byte-untouched, no manifest, no v2 body. locked_path = scene_path("fixture", "ep_001", "BATCH_004") assert locked_path.read_bytes() == original["BATCH_004"] assert load_manifest("fixture", "ep_001", "BATCH_004") is None assert not scene_version_path("fixture", "ep_001", "BATCH_004", 2).exists()