from __future__ import annotations from copy import deepcopy from io import BytesIO import json import pytest from PIL import Image from recoil.pipeline._lib import story_gate as sg def _write_sidecar(png_path, payload: dict) -> None: sidecar = f"{png_path}.json" with open(sidecar, "w", encoding="utf-8") as fh: json.dump(payload, fh) def test_from_sidecar_splits_live_prompt_order_and_grid(tmp_path): png = tmp_path / "EP001_CONT_004_v03.png" prompt = ( "LOOK BLOCK\n" "SCENE CONTEXT\n" "Jade saw CRYO-07 spark in the prior beat.\n" "AUTHORING\n" "1. Jade turns toward the pod.\n" "2. The pod opens.\n" ) panels = [ {"segment_id": "EP001_SH10", "setting": "A"}, {"segment_id": "EP001_SH11", "setting": "B"}, ] _write_sidecar( png, { "prompt": prompt, "panels": panels, "source_sha256": "abc123", "version": 3, "character_descriptions": {"JADE": "Lean salvager."}, }, ) packet = sg.StoryGatePacket.from_sidecar(png) assert packet.board_id == "EP001_CONT_004_v03" assert packet.board_png == png assert packet.slots == 4 assert packet.grid_cols == 2 assert packet.grid_rows == 2 assert packet.generation_prompt == prompt assert packet.scene_context == ( "SCENE CONTEXT\n" "Jade saw CRYO-07 spark in the prior beat.\n" ) assert packet.beats_text == ( "AUTHORING\n" "1. Jade turns toward the pod.\n" "2. The pod opens.\n" ) assert packet.panels == panels assert packet.source_sha256 == "abc123" assert packet.character_descriptions == {"JADE": "Lean salvager."} def test_from_sidecar_without_scene_context_keeps_beats(tmp_path): png = tmp_path / "EP001_CONT_005_v01.png" prompt = "LOOK BLOCK\nAUTHORING\n1. A single beat.\n" _write_sidecar( png, { "prompt": prompt, "panels": [{"segment_id": "EP001_SH10"}], "source_sha256": "abc123", "version": 1, }, ) packet = sg.StoryGatePacket.from_sidecar(png) assert packet.scene_context is None assert packet.beats_text == "AUTHORING\n1. A single beat.\n" def test_from_sidecar_fail_loud_on_missing_prompt(tmp_path): png = tmp_path / "EP001_CONT_005_v01.png" _write_sidecar( png, { "panels": [{"segment_id": "EP001_SH10"}], "source_sha256": "abc123", "version": 1, }, ) with pytest.raises(ValueError, match="prompt"): sg.StoryGatePacket.from_sidecar(png) def test_from_sidecar_fail_loud_on_missing_authoring_marker(tmp_path): png = tmp_path / "EP001_CONT_005_v01.png" _write_sidecar( png, { "prompt": "LOOK BLOCK\n1. A single beat.\n", "panels": [{"segment_id": "EP001_SH10"}], "source_sha256": "abc123", "version": 1, }, ) with pytest.raises(ValueError, match="AUTHORING"): sg.StoryGatePacket.from_sidecar(png) def test_crop_panels_row_major_quadrants(tmp_path): png = tmp_path / "board.png" image = Image.new("RGB", (40, 40)) quadrants = [ ((255, 0, 0), (0, 0, 20, 20)), ((0, 255, 0), (20, 0, 40, 20)), ((0, 0, 255), (0, 20, 20, 40)), ((255, 255, 0), (20, 20, 40, 40)), ] for color, box in quadrants: image.paste(color, box) image.save(png) crops = sg.crop_panels(png, 2, 2, 4) assert len(crops) == 4 for crop_bytes, (expected, _box) in zip(crops, quadrants): crop = Image.open(BytesIO(crop_bytes)).convert("RGB") assert crop.getpixel((10, 10)) == expected def _good_verdict() -> dict: return { "schema_version": sg.SCHEMA_VERSION, "judge_model": "fixture-model", "prompt_version": sg.PROMPT_VERSION, "board_id": "EP001_CONT_004_v03", "source_sha256": "abc123", "text_stageability": None, "panels": [ { "index": 1, "description": "Jade faces the pod.", "forced_checks": { name: { "passed": True, "severity": "SOFT", "confidence": 0.8, "reason": "ok", } for name in ( "depicts_beat", "spatially_possible", "eyeline_consistent", "object_of_gaze_in_frame_and_front", "causal_setup_present", ) }, "fix_hint_injectable": False, }, { "index": 2, "description": "The pod opens.", "forced_checks": { name: { "passed": True, "severity": "SOFT", "confidence": 0.7, "reason": "ok", } for name in ( "depicts_beat", "spatially_possible", "eyeline_consistent", "object_of_gaze_in_frame_and_front", "causal_setup_present", ) }, "fix_hint_injectable": False, }, ], "transitions": [ { "from": 1, "to": 2, "forced_checks": { "causal_setup_present": { "passed": True, "severity": "SOFT", "confidence": 0.7, "reason": "Panel 1 establishes the pod.", } }, "fix_hint_injectable": False, } ], "routing": { "class": "ok", "confidence": 0.9, "evidence": "No staging issue.", }, } def test_validate_verdict_accepts_known_good_fixture(): assert sg.validate_verdict(_good_verdict()) == [] def test_validate_verdict_names_bad_route_missing_description_and_confidence(): verdict = deepcopy(_good_verdict()) verdict["routing"]["class"] = "bad_route" verdict["panels"][0]["description"] = "" verdict["panels"][0]["forced_checks"]["depicts_beat"]["confidence"] = 1.2 problems = sg.validate_verdict(verdict) assert any("routing.class" in problem and "bad_route" in problem for problem in problems) assert any("description" in problem for problem in problems) assert any("confidence" in problem and "out of range" in problem for problem in problems) def test_validate_verdict_requires_transitions_for_multi_panel(): verdict = deepcopy(_good_verdict()) verdict["transitions"] = [] problems = sg.validate_verdict(verdict) assert any("transitions empty" in problem for problem in problems) def test_story_gate_mode_env_override_config_default_and_invalid(monkeypatch): monkeypatch.delenv("RECOIL_STORY_GATE", raising=False) monkeypatch.setattr(sg, "load_project_config", lambda _project: {}) assert sg.story_gate_mode("fixture") == "off" monkeypatch.setattr(sg, "load_project_config", lambda _project: {"story_gate_mode": "shadow"}) assert sg.story_gate_mode("fixture") == "shadow" monkeypatch.setenv("RECOIL_STORY_GATE", "off") monkeypatch.setattr(sg, "load_project_config", lambda _project: {"story_gate_mode": "shadow"}) assert sg.story_gate_mode("fixture") == "off" monkeypatch.setenv("RECOIL_STORY_GATE", "bogus") with pytest.raises(ValueError, match="bogus"): sg.story_gate_mode("fixture") monkeypatch.delenv("RECOIL_STORY_GATE", raising=False) monkeypatch.setattr(sg, "load_project_config", lambda _project: {"story_gate_mode": "bogus"}) with pytest.raises(ValueError, match="bogus"): sg.story_gate_mode("fixture") def test_story_gate_enforce_guard_is_explicit(): with pytest.raises(NotImplementedError, match="enforce"): sg.StoryGate("enforce")