"""Spatial validator — REC-111 PR-3 Phase 5. Covers all four consult-locked rules (unknown sublocation, stub ref, mid-batch jump, non-adjacent transition), the None-registry silence, AND the dispatch_payload wiring (BLOCK raises DispatchPayloadError; RECOIL_SPATIAL_OVERRIDE downgrades). Rules tests and the wiring test live in this ONE file so the gate never imports a known-red module. Self-contained: builds a fake decomposed-location tree under tmp_path (real >1KB PNG refs via PIL random-noise + a <1KB stub), so it depends on neither the committed project data nor the Dropbox data root. """ from __future__ import annotations import json import os import types import pytest from PIL import Image from recoil.core.paths import ProjectPaths from recoil.pipeline._lib import dispatch_payload as dp from recoil.pipeline._lib.plan_loader import CanonicalShot, CharacterEntry from recoil.pipeline._lib.prose_validator import Severity from recoil.pipeline._lib.shot_primitive import ShotPrimitive from recoil.pipeline._lib.spatial_validator import verify_spatial from recoil.pipeline._lib.sublocation_registry import location_base_dir LOCATION_ID = "int_lower_decks_maintenance_shaft" def _write_noise_png(path, size=(64, 64)): """A 64x64 RGB random-noise PNG — incompressible, so well over 1KB.""" path.parent.mkdir(parents=True, exist_ok=True) w, h = size img = Image.frombytes("RGB", (w, h), os.urandom(w * h * 3)) img.save(path, "PNG") def _primitive(segments, *, shot_id="EP001_PASS_001"): return ShotPrimitive( shot_id=shot_id, scene_index=0, shot_type="OTS", target_editorial_duration_s=6.0, intent="x", timing_segments=segments, location_id=LOCATION_ID, ) # ── rules fixture: a decomposed location with 3 real refs + 1 stub ────────── @pytest.fixture def tree(tmp_path): paths = ProjectPaths(project_root=tmp_path / "tartarus") base = location_base_dir(paths, LOCATION_ID) sublocs = base / "sublocations" for name in ("pod_platform", "shaft_lip", "the_drop"): _write_noise_png(sublocs / f"sublocation_{name}_v01.png") # A 99-byte stub masquerading as a png — fails the REC-125 ref guard. (sublocs / "sublocation_broken_v01.png").parent.mkdir(parents=True, exist_ok=True) (sublocs / "sublocation_broken_v01.png").write_bytes(b"x" * 99) registry = { "schema_version": 1, "location_id": LOCATION_ID, "sublocations": { "pod_platform": { "ref": "sublocations/sublocation_pod_platform_v01.png", "source_sha256": "a", }, "shaft_lip": { "ref": "sublocations/sublocation_shaft_lip_v01.png", "source_sha256": "b", }, "the_drop": { "ref": "sublocations/sublocation_the_drop_v01.png", "source_sha256": "c", }, "broken": { "ref": "sublocations/sublocation_broken_v01.png", "source_sha256": "d", }, }, # shaft_lip<->pod_platform and pod_platform<->the_drop are adjacent; # shaft_lip<->the_drop is NOT. "adjacency": [["shaft_lip", "pod_platform"], ["pod_platform", "the_drop"]], } (base / "location.json").write_text(json.dumps(registry)) return paths, base, registry # ── rule: None registry is silent (undecomposed) ─────────────────────────── def test_registry_none_is_silent(tree): _paths, base, _reg = tree p = _primitive([{"sublocation": "pod_platform"}]) assert verify_spatial(p, registry=None, base_dir=base) == [] # ── rule: a valid known sublocation produces nothing ──────────────────────── def test_valid_sublocation_passes(tree): _paths, base, reg = tree p = _primitive([{"sublocation": "pod_platform"}]) assert verify_spatial(p, registry=reg, base_dir=base) == [] # ── BLOCK rule: unknown sublocation ───────────────────────────────────────── def test_unknown_sublocation_blocks(tree): _paths, base, reg = tree p = _primitive([{"sublocation": "ghost_room"}]) results = verify_spatial(p, registry=reg, base_dir=base) blocks = [r for r in results if r.severity is Severity.BLOCK] assert [r.check for r in blocks] == ["spatial_unknown_sublocation"] assert "ghost_room" in blocks[0].message # ── BLOCK rule: stub establishing ref ─────────────────────────────────────── def test_stub_ref_blocks(tree): _paths, base, reg = tree p = _primitive([{"sublocation": "broken"}]) results = verify_spatial(p, registry=reg, base_dir=base) blocks = [r for r in results if r.severity is Severity.BLOCK] assert [r.check for r in blocks] == ["spatial_stub_ref"] # ── WARN rule: mid-batch jump between adjacent sublocations ───────────────── def test_midbatch_jump_warns_adjacent(tree): _paths, base, reg = tree p = _primitive([{"sublocation": "shaft_lip"}, {"sublocation": "pod_platform"}]) results = verify_spatial(p, registry=reg, base_dir=base) checks = {r.check for r in results} assert "spatial_midbatch_jump" in checks # Adjacent pair → no non-adjacent warning. assert "spatial_nonadjacent_transition" not in checks assert all(r.severity is Severity.WARN for r in results) # ── WARN rule: non-adjacent transition (also fires mid-batch jump) ────────── def test_nonadjacent_transition_warns(tree): _paths, base, reg = tree p = _primitive([{"sublocation": "shaft_lip"}, {"sublocation": "the_drop"}]) results = verify_spatial(p, registry=reg, base_dir=base) checks = {r.check for r in results} assert "spatial_midbatch_jump" in checks assert "spatial_nonadjacent_transition" in checks assert all(r.severity is Severity.WARN for r in results) # ── same sublocation across segments → no jump ────────────────────────────── def test_same_sublocation_no_jump(tree): _paths, base, reg = tree p = _primitive([{"sublocation": "pod_platform"}, {"sublocation": "pod_platform"}]) assert verify_spatial(p, registry=reg, base_dir=base) == [] # ════════════════════════════════════════════════════════════════════════════ # dispatch_payload wiring — BLOCK raises; RECOIL_SPATIAL_OVERRIDE downgrades. # Mirrors the Phase 3 authored-r2v_multi build-seam test (monkeypatch the # authored-path build call + ref collection so no model spend occurs). # ════════════════════════════════════════════════════════════════════════════ def _shot(shot_id: str, char_id: str, sublocation: str) -> CanonicalShot: raw = { "shot_id": shot_id, "scene_index": 2, "duration_s": 3.0, "shot_type": "OTS", "camera_side": "B", "screen_direction": "left-to-right", "strategy": "shot_spec", "source_text": f"{char_id.title()} moves through pressure.", "location_id": LOCATION_ID, "asset_data": {"characters": [char_id], "location_id": LOCATION_ID}, "spatial_data": {"sublocation": sublocation}, "prompt_data": { "shot_type": "OTS", "action_line": f"{char_id.title()} crosses the bay.", "emotion_line": "Breath held tight.", }, } return CanonicalShot( shot_id=shot_id, scene_index=2, sequence_id=None, pipeline="video", previs_model="gemini-3-pro-image-preview", video_model="seeddance-2.0", location_id=LOCATION_ID, characters=[CharacterEntry(char_id=char_id)], shot_type="OTS", duration_s=3.0, is_env_only=False, has_dialogue=False, aspect_ratio="9:16", raw=raw, ) def _ctx(sublocation: str) -> dp.PayloadContext: jade = _shot("EP002_SH01", "JADE", sublocation) wren = _shot("EP002_SH02", "WREN", sublocation) return dp.PayloadContext( project="tartarus", modality="r2v_multi", shot_id="EP002_PASS_009", shot=jade, batch_shots=[jade, wren], model_id="seeddance-2.0", bible={}, ) @pytest.fixture def wiring_env(tmp_path, monkeypatch): """Decomposed-location tree on disk + dispatch_payload seams monkeypatched.""" dp._project_config_cache.clear() paths = ProjectPaths(project_root=tmp_path / "tartarus") base = location_base_dir(paths, LOCATION_ID) _write_noise_png(base / "sublocations" / "sublocation_pod_platform_v01.png") registry = { "schema_version": 1, "location_id": LOCATION_ID, "sublocations": { "pod_platform": { "ref": "sublocations/sublocation_pod_platform_v01.png", "source_sha256": "a", } }, "adjacency": [], } (base / "location.json").write_text(json.dumps(registry)) # Registry loads resolve against the tmp tree (not the Dropbox data root). monkeypatch.setattr( dp, "ProjectPaths", types.SimpleNamespace(for_project=lambda _project: paths) ) # No model spend: stub the author call + ref collection + project config. monkeypatch.setattr( dp, "author_pass", lambda *a, **k: "[0:00-0:03] Jade pushes in.\n[0:03-0:06] Wren braces beside Jade.", ) monkeypatch.setattr(dp, "load_project_config", lambda _project: {}) monkeypatch.setattr( dp, "_collect_reference_images", lambda *a, **k: (["/tmp/jade.png", "/tmp/wren.png"], {"identity_1": 1, "identity_2": 2}), ) monkeypatch.delenv("RECOIL_SPATIAL_OVERRIDE", raising=False) monkeypatch.delenv("RECOIL_WORLD_STATE_PASS", raising=False) yield paths dp._project_config_cache.clear() def test_dispatch_blocks_on_unknown_sublocation(wiring_env): with pytest.raises(dp.DispatchPayloadError, match="spatial"): dp.build_unified_payload(_ctx("ghost_room")) def test_dispatch_override_downgrades_block(wiring_env, monkeypatch): monkeypatch.setenv("RECOIL_SPATIAL_OVERRIDE", "1") payload = dp.build_unified_payload(_ctx("ghost_room")) assert payload["shot_id"] == "EP002_PASS_009" def test_registry_entry_without_ref_blocks(tmp_path): """Merge-gate r2: a malformed registry entry ({} / empty ref) must BLOCK.""" registry = {"sublocations": {"pod_platform": {}}, "adjacency": []} primitive = _primitive([{"sublocation": "pod_platform"}]) results = verify_spatial(primitive, registry=registry, base_dir=tmp_path) assert any( r.severity == Severity.BLOCK and r.check == "spatial_stub_ref" for r in results )