from __future__ import annotations import json from pathlib import Path import pytest from recoil.execution.step_types import ProjectPaths from recoil.pipeline._lib import derivation_manifest, episode_script from recoil.pipeline._lib.derivation_sha import plan_structural_sha from recoil.pipeline.cli import generate PROJECT = "fixture" EPISODE = 1 EPISODE_STR = "ep_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)) return project_root def _shot(batch_index: int, shot_index: int) -> dict: ordinal = (batch_index - 1) * 3 + shot_index return { "shot_id": f"EP001_SH{ordinal:02d}", "scene_index": batch_index, "pipeline": "video", "video_model": "seeddance-2.0", "asset_data": { "characters": [], "location_id": f"LOC_{batch_index}", }, "prompt_data": {"shot_type": "MS"}, "routing_data": { "target_editorial_duration_s": 2.0, "is_env_only": False, "has_dialogue": False, }, "spatial_data": {}, "aspect_ratio": "9:16", } def _plan_payload() -> dict: return { "episode_id": "ep_001", "project": PROJECT, "shots": [ _shot(batch_index, shot_index) for batch_index in range(1, 5) for shot_index in range(1, 4) ], } def _write_script(text: str) -> Path: paths = derivation_manifest.ProjectPaths.for_project(PROJECT) paths.episodes_dir.mkdir(parents=True, exist_ok=True) script_path = paths.episodes_dir / "ep_001.md" script_path.write_text(text, encoding="utf-8") return script_path def _write_plan(payload: dict | None = None) -> Path: paths = ProjectPaths.for_episode(PROJECT, EPISODE) paths.plans_dir.mkdir(parents=True, exist_ok=True) plan_path = paths.plans_dir / "ep_001_plan.json" plan_path.write_text(json.dumps(payload or _plan_payload(), indent=2), encoding="utf-8") return plan_path def _stamp_plan_chain( *, camera_script_sha: str | None = None, plan_camera_sha: str = "sha256:CAMERA", ) -> None: script_sha = camera_script_sha or episode_script.episode_script_sha(PROJECT, EPISODE) plan = _plan_payload() derivation_manifest.stamp_stage( PROJECT, EPISODE, "camera_tested", kind="derived", content_sha="sha256:CAMERA", 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(plan), content_sha="sha256:PLAN", source={ "camera_tested_content_sha": plan_camera_sha, "bible_content_sha": "sha256:BIBLE", }, builder="test", ) def _all_fresh() -> Path: _write_script("INT. LAB - DAY\nA test scene plays out.\n") plan_path = _write_plan() _stamp_plan_chain() assert derivation_manifest.freshness(PROJECT, EPISODE, "plan") == (True, None) return plan_path def _base_args(*extra: str) -> list[str]: return [ "--project", PROJECT, "--episode", str(EPISODE), *extra, ] def _install_spies( monkeypatch: pytest.MonkeyPatch, *, storyboard_writes_plan: bool = False, ) -> dict[str, list]: calls: dict[str, list] = { "camera_test": [], "load_bible": [], "storyboard": [], "gate": [], "batches": [], } def fake_camera_test(self, episode_num: int): # noqa: ANN001 calls["camera_test"].append(episode_num) def fake_load_bible(self): # noqa: ANN001 calls["load_bible"].append(True) return {"stub": "bible"} def fake_storyboard(self, episode: int, bible): # noqa: ANN001 calls["storyboard"].append((episode, bible)) if storyboard_writes_plan: (self.plans_dir / "ep_001_plan.json").write_text( json.dumps(_plan_payload(), indent=2), encoding="utf-8", ) return {"sentinel": "plan"} def fake_gate(project: str, episode: int) -> int: calls["gate"].append((project, episode)) return generate.EXIT_OK async def fake_batches(self, canonical_plan, **kwargs): # noqa: ANN001 calls["batches"].append((canonical_plan, kwargs)) return {"written": [], "skipped": []} class FakeLearningEngine: def __init__(self, project: str) -> None: self.project = project def ingest_retry(self, **kwargs) -> None: # noqa: ANN001 calls.setdefault("learning", []).append(kwargs) def flush(self) -> None: calls.setdefault("learning_flush", []).append(True) monkeypatch.setattr(generate.IngestPipeline, "run_camera_test", fake_camera_test) monkeypatch.setattr(generate.IngestPipeline, "_load_bible", fake_load_bible) monkeypatch.setattr(generate.IngestPipeline, "run_storyboard_pass", fake_storyboard) monkeypatch.setattr(generate, "_run_gate", fake_gate) monkeypatch.setattr(generate.EpisodeRunner, "run_episode_batches", fake_batches) monkeypatch.setattr(generate, "LearningEngine", FakeLearningEngine) return calls def _install_raise_spies(monkeypatch: pytest.MonkeyPatch) -> None: def fail(*args, **kwargs): # noqa: ANN001 raise AssertionError("derive entry point must not run") async def fail_async(*args, **kwargs): # noqa: ANN001 raise AssertionError("derive entry point must not run") monkeypatch.setattr(generate.IngestPipeline, "run_camera_test", fail) monkeypatch.setattr(generate.IngestPipeline, "run_storyboard_pass", fail) monkeypatch.setattr(generate, "_run_gate", fail) monkeypatch.setattr(generate.EpisodeRunner, "run_episode_batches", fail_async) def test_reuse_plan_cache_fresh_skips_plan_but_runs_extract( monkeypatch: pytest.MonkeyPatch, ) -> None: plan_path = _all_fresh() before = plan_path.read_bytes() calls = _install_spies(monkeypatch) code = generate._run_rederive_cli(_base_args("--reuse-plan-cache")) assert code == generate.EXIT_OK assert calls["camera_test"] == [] assert calls["storyboard"] == [] assert calls["gate"] == [(PROJECT, EPISODE)] assert plan_path.read_bytes() == before def test_reuse_plan_cache_dry_run_omits_default_regenerate_warning( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: _all_fresh() calls = _install_spies(monkeypatch) code = generate._run_rederive_cli(_base_args("--reuse-plan-cache", "--dry-run")) captured = capsys.readouterr() assert code == generate.EXIT_OK assert calls["storyboard"] == [] assert "Camera Test + Plan Pass will regenerate" not in captured.err assert "still regenerates the camera-test + plan" not in captured.out def test_reuse_plan_cache_stale_plan_fails_closed_before_derive( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: _write_script("INT. LAB - DAY\nA test scene plays out.\n") _write_plan() _stamp_plan_chain(plan_camera_sha="sha256:STALE-CAMERA") assert derivation_manifest.freshness(PROJECT, EPISODE, "plan") == (False, "plan") _install_raise_spies(monkeypatch) code = generate._run_rederive_cli(_base_args("--reuse-plan-cache")) captured = capsys.readouterr() assert code == generate.EXIT_PARTIAL assert "plan" in captured.err assert "--force-replan" in captured.err def test_reuse_plan_cache_stale_camera_test_fails_closed_before_derive( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: _write_script("INT. LAB - DAY\nOriginal script text.\n") _write_plan() old_sha = episode_script.episode_script_sha(PROJECT, EPISODE) _stamp_plan_chain(camera_script_sha=old_sha) _write_script("INT. LAB - DAY\nEdited script text with a new beat.\n") assert derivation_manifest.freshness(PROJECT, EPISODE, "plan") == ( False, "camera_tested", ) _install_raise_spies(monkeypatch) code = generate._run_rederive_cli(_base_args("--reuse-plan-cache")) captured = capsys.readouterr() assert code == generate.EXIT_PARTIAL assert "camera_tested" in captured.err assert "--force-replan" in captured.err @pytest.mark.parametrize( "flags", [ ("--reuse-plan-cache", "--force-replan"), *[ ("--force-replan", legacy) for legacy in ("--skip-camera-test", "--skip-plan", "--skip-extract") ], *[ ("--reuse-plan-cache", legacy) for legacy in ("--skip-camera-test", "--skip-plan", "--skip-extract") ], *[ ("--reuse-plan-cache", verb) for verb in ("--from-script", "--board-only", "--conform", "--revert") ], *[ ("--force-replan", verb) for verb in ("--from-script", "--board-only", "--conform", "--revert") ], ], ) def test_reuse_plan_cache_forbidden_flag_matrix(flags: tuple[str, str]) -> None: args = _base_args(*flags) if "--from-script" in flags or "--board-only" in flags: args.extend(["--batch", "EP001_CONT_001"]) if "--conform" in flags or "--revert" in flags: args.extend(["--batch", "EP001_CONT_001", "--to-version", "1"]) with pytest.raises(SystemExit): generate._run_rederive_cli(args) def test_force_replan_runs_full_derive(monkeypatch: pytest.MonkeyPatch) -> None: _write_script("INT. LAB - DAY\nA test scene plays out.\n") calls = _install_spies(monkeypatch, storyboard_writes_plan=True) code = generate._run_rederive_cli(_base_args("--force-replan")) assert code == generate.EXIT_OK assert calls["camera_test"] == [EPISODE] assert calls["load_bible"] == [True] assert calls["storyboard"] == [(EPISODE, {"stub": "bible"})] assert calls["gate"] == [(PROJECT, EPISODE)] assert len(calls["batches"]) == 1 def test_reuse_plan_cache_truthful_help( capsys: pytest.CaptureFixture[str], ) -> None: with pytest.raises(SystemExit) as exc: generate._run_rederive_cli(["--help"]) captured = capsys.readouterr() normalized = " ".join(captured.out.split()) assert exc.value.code == 0 assert "wired in a later phase" not in captured.out assert "--skip-plan" in captured.out assert "--skip-extract" in captured.out assert "DEPRECATED" in captured.out assert "currency" in captured.out assert "--reuse-plan-cache" in captured.out assert "Reuse the on-disk plan cache WITHOUT a currency check" in normalized assert "Reuse the on-disk extract cache WITHOUT a currency check" in normalized assert ( "wired in a later phase" not in Path("recoil/pipeline/cli/generate.py").read_text(encoding="utf-8") )