from __future__ import annotations import copy import hashlib import json import logging from pathlib import Path import pytest from recoil.pipeline._lib import derivation_manifest, plan_overrides from recoil.pipeline._lib.plan_overrides import PlanOverridesError from recoil.pipeline.orchestrator import ingest_pipeline as ip def _camera_tested(source_text: str = "JANE studies the empty bridge.") -> ip.CameraTestedEpisode: return ip.CameraTestedEpisode.model_validate( { "episode_id": "EP001", "project": "demo", "total_shots": 1, "shots": [ { "shot_index": 1, "scene_index": 1, "source_text": source_text, "has_dialogue": False, "characters_mentioned": ["JANE"], "location_hint": "INT. BRIDGE", } ], } ) def _bible() -> ip.GlobalBible: return ip.GlobalBible.model_validate( { "project": "demo", "total_episodes": 1, "characters": { "JANE": { "char_id": "JANE", "display_name": "Jane", "visual_description": "mid 30s, focused, tired eyes", "episodes": [1], "phases": [ { "phase_id": "jane_ep1", "start_ep": 1, "end_ep": 1, "wardrobe_description": "navy work jacket and black boots", } ], } }, "locations": { "bridge": { "location_id": "bridge", "aliases": ["INT. BRIDGE"], "description": "compact command bridge with dim consoles", } }, "props": {}, } ) def _creative_output_json() -> str: return json.dumps( { "episode_id": "EP001", "total_shots": 1, "shots": [ { "shot_index": 1, "prompt_skeleton": { "subject_line": "Jane leans over the command console", "environment_line": "dim compact bridge, cool console glow", "action_line": "still frame, quiet tension", "emotion_line": "controlled worry", "motion_line": "Jane leans closer to the console as the glow pulses across her face.", }, "shot_type": "CU", "camera_movement": "static", "kinetic_action": "console glow rolls across the lens", "target_editorial_duration_s": 4, "narrative_requires_match_cut": False, "light_motivator": "console glow", } ], } ) def _shot(source_text: str, *, shot_type: str = "MS") -> dict: return { "shot_id": "EP001_SH01", "shot_index": 1, "scene_index": 1, "source_text": source_text, "routing_data": { "target_editorial_duration_s": 4, "has_dialogue": False, "num_characters": 1, "is_env_only": False, "narrative_requires_match_cut": False, }, "prompt_data": { "shot_type": shot_type, "camera_movement": "static", "focal_length": "50mm", "kinetic_action": "console glow rolls across the lens", "lighting": {"sources": [{"motivator": "console glow"}]}, "prompt_skeleton": { "subject_line": "Jane leans over the command console", "environment_line": "dim compact bridge, cool console glow", "action_line": "still frame, quiet tension", "emotion_line": "controlled worry", "motion_line": "Jane leans closer to the console as the glow pulses across her face.", }, }, "spatial_data": {}, "asset_data": { "location_id": "bridge", "characters": [ { "char_id": "JANE", "wardrobe_phase_id": "jane_ep1", "emotion_keyword": "focused", } ], }, "audio_data": {}, } def _episode_plan(project: str, source_text: str, *, shot_type: str = "MS") -> ip.EpisodePlan: return ip.EpisodePlan.model_validate( { "episode_id": "EP001", "project": project, "total_shots": 1, "shots": [_shot(source_text, shot_type=shot_type)], } ) def _source_hash(source_text: str) -> str: return hashlib.md5(source_text.encode("utf-8")).hexdigest() def _write_prior_plan( pipeline: ip.IngestPipeline, project: str, source_text: str, *, shot_updates: dict | None = None, ) -> dict: plan = _episode_plan(project, source_text).model_dump(mode="json") plan["source_hash"] = _source_hash(source_text) plan["shots"][0]["source_text_hash"] = _source_hash(source_text) if shot_updates: plan["shots"][0].update(shot_updates) path = pipeline.episodes_dir / "ep_001_plan.json" path.write_text(json.dumps(plan), encoding="utf-8") return plan def _write_overrides( project: str, target_span_hash: str, fields: dict, ) -> None: path = plan_overrides.overrides_path(project, 1) path.parent.mkdir(parents=True, exist_ok=True) path.write_text( json.dumps( { "schema_version": 1, "episode_id": "EP001", "overrides": [ { "shot_id": "EP001_SH01", "target_span_hash": target_span_hash, "fields": fields, "authored_at": "2026-06-24T00:00:00Z", } ], } ), encoding="utf-8", ) def _pipeline( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, project: str, source_text: str, ) -> ip.IngestPipeline: projects_root = tmp_path / "projects" projects_root.mkdir(exist_ok=True) monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(projects_root)) pipeline = ip.IngestPipeline(project=project, project_root=projects_root / project) monkeypatch.setattr(pipeline, "_load_camera_tested", lambda _ep: _camera_tested(source_text)) monkeypatch.setattr(pipeline, "_load_episode_script", lambda _ep: source_text) monkeypatch.setattr(pipeline, "_load_project_config", lambda: {}) monkeypatch.setattr(pipeline, "_send_to_review_queue", lambda *a, **k: None) monkeypatch.setattr(ip, "call_opus_oauth", lambda *a, **k: _creative_output_json()) monkeypatch.setattr( ip.IngestPipeline, "_assemble_plan_from_creative", lambda self, _creative, _ct, _bible, _ep: _episode_plan(project, source_text), ) derivation_manifest.stamp_stage( project, 1, "camera_tested", kind="derived", content_sha="sha256:CT", structural_sha=None, source={"script_sha": "script"}, builder="stage0.camera_test", ) derivation_manifest.stamp_bible( project, content_sha="sha256:BIBLE", builder="stage1.breakdown", built_at="2026-06-24T00:00:00+00:00", ) return pipeline def _run_and_load(pipeline: ip.IngestPipeline) -> dict: pipeline.run_storyboard_pass(1, bible=_bible()) return json.loads((pipeline.episodes_dir / "ep_001_plan.json").read_text(encoding="utf-8")) def test_fresh_override_reaches_written_plan(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: project = "_override_fresh" source_text = "JANE studies the empty bridge." pipeline = _pipeline(tmp_path, monkeypatch, project, source_text) _write_prior_plan(pipeline, project, source_text) _write_overrides( project, _source_hash(source_text), {"prompt_data": {"shot_type": "ECU"}}, ) plan = _run_and_load(pipeline) assert plan["shots"][0]["prompt_data"]["shot_type"] == "ECU" assert plan["override_flags"] == [] def test_stale_override_is_flagged_not_applied_and_plan_completes( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, ) -> None: project = "_override_stale" source_text = "JANE studies the empty bridge." pipeline = _pipeline(tmp_path, monkeypatch, project, source_text) _write_prior_plan( pipeline, project, source_text, shot_updates={"source_text_hash": "stale_hash"}, ) _write_overrides(project, "stale_hash", {"prompt_data": {"shot_type": "ECU"}}) caplog.set_level(logging.WARNING, logger=ip.__name__) plan = _run_and_load(pipeline) assert plan["shots"][0]["prompt_data"]["shot_type"] == "MS" assert plan["override_flags"] == [ { "shot_id": "EP001_SH01", "reason": "stale_span", "target_span_hash": "stale_hash", "live_hash": _source_hash(source_text), } ] assert any( "EP001_SH01" in record.message and "stale_span" in record.message for record in caplog.records ) def test_absent_overrides_file_is_noop_after_prior_plan_merge( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: project = "_override_absent" source_text = "JANE studies the empty bridge." pipeline = _pipeline(tmp_path, monkeypatch, project, source_text) _write_prior_plan( pipeline, project, source_text, shot_updates={"selected_take_id": "take_from_prior"}, ) plan = _run_and_load(pipeline) assert plan["shots"][0]["prompt_data"]["shot_type"] == "MS" assert plan["shots"][0]["selected_take_id"] == "take_from_prior" assert plan["shots"][0]["needs_review"] is False assert plan["override_flags"] == [] def test_corrupt_overrides_file_fails_loud( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: project = "_override_corrupt" source_text = "JANE studies the empty bridge." pipeline = _pipeline(tmp_path, monkeypatch, project, source_text) prior = _write_prior_plan(pipeline, project, source_text) path = plan_overrides.overrides_path(project, 1) path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(b"{bad") with pytest.raises(PlanOverridesError): pipeline.run_storyboard_pass(1, bible=_bible()) persisted = json.loads((pipeline.episodes_dir / "ep_001_plan.json").read_text(encoding="utf-8")) assert persisted == prior def test_prior_review_state_survives_with_fresh_override( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: project = "_override_review_state" source_text = "JANE studies the empty bridge." pipeline = _pipeline(tmp_path, monkeypatch, project, source_text) review_state = { "manual_prompt_override": "sentinel manual prompt", "human_approvals": {"director": True}, "review_status": "approved", "selected_take_id": "take_sentinel", } _write_prior_plan(pipeline, project, source_text, shot_updates=copy.deepcopy(review_state)) _write_overrides( project, _source_hash(source_text), {"prompt_data": {"shot_type": "ECU"}}, ) plan = _run_and_load(pipeline) shot = plan["shots"][0] for key, value in review_state.items(): assert shot[key] == value assert shot["prompt_data"]["shot_type"] == "ECU" assert plan["override_flags"] == []