from __future__ import annotations import json from pathlib import Path import pytest from recoil.pipeline._lib.render_schema import CameraTestedEpisode from recoil.pipeline.orchestrator import ingest_pipeline as ip def _camera_tested() -> CameraTestedEpisode: return CameraTestedEpisode.model_validate( { "episode_id": "EP001", "project": "demo", "total_shots": 1, "shots": [ { "shot_index": 1, "scene_index": 1, "source_text": "JANE studies the empty bridge.", "has_dialogue": False, "characters_mentioned": ["JANE"], "location_hint": "INT. BRIDGE", } ], } ) def _global_bible_json() -> str: return json.dumps( { "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", "description": "compact command bridge with dim consoles", } }, "props": {}, } ) @pytest.fixture() def pipeline(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> ip.IngestPipeline: # REC-164: breakdown now stamps the project-level bible record after the # write, which resolves ProjectPaths.for_project(project). Point the projects # root at a tmp dir so the conftest shim auto-creates the project and the # stamp stays hermetic (the manifest write failing FAILs the stage by design). projects_root = tmp_path / "projects" projects_root.mkdir() monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(projects_root)) pipe = object.__new__(ip.IngestPipeline) pipe.project = "demo" pipe.project_root = tmp_path pipe.dry_run = False pipe.extraction_model = "opus-4.6" pipe.camera_tested_dir = tmp_path / "camera_tested" pipe.bible_path = tmp_path / "global_bible.json" monkeypatch.setattr(pipe, "_load_camera_tested", lambda _episode_num: _camera_tested()) monkeypatch.setattr(pipe, "_load_character_bible", lambda: "") monkeypatch.setattr(pipe, "_load_project_config", lambda: {}) monkeypatch.setattr(pipe, "_load_breakdown", lambda: {}) monkeypatch.setattr(pipe, "_save_json", lambda _path, _data: None) monkeypatch.setattr(pipe, "_send_to_review_queue", lambda *_args, **_kwargs: None) return pipe def test_claude_breakdown_uses_opus_oauth_with_schema( pipeline: ip.IngestPipeline, monkeypatch: pytest.MonkeyPatch, ) -> None: calls: list[dict] = [] def fake_call_opus(model_id, system_prompt, user_prompt, *, json_schema=None, timeout=600, **_kwargs): calls.append( { "model_id": model_id, "system_prompt": system_prompt, "user_prompt": user_prompt, "json_schema": json_schema, "timeout": timeout, } ) return _global_bible_json() def fail_gemini(*_args, **_kwargs): raise AssertionError("_call_gemini should not be used for claude breakdown") monkeypatch.setattr(ip, "BREAKDOWN_MODEL", "claude-opus-4-8") monkeypatch.setattr(ip, "call_opus_oauth", fake_call_opus) monkeypatch.setattr(ip.IngestPipeline, "_call_gemini", fail_gemini) result = pipeline.run_breakdown_pass([1]) assert result is not None assert result.project == "demo" assert result.characters["JANE"].display_name == "Jane" assert calls[0]["model_id"] == "claude-opus-4-8" assert calls[0]["json_schema"] == ip.GlobalBible.model_json_schema() assert calls[0]["timeout"] == 2400 assert "strictly valid JSON matching the provided GlobalBible schema" in calls[0]["system_prompt"] def test_claude_breakdown_parses_fenced_json_response( pipeline: ip.IngestPipeline, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr(ip, "BREAKDOWN_MODEL", "claude-opus-4-8") monkeypatch.setattr( ip, "call_opus_oauth", lambda *_args, **_kwargs: f"```json\n{_global_bible_json()}\n```", ) monkeypatch.setattr( ip.IngestPipeline, "_call_gemini", lambda *_args, **_kwargs: pytest.fail("_call_gemini should not be used"), ) result = pipeline.run_breakdown_pass([1]) assert result is not None assert result.total_episodes == 1 def test_claude_breakdown_dry_run_calls_neither_model( pipeline: ip.IngestPipeline, monkeypatch: pytest.MonkeyPatch, ) -> None: pipeline.dry_run = True monkeypatch.setattr(ip, "BREAKDOWN_MODEL", "claude-opus-4-8") monkeypatch.setattr(ip, "call_opus_oauth", lambda *_args, **_kwargs: pytest.fail("call_opus_oauth called")) monkeypatch.setattr( ip.IngestPipeline, "_call_gemini", lambda *_args, **_kwargs: pytest.fail("_call_gemini called"), ) assert pipeline.run_breakdown_pass([1]) is None def test_gemini_extraction_override_still_routes_breakdown_to_gemini( pipeline: ip.IngestPipeline, monkeypatch: pytest.MonkeyPatch, ) -> None: gemini_calls: list[dict] = [] pipeline.extraction_model = "gemini-2.5-flash" def fake_gemini(self, **kwargs): gemini_calls.append(kwargs) return _global_bible_json() monkeypatch.setattr(ip, "BREAKDOWN_MODEL", "claude-opus-4-8") monkeypatch.setattr(ip, "call_opus_oauth", lambda *_args, **_kwargs: pytest.fail("call_opus_oauth called")) monkeypatch.setattr(ip.IngestPipeline, "_call_gemini", fake_gemini) result = pipeline.run_breakdown_pass([1]) assert result is not None assert gemini_calls[0]["model"] == "gemini-2.5-flash" assert gemini_calls[0]["response_schema"] == ip.GlobalBible.model_json_schema() def test_claude_breakdown_invalid_json_retries_on_opus_then_succeeds( pipeline: ip.IngestPipeline, monkeypatch: pytest.MonkeyPatch, ) -> None: responses = iter(["not json", _global_bible_json()]) calls: list[dict] = [] def fake_call_opus(model_id, system_prompt, user_prompt, *, json_schema=None, timeout=600, **_kwargs): calls.append( { "model_id": model_id, "system_prompt": system_prompt, "user_prompt": user_prompt, "json_schema": json_schema, "timeout": timeout, } ) return next(responses) monkeypatch.setattr(ip, "BREAKDOWN_MODEL", "claude-opus-4-8") monkeypatch.setattr(ip, "call_opus_oauth", fake_call_opus) monkeypatch.setattr( ip.IngestPipeline, "_call_gemini", lambda *_args, **_kwargs: pytest.fail("_call_gemini should not be used"), ) result = pipeline.run_breakdown_pass([1]) assert result is not None assert len(calls) == 2 assert calls[0]["json_schema"] == ip.GlobalBible.model_json_schema() assert calls[1]["json_schema"] == ip.GlobalBible.model_json_schema() assert calls[0]["timeout"] == 2400 assert calls[1]["timeout"] == 2400 assert "YOUR PREVIOUS OUTPUT HAD ERRORS" in calls[1]["user_prompt"]