"""REC-219: permanent-attachment worn props auto-inject onto present carriers.""" from __future__ import annotations import json import sys from pathlib import Path import pytest from PIL import Image _REPO_ROOT = Path(__file__).resolve().parents[4] if str(_REPO_ROOT) not in sys.path: sys.path.insert(0, str(_REPO_ROOT)) from recoil.core.paths import ProjectPaths # noqa: E402 from recoil.core.ref_types import RefAsset, ReferenceBundle # noqa: E402 from recoil.pipeline._lib import board_builder as bb # noqa: E402 from recoil.pipeline._lib import dispatch_payload as dp # noqa: E402 from recoil.pipeline._lib.board_builder import BoardBuilderError # noqa: E402 from recoil.pipeline.core.registry import MODALITY_STORYBOARD, RunResult # noqa: E402 from recoil.pipeline.core.take import Beat, Scene # noqa: E402 from recoil.pipeline.core.persistence import scene_path, save_scene # noqa: E402 _PROJECT = "fixture_project" _BATCH = "EP001_CONT_004" _PROP = "debt_counter" _DESC = "Amber-glowing wrist device strapped tight to the left wrist, LCD counter display." @pytest.fixture def board_env(tmp_path, monkeypatch): dp._project_config_cache.clear() monkeypatch.setenv("RECOIL_PROJECTS_ROOT", str(tmp_path)) paths = ProjectPaths(project_root=tmp_path / _PROJECT) paths.project_root.mkdir(parents=True) paths.project_config_path.write_text(json.dumps({}), encoding="utf-8") monkeypatch.setattr(ProjectPaths, "for_project", classmethod(lambda cls, project=None: paths)) reg = paths.assets_dir / "prop" / "prop.json" reg.parent.mkdir(parents=True, exist_ok=True) reg.write_text(json.dumps({_PROP: {}}), encoding="utf-8") prop_dir = paths.pool_dir("prop", _PROP, "identity", look="base") prop_dir.mkdir(parents=True, exist_ok=True) Image.new("RGB", (8, 8)).save(prop_dir / f"{_PROP}_identity_v01.png") char_dir = paths.project_root / "assets" / "char" / "jade" / "base" char_dir.mkdir(parents=True, exist_ok=True) front = char_dir / "jade_front_base_v01.png" profile = char_dir / "jade_profile_base_v01.png" Image.new("RGB", (8, 8)).save(front) Image.new("RGB", (8, 8)).save(profile) subloc = ( paths.asset_look_dir("loc", "int_lab", "base") / "sublocations" / "sublocation_pod_platform_v01.png" ) subloc.parent.mkdir(parents=True, exist_ok=True) Image.new("RGB", (8, 8)).save(subloc) monkeypatch.setattr( bb, "resolve_character_bundle", lambda paths, char_id, phase=None: ReferenceBundle(( RefAsset(path=front, role="identity", subject=char_id, kind="turn", view="front"), RefAsset(path=profile, role="identity", subject=char_id, kind="turn", view="profile"), )), ) monkeypatch.setattr(bb, "sublocation_ref", lambda paths, lid, name: subloc) monkeypatch.setattr(bb, "validate_ref_file", lambda path: None) monkeypatch.setattr(bb, "derive_settings", lambda segments, **_k: [dict(s, setting=s.get("intent", "")) for s in segments]) monkeypatch.setattr(bb, "select_board_model", lambda **_k: "gpt-image-2") yield paths dp._project_config_cache.clear() def _write_project_config(paths: ProjectPaths, body: dict) -> None: dp._project_config_cache.clear() paths.project_config_path.write_text(json.dumps(body), encoding="utf-8") def _write_bible(paths: ProjectPaths, *, permanent: bool = True, attached_to: str | None = "JADE") -> None: paths.global_bible_path.parent.mkdir(parents=True, exist_ok=True) paths.global_bible_path.write_text( json.dumps({ "characters": {"JADE": {"visual_description": "Lean salvager."}}, "props": { _PROP: { "attached_to": attached_to, "is_permanent_attachment": permanent, "description": _DESC, } }, }), encoding="utf-8", ) def _shot(intent: str, *, jade: bool = True, suppress_worn=None) -> dict: characters = [{"char_id": "JADE", "wardrobe_phase_id": "p1"}] if jade else [] shot = { "shot_id": "EP001_SH10", "scene_index": 1, "duration_s": 1.0, "intent": intent, "asset_data": {"characters": characters, "location_id": "int_lab"}, "spatial_data": {"sublocation": "pod_platform"}, } if suppress_worn is not None: shot["suppress_worn"] = suppress_worn return shot def _save_scene_for(paths: ProjectPaths, shot: dict) -> None: beat = Beat( beat_id="BATCH_004", beat_metadata={ "scene_id": "BATCH_004", "modality": "r2v_multi", "shot": shot, "batch_shots": [shot], "batch_summary": {"shared_characters": [], "shared_location_id": "int_lab"}, }, board=None, ) save_scene( Scene( scene_id="BATCH_004", beats=[beat], scene_metadata={"episode": "ep_001", "project": _PROJECT}, ), scene_path(_PROJECT, "ep_001", "BATCH_004"), ) def _install_fake_dispatch(monkeypatch): captured: list[dict] = [] def fake_dispatch(modality, payload, *, context): from recoil.pipeline.core.receipts import GenerationReceipt captured.append(payload) save_dir = Path(payload["save_dir"]) save_dir.mkdir(parents=True, exist_ok=True) png = save_dir / f"{payload['filename_stem']}.png" Image.new("RGB", (20, 20)).save(png) Path(f"{png}.json").write_text(json.dumps(payload["sidecar_extra"]), encoding="utf-8") return GenerationReceipt( receipt_id=f"r{len(captured)}", modality=MODALITY_STORYBOARD, caller_id="board_builder", project=_PROJECT, episode=1, shot_id=_BATCH, timestamp_utc="2026-06-20T00:00:00Z", run_result=RunResult( id=f"r{len(captured)}", modality=MODALITY_STORYBOARD, output_path=str(png), metadata={}, success=True, error=None, ), ) monkeypatch.setattr(bb, "dispatch", fake_dispatch) return captured def _build_payload(paths: ProjectPaths, monkeypatch, shot: dict) -> dict: _save_scene_for(paths, shot) captured = _install_fake_dispatch(monkeypatch) bb.build_and_dispatch_board(_PROJECT, 1, _BATCH, step_runner=object()) return captured[-1] def _assert_no_standalone_prop_ref(payload: dict) -> None: assert not any(_PROP in str(ref) for ref in payload["reference_images"]), payload["reference_images"] assert not any(entry.get("slug") == _PROP for entry in payload["sidecar_extra"].get("prop_refs", [])) def test_autoinject_undetected_permanent_prop(board_env, monkeypatch): _write_bible(board_env) payload = _build_payload(board_env, monkeypatch, _shot("Jade studies the amber readout.")) assert payload["sidecar_extra"]["worn_props"]["JADE"] == [_PROP] # REC-247: an auto-injected permanent worn prop WITH a hero ref now gets the ref # (labeled worn in the prompt) so the model can match its appearance — while the # carrier prose still emits. Suppress_worn / toggle-off keep the prose-only path. assert any(_PROP in str(ref) for ref in payload["reference_images"]) assert any(e.get("slug") == _PROP for e in payload["sidecar_extra"].get("prop_refs", [])) def test_autoinject_skips_when_carrier_absent(board_env, monkeypatch): _write_bible(board_env) payload = _build_payload(board_env, monkeypatch, _shot("An amber readout flickers.", jade=False)) assert payload["sidecar_extra"]["worn_props"] == {} def test_global_toggle_off_reverts(board_env, monkeypatch): _write_bible(board_env) _write_project_config(board_env, {"worn_prop_auto_inject": False}) payload = _build_payload(board_env, monkeypatch, _shot("Jade studies the amber readout.")) assert payload["sidecar_extra"]["worn_props"] == {} def test_per_board_suppress(board_env, monkeypatch): _write_bible(board_env) payload = _build_payload( board_env, monkeypatch, _shot("Jade studies the amber readout.", suppress_worn=[_PROP]), ) assert payload["sidecar_extra"]["worn_props"] == {} def test_non_permanent_prop_unaffected(board_env, monkeypatch): _write_bible(board_env, permanent=False) payload = _build_payload(board_env, monkeypatch, _shot("Jade studies the amber readout.")) assert payload["sidecar_extra"]["worn_props"] == {} def test_no_double_add(board_env, monkeypatch): _write_bible(board_env) payload = _build_payload(board_env, monkeypatch, _shot("Jade checks her debt counter.")) assert payload["sidecar_extra"]["worn_props"]["JADE"] == [_PROP] def test_finish_reuses_autoinjected_worn(board_env, monkeypatch): _write_bible(board_env) shot = _shot("Jade studies the amber readout.") _save_scene_for(board_env, shot) captured = _install_fake_dispatch(monkeypatch) result = bb.build_and_dispatch_board(_PROJECT, 1, _BATCH, step_runner=object()) artifact = result["artifact"] monkeypatch.setattr( bb, "resolve_board_for_spend", lambda project, episode, beat: (True, { "artifact": artifact, "source_sha256": result["source_sha256"], "fingerprint_version": result["fingerprint_version"], }), ) bb.render_board_finish( _PROJECT, 1, _BATCH, step_runner=object(), expected_version=1, ) finish_payload = captured[-1] assert _DESC.lower() in finish_payload["prompt"].lower(), finish_payload["prompt"][:800] # REC-247: the photoreal finish reuses the worn prop ref too (labeled worn), so the # counter matches the hero in the final render — the worn prose still emits above. assert any(_PROP in str(ref) for ref in finish_payload["reference_images"]) def test_toggle_off_keeps_detected_worn(board_env, monkeypatch): _write_bible(board_env) _write_project_config(board_env, {"worn_prop_auto_inject": False}) payload = _build_payload(board_env, monkeypatch, _shot("Jade checks her debt counter.")) assert payload["sidecar_extra"]["worn_props"]["JADE"] == [_PROP] _assert_no_standalone_prop_ref(payload) def test_suppress_worn_keeps_detected_worn(board_env, monkeypatch): _write_bible(board_env) payload = _build_payload( board_env, monkeypatch, _shot("Jade checks her debt counter.", suppress_worn=[_PROP]), ) assert payload["sidecar_extra"]["worn_props"]["JADE"] == [_PROP] _assert_no_standalone_prop_ref(payload) @pytest.mark.parametrize("entry", ["bogus_prop", 7]) def test_suppress_unknown_slug_fails_loud(board_env, monkeypatch, entry): _write_bible(board_env) _save_scene_for(board_env, _shot("Jade studies the amber readout.", suppress_worn=[entry])) _install_fake_dispatch(monkeypatch) with pytest.raises(BoardBuilderError, match=r"suppress_worn: unknown prop slug"): bb.build_and_dispatch_board(_PROJECT, 1, _BATCH, step_runner=object())