# recoil/pipeline/tests/test_coverage_pass_reconciliation.py """Reconciliation logic for build_coverage_passes.py. The reconciliation predicate must preserve any pass with `prompt_override` or `ref_override` populated, regardless of `status`. The slot extractor must distinguish planner-style pass_ids (where preserved entry replaces auto entry) from manually-added ids like TEST_480P_FAL (where preserved entry coexists). """ import importlib.util import json import sys from pathlib import Path import pytest from recoil.core.naming import build_filename from recoil.pipeline.tools.reconcile_grouping import find_grouping_drift # Load build_coverage_passes.py as a module without triggering its CLI main(). _BCP_PATH = ( Path(__file__).resolve().parent.parent / "tools" / "build_coverage_passes.py" ) @pytest.fixture(scope="module") def bcp(): """Import build_coverage_passes as a module for unit-testing helpers.""" # Add pipeline root to path so build_coverage_passes can resolve its imports. pipeline_root = _BCP_PATH.parent.parent if str(pipeline_root) not in sys.path: sys.path.insert(0, str(pipeline_root)) spec = importlib.util.spec_from_file_location("build_coverage_passes", _BCP_PATH) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod def test_pass_slot_strips_format_suffix(bcp): assert bcp._pass_slot("EP001_PASS_017_A_WREN_B") == "EP001_PASS_017_A_WREN" assert bcp._pass_slot("EP001_PASS_017_A_WREN") == "EP001_PASS_017_A_WREN" assert bcp._pass_slot("EP002_PASS_005_B_JADE_C") == "EP002_PASS_005_B_JADE" def test_pass_slot_returns_full_id_for_manual_entries(bcp): """Non-planner pass_ids (test fixtures, custom) get unique slots.""" assert ( bcp._pass_slot("EP001_PASS_002_TEST_480P_ATLAS") == "EP001_PASS_002_TEST_480P_ATLAS" ) assert ( bcp._pass_slot("EP001_PASS_002_TEST_480P_FAL") == "EP001_PASS_002_TEST_480P_FAL" ) # Auto and manual variants on the same shot range get DIFFERENT slots, # so reconciliation will COEXIST them rather than overwriting. auto_slot = bcp._pass_slot("EP001_PASS_002_A_JADE_B") manual_slot = bcp._pass_slot("EP001_PASS_002_TEST_480P_FAL") assert auto_slot != manual_slot def test_should_preserve_status_edited(bcp): """Status-based preservation (legacy behavior — Console-edited passes).""" p = {"status": "edited", "generation_config": {}} assert bcp._should_preserve(p) is True def test_should_preserve_status_rendered(bcp): p = {"status": "rendered", "generation_config": {}} assert bcp._should_preserve(p) is True def test_should_preserve_prompt_override_on_draft(bcp): """The fix: prompt_override on a draft pass MUST be preserved (PASS_017 case).""" p = {"status": "draft", "generation_config": {"prompt_override": "Wren leaps."}} assert bcp._should_preserve(p) is True def test_should_preserve_ref_override_on_draft(bcp): """The fix: ref_override on a draft pass MUST be preserved.""" p = { "status": "draft", "generation_config": { "ref_override": [ {"path": "output/refs/wren/hero.jpeg", "ref_type": "identity"} ], }, } assert bcp._should_preserve(p) is True def test_should_not_preserve_plain_draft(bcp): """Plain draft with no overrides — auto-generated, fine to regenerate.""" p = { "status": "draft", "generation_config": {"model": "seeddance-2.0", "mode": "r2v"}, } assert bcp._should_preserve(p) is False def test_should_not_preserve_empty_overrides(bcp): """Empty list/string for overrides should NOT count as a hand-edit.""" p = { "status": "draft", "generation_config": {"prompt_override": "", "ref_override": []}, } assert bcp._should_preserve(p) is False def test_should_not_preserve_missing_generation_config(bcp): """Missing generation_config doesn't crash; treated as no override.""" p = {"status": "draft"} assert bcp._should_preserve(p) is False def _grouping(strategy="coverage", ordinal=11, shot_ids=None): return { "strategy": strategy, "ordinal": ordinal, "shot_ids": list(shot_ids or ["EP001_SH01", "EP001_SH02"]), "source_pass_id": None, } def _project_fixture(tmp_path, monkeypatch): (tmp_path / ".recoil-data-root").write_text("recoil-data-root\n") monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(tmp_path)) project_dir = tmp_path / "fixture" project_dir.mkdir() return project_dir def _write_video(project_dir, filename): video_dir = project_dir / "renders" / "ep_001" video_dir.mkdir(parents=True, exist_ok=True) video_path = video_dir / filename video_path.write_bytes(b"fake mp4") return video_path def _write_sidecar(video_path, grouping, *, video_path_value=None): sidecar_path = video_path.with_suffix(video_path.suffix + ".json") sidecar_path.write_text( json.dumps( { "video_path": str(video_path_value or video_path), "provenance": {"grouping": grouping}, }, indent=2, ) ) return sidecar_path def _write_scene(project_dir, scene_id, output_path, grouping): scenes_dir = project_dir / "_pipeline" / "state" / "orchestration" / "scenes" scenes_dir.mkdir(parents=True, exist_ok=True) scene_path = scenes_dir / f"ep_001_{scene_id}.json" scene_path.write_text( json.dumps( { "schema_version": "test", "scene_id": scene_id, "beats": [ { "beat_id": f"{scene_id}_beat", "beat_metadata": {"grouping": grouping}, "takes": [ { "take_id": f"{scene_id}_take", "take_index": 0, "workflow": { "workflow_id": f"{scene_id}_workflow", "steps": [ { "step_id": "video", "modality": "r2v_multi", "receipt": { "run_result": { "output_path": str(output_path) } }, } ], }, } ], } ], }, indent=2, ) ) return scene_path def test_grouping_drift_clean_fixture_yields_zero_findings(tmp_path, monkeypatch): project_dir = _project_fixture(tmp_path, monkeypatch) grouping = _grouping(ordinal=11) filename = build_filename( episode=1, strategy="coverage", ordinal=11, shot_ids=grouping["shot_ids"], take=1, ) video_path = _write_video(project_dir, filename) _write_sidecar(video_path, grouping) _write_scene(project_dir, "CLEAN", video_path, grouping) assert find_grouping_drift("fixture", 1) == [] def test_output_path_drift_flags_missing_scene_receipt_artifact(tmp_path, monkeypatch): project_dir = _project_fixture(tmp_path, monkeypatch) grouping = _grouping(ordinal=11) missing = project_dir / "renders" / "ep_001" / "EP001_COV_011_SH01_02_take1.mp4" _write_scene(project_dir, "MISSING", missing, grouping) findings = find_grouping_drift("fixture", 1) assert [f.kind for f in findings] == ["missing_output_path"] assert findings[0].details["output_path"] == str(missing) def test_sidecar_drift_flags_stale_video_path(tmp_path, monkeypatch): project_dir = _project_fixture(tmp_path, monkeypatch) grouping = _grouping(ordinal=11) filename = build_filename( episode=1, strategy="coverage", ordinal=11, shot_ids=grouping["shot_ids"], take=1, ) video_path = _write_video(project_dir, filename) _write_sidecar(video_path, grouping, video_path_value=video_path.with_name("old.mp4")) findings = find_grouping_drift("fixture", 1) assert [f.kind for f in findings] == ["sidecar_video_path_mismatch"] assert findings[0].details["actual_video_path"] == str(video_path) def test_grouping_drift_flags_filename_scene_ordinal_mismatch(tmp_path, monkeypatch): project_dir = _project_fixture(tmp_path, monkeypatch) scene_grouping = _grouping(ordinal=12) filename = build_filename( episode=1, strategy="coverage", ordinal=11, shot_ids=scene_grouping["shot_ids"], take=1, ) video_path = _write_video(project_dir, filename) _write_scene(project_dir, "ORDINAL", video_path, scene_grouping) findings = find_grouping_drift("fixture", 1) assert [f.kind for f in findings] == ["filename_scene_grouping_mismatch"] assert findings[0].details["filename"]["ordinal"] == 11 assert findings[0].details["scene_grouping"]["ordinal"] == 12 def test_sidecar_grouping_drift_flags_filename_disagreement(tmp_path, monkeypatch): project_dir = _project_fixture(tmp_path, monkeypatch) filename_grouping = _grouping(ordinal=11) sidecar_grouping = _grouping(ordinal=12) filename = build_filename( episode=1, strategy="coverage", ordinal=11, shot_ids=filename_grouping["shot_ids"], take=1, ) video_path = _write_video(project_dir, filename) _write_sidecar(video_path, sidecar_grouping) findings = find_grouping_drift("fixture", 1) assert [f.kind for f in findings] == ["sidecar_grouping_mismatch"] assert findings[0].details["sidecar_grouping"]["ordinal"] == 12 def test_collapse_drift_flags_distinct_shot_sets_collapsed_to_001( tmp_path, monkeypatch, ): project_dir = _project_fixture(tmp_path, monkeypatch) first = _grouping(ordinal=1, shot_ids=["EP001_SH01", "EP001_SH02"]) second = _grouping(ordinal=1, shot_ids=["EP001_SH03", "EP001_SH04"]) first_video = _write_video( project_dir, build_filename( episode=1, strategy="coverage", ordinal=1, shot_ids=first["shot_ids"], take=1, ), ) second_video = _write_video( project_dir, build_filename( episode=1, strategy="coverage", ordinal=1, shot_ids=second["shot_ids"], take=1, ), ) _write_sidecar(first_video, first) _write_sidecar(second_video, second) _write_scene(project_dir, "COLLAPSE_A", first_video, first) _write_scene(project_dir, "COLLAPSE_B", second_video, second) findings = find_grouping_drift("fixture", 1) assert [f.kind for f in findings] == ["collapsed_ordinal_001"] assert findings[0].details["shot_sets"] == [ ["EP001_SH01", "EP001_SH02"], ["EP001_SH03", "EP001_SH04"], ]