from __future__ import annotations import logging import re from pathlib import Path import fastapi import pytest from recoil.api.schemas.board import BeatRefModel, StoryGateSummary from recoil.pipeline.core.persistence import load_scene, save_scene from recoil.pipeline.core.take import Beat, Scene from recoil.pipeline.orchestrator.batch_selector import BeatRef, resolve from recoil.workspace import readmodel as rm def _make_project(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> tuple[str, Path, Path, Path]: projects_root = tmp_path / "projects" projects_root.mkdir() (projects_root / ".recoil-data-root").write_text("recoil-data-root\n") project = "fixture" project_root = projects_root / project project_root.mkdir() (project_root / "project_config.json").write_text("{}") monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(projects_root)) scenes_dir = project_root / "_pipeline" / "state" / "orchestration" / "scenes" storyboards_dir = project_root / "prep" / "ep_001" / "storyboards" scenes_dir.mkdir(parents=True) storyboards_dir.mkdir(parents=True) return project, project_root, scenes_dir, storyboards_dir def _board(artifact: str) -> dict: return { "status": "proposed", "artifact": artifact, "source_sha256": "deadbeef", "approved_by": None, "updated_at": "2026-06-20T19:52:19Z", "story_gate": {"mode": "shadow", "confidence": 0.92}, } def _write_scene(path: Path, board: dict | None) -> Path: save_scene( Scene( scene_id="BATCH_001", beats=[Beat(beat_id="EP001_CONT_001", board=board)], ), path, ) return path def _ref(scene_path: Path, project: str = "fixture") -> BeatRef: return BeatRef( project=project, episode=1, strategy="continuity", ordinal=1, scene_id="BATCH_001", selector="EP001_CONT_001", scene_path=scene_path, ) def _has_warning(caplog: pytest.LogCaptureFixture, message: str, artifact: str) -> bool: return any( record.levelno == logging.WARNING and record.getMessage() == message and getattr(record, "selector", None) == "EP001_CONT_001" and getattr(record, "artifact", None) == artifact for record in caplog.records ) def test_get_board_controlled_fixture_is_pointer_faithful(tmp_path, monkeypatch): project, project_root, scenes_dir, storyboards_dir = _make_project(tmp_path, monkeypatch) pointer = "prep/ep_001/storyboards/EP001_CONT_001_v04.png" _write_scene(scenes_dir / "ep_001_BATCH_001.json", _board(pointer)) (storyboards_dir / "EP001_CONT_001_v04.png.json").write_text( '{"status": "candidate", "approved_by": null}' ) (storyboards_dir / "EP001_CONT_001_v05.png.json").write_text( '{"status": "candidate", "approved_by": null}' ) (project_root / pointer).write_bytes(b"png") (storyboards_dir / "EP001_CONT_001_v05.png").write_bytes(b"png") m = rm.get_board("EP001_CONT_001", project) assert m.board_artifact == pointer assert m.provenance.status == "proposed" assert m.provenance.version == 4 assert isinstance(m.provenance.gate, StoryGateSummary) assert m.provenance.gate.mode == "shadow" assert m.provenance.gate.confidence == 0.92 assert m.newer_unpointed_versions == 1 assert m.versions == [4, 5] assert m.board_artifact != "prep/ep_001/storyboards/EP001_CONT_001_v05.png" assert "prep/ep_001/storyboards/EP001_CONT_001_v04.png" in m.candidates assert "prep/ep_001/storyboards/EP001_CONT_001_v05.png" in m.candidates assert all(not c.startswith("/") for c in m.candidates) assert all(".." not in Path(c).parts for c in m.candidates) assert isinstance(m.ref, BeatRefModel) assert m.ref.selector == "EP001_CONT_001" def test_get_board_no_pointer_never_falls_back_to_latest_sidecar(tmp_path, monkeypatch): project, _project_root, scenes_dir, storyboards_dir = _make_project(tmp_path, monkeypatch) scene_path = _write_scene(scenes_dir / "ep_001_BATCH_001.json", None) (storyboards_dir / "EP001_CONT_001_v99.png.json").write_text('{"status": "candidate"}') monkeypatch.setattr(rm, "resolve", lambda any_id, project: _ref(scene_path, project)) m = rm.get_board("EP001_CONT_001", project) assert m.board_artifact is None assert m.board is None assert m.provenance is None assert m.reason == "no_board_pointer" assert m.newer_unpointed_versions == 0 def test_get_board_missing_pointed_sidecar_is_named_and_nonfatal( tmp_path, monkeypatch, caplog ): project, _project_root, scenes_dir, _storyboards_dir = _make_project(tmp_path, monkeypatch) pointer = "prep/ep_001/storyboards/EP001_CONT_001_v04.png" scene_path = _write_scene(scenes_dir / "ep_001_BATCH_001.json", _board(pointer)) monkeypatch.setattr(rm, "resolve", lambda any_id, project: _ref(scene_path, project)) with caplog.at_level(logging.WARNING): m = rm.get_board("EP001_CONT_001", project) assert m.board_artifact == pointer assert m.board is None assert m.detail_status == "pointed_sidecar_missing" assert _has_warning(caplog, "pointed_sidecar_missing", pointer) @pytest.mark.parametrize( "sidecar_body", [ "{not json", "[1, 2, 3]", ], ) def test_get_board_corrupt_pointed_sidecar_is_named_and_nonfatal( tmp_path, monkeypatch, caplog, sidecar_body ): project, project_root, scenes_dir, _storyboards_dir = _make_project(tmp_path, monkeypatch) pointer = "prep/ep_001/storyboards/EP001_CONT_001_v04.png" scene_path = _write_scene(scenes_dir / "ep_001_BATCH_001.json", _board(pointer)) (project_root / f"{pointer}.json").write_text(sidecar_body) monkeypatch.setattr(rm, "resolve", lambda any_id, project: _ref(scene_path, project)) with caplog.at_level(logging.WARNING): m = rm.get_board("EP001_CONT_001", project) assert m.board_artifact == pointer assert m.provenance.status == "proposed" assert m.board is None assert m.detail_status == "pointed_sidecar_corrupt" assert _has_warning(caplog, "pointed_sidecar_corrupt", pointer) def test_get_board_null_version_fallback_does_not_badge(tmp_path, monkeypatch): project, _project_root, scenes_dir, storyboards_dir = _make_project(tmp_path, monkeypatch) pointer = "prep/ep_001/storyboards/EP001_CONT_001.png" scene_path = _write_scene(scenes_dir / "ep_001_BATCH_001.json", _board(pointer)) (storyboards_dir / "EP001_CONT_001_v05.png.json").write_text('{"status": "candidate"}') monkeypatch.setattr(rm, "resolve", lambda any_id, project: _ref(scene_path, project)) m = rm.get_board("EP001_CONT_001", project) assert m.board_artifact == pointer assert m.provenance.version is None assert m.newer_unpointed_versions == 0 def test_readmodel_module_is_transport_free() -> None: assert not hasattr(rm, "app") assert not hasattr(rm, "router") assert not any( isinstance(value, (fastapi.FastAPI, fastapi.APIRouter)) for value in vars(rm).values() ) def _tartarus_ep001_available() -> bool: try: resolve("EP001_CONT_001", "tartarus") return True except Exception: return False @pytest.mark.skipif( not _tartarus_ep001_available(), reason="live tartarus EP001 unavailable in this environment (hermetic CI)", ) def test_get_board_live_tartarus_smoke_is_pointer_faithful(): m = rm.get_board("EP001_CONT_001", "tartarus") ref = resolve("EP001_CONT_001", "tartarus") scene = load_scene(ref.scene_path) pointer = scene.beats[0].board pointed_version = int(re.search(r"_v(\d+)(?=\.png$)", pointer["artifact"]).group(1)) assert m.board_artifact is not None assert m.board_artifact == pointer["artifact"] assert m.provenance.version == pointed_version if m.newer_unpointed_versions: assert m.provenance.version < max(m.versions)