"""W21B-wave-4 C4 (part 2): background_prompt_step v12 projection orchestration. Deterministic tests for the step-level wiring that feeds the projection-card guard: the two checkpoint→lookup builders and the inject-vs-fallback decision the step makes per bg. No LLM / VLM / image / DB — the C2 gate (card cp) and C3 router (plan cp) already ran; this exercises only the consumer glue. """ from __future__ import annotations from unittest.mock import MagicMock, patch from app.core.steps.background_prompt_step import ( BackgroundPromptStep, _build_plan_node_lookup, _build_projection_card_lookup, ) from app.modules.pipeline.background_prompt_projection import ( resolve_projection_plate_injection, ) _PLATE = "좁은 옥탑방 실내, 색바랜 벽지와 낡은 장판, 알루미늄 새시 창." def _card_cp(card_id: str = "cid_pass", state: str = "pass"): """shot_projection_card cp shape (C2): data.cards keyed ``{bg}::{shot}``.""" return { "data": { "cards": { "L05B07::S5_Shot7": { "bg_id": "L05B07", "shot_id": "S5_Shot7", "card_state": state, "card": { "card_id": card_id, "vlm_output": { "bg_plate_visible_description": _PLATE, }, }, }, # a precondition-blocked entry carries no envelope. "L09B01::S9_Shot1": { "bg_id": "L09B01", "shot_id": "S9_Shot1", "card_state": "blocked", "fallback_reason": "hard_precondition_missing", }, } } } def _plan_cp(*, card_id: str = "cid_pass", state: str = "pass"): """shot_aware_bg_render_plan cp shape (C3): per_fp → graph.nodes[].""" return { "data": { "per_fp": { "fp_l05_main": { "graph": { "nodes": [ { "bg_id": "L05B07", "anchor_shot_id": "S5_Shot7", "projection_card_id": card_id, "projection_card_hash": card_id, "projection_card_state": state, "projection_card_source_bg_id": "L05B07", "projection_card_inherited": False, }, { "bg_id": "L09B01", "anchor_shot_id": "", "projection_card_id": "", "projection_card_hash": "", "projection_card_state": "blocked", "projection_card_source_bg_id": "L09B01", "projection_card_inherited": False, }, ] } } } } } def test_build_plan_node_lookup_extracts_bg_nodes(): lookup = _build_plan_node_lookup(_plan_cp()) assert set(lookup) == {"L05B07", "L09B01"} assert lookup["L05B07"]["projection_card_state"] == "pass" assert lookup["L09B01"]["projection_card_state"] == "blocked" def test_build_plan_node_lookup_safe_on_missing_or_broken(): assert _build_plan_node_lookup(None) == {} assert _build_plan_node_lookup({}) == {} assert _build_plan_node_lookup({"data": {"per_fp": "garbage"}}) == {} assert _build_plan_node_lookup({"data": {"per_fp": {"f": {}}}}) == {} def test_build_projection_card_lookup_rekeys_by_bg_shot(): lookup = _build_projection_card_lookup(_card_cp()) assert ("L05B07", "S5_Shot7") in lookup assert ("L09B01", "S9_Shot1") in lookup assert lookup[("L05B07", "S5_Shot7")]["card_state"] == "pass" def test_build_projection_card_lookup_safe_on_missing(): assert _build_projection_card_lookup(None) == {} assert _build_projection_card_lookup({}) == {} # entry without bg_id/shot_id is skipped. cp = {"data": {"cards": {"x": {"card_state": "pass"}}}} assert _build_projection_card_lookup(cp) == {} def test_orchestration_pass_bg_injects_plate_prose(): """pass bg: lookups feed the guard → inject=True with the plate prose.""" plan_lookup = _build_plan_node_lookup(_plan_cp()) card_lookup = _build_projection_card_lookup(_card_cp()) decision = resolve_projection_plate_injection( plan_node=plan_lookup["L05B07"], card_lookup=card_lookup ) assert decision["inject"] is True assert decision["plate_prose"] == _PLATE assert decision["card_id"] == "cid_pass" def test_orchestration_blocked_bg_falls_back(): """blocked bg: guard returns inject=False with the plan state as reason.""" plan_lookup = _build_plan_node_lookup(_plan_cp()) card_lookup = _build_projection_card_lookup(_card_cp()) decision = resolve_projection_plate_injection( plan_node=plan_lookup["L09B01"], card_lookup=card_lookup ) assert decision["inject"] is False assert decision["fallback_reason"] == "blocked" assert decision["plate_prose"] == "" def test_orchestration_stale_when_plan_id_disagrees_with_card(): """plan built against an older card id → guard re-join catches it (stale).""" plan_lookup = _build_plan_node_lookup(_plan_cp(card_id="OLD_cid")) # checkpoint now carries a different (newer) card id. card_lookup = _build_projection_card_lookup(_card_cp(card_id="NEW_cid")) decision = resolve_projection_plate_injection( plan_node=plan_lookup["L05B07"], card_lookup=card_lookup ) assert decision["inject"] is False assert decision["fallback_reason"] == "projection_card_stale" # ----- _execute integration: per-bg effective prompt version routing ----- # (Codex C4 part2 Required: selector "12" must route inject=False bgs to the # real v10 pack/path, byte-identical to selector "10", not a v12 sentinel.) def _overlay(bg_id: str): return { "bg_id": bg_id, "fp_id": "fp_l05_main", "use_numbered_elements": [1], "ignore_numbered_elements": [], "base_markers_to_reference": [ { "number": 1, "label": "shared living zone", "category": "area", "position_hint": "center", "base_layer_decision": "base_structural_unit", "render_role": "structural_skeleton", "top_down_legibility": "clear", } ], "transient_markers_to_describe": [], "ignored_state_overlay_markers": [], "target_unit_marker_numbers": [1], "dominant_target_unit_marker_number": None, "clean_background_expected": False, "diagnostics": [], } def _bg(bg_id: str, shot: str): return { "bg_id": bg_id, "loc_id": bg_id[:3], "sub_location": "interior", "state_label_raw": "day interior plate", "surface_role": "interior_room", "depends_on_fp": ["fp_l05_main"], "depends_on_bg": [], "applies_to_shots": [shot], } def _cps_for_selector_12(): """cp set with a pass bg (L05B07) and a blocked/fallback bg (L09B01).""" return { "background_master_plan": { "data": { "plans": { "g1": { "status": "ok", "plan": { "backgrounds": [ _bg("L05B07", "S5_Shot7"), _bg("L09B01", "S9_Shot1"), ] }, } } } }, "floor_plan_render": { "data": { "floor_plans": { "fp_l05_main": {"status": "ok", "png_path": "/tmp/fp.png"} } } }, "floor_plan_prompt": {"data": {"floor_plans": {}}}, "floor_plan_overlay_payload": { "data": { "overlays": { "L05B07": _overlay("L05B07"), "L09B01": _overlay("L09B01"), } } }, "shot_aware_bg_render_plan": _plan_cp(), "shot_projection_card": _card_cp(), "scene_save": {"data": {"segments": []}}, "shot_validator": None, "shot_selection": None, "visual_world_rules": {"data": {"rules_text": "regional rule body"}}, "world_guide": None, } def _run_execute_capturing(selector: str, cps: dict) -> dict: """Run _execute with a given selector, capturing run_background_prompt's per-bg ``prompt_version`` + ``user_prompt`` (build runs for real).""" captured: dict = {} def fake_run(**kwargs): captured[kwargs["expected_bg_id"]] = { "prompt_version": kwargs["prompt_version"], "user_prompt": kwargs["user_prompt"], } return { "bg_id": kwargs["expected_bg_id"], "t2i_prompt": "x" * 60, "ref_guide": "g", "shot_guides": [ {"shot_id": s, "guide_text": "h"} for s in kwargs["applies_to_shots"] ], "objects_owned_by_background": ["door", "window"], } step = BackgroundPromptStep.__new__(BackgroundPromptStep) step.project_id = "p" step.episode_id = "e" step.project_config = {} step.build_opik_metadata = MagicMock(return_value={}) with patch("app.core.config.settings.background_mode", "on"), patch( "app.core.config.settings.background_prompt_version", selector ), patch.object( step, "_load_prev_checkpoint", side_effect=lambda sid: cps.get(sid) ), patch( "app.core.steps.background_prompt_step.run_background_prompt", side_effect=fake_run, ), patch( "app.core.steps.background_master_plan_step._select_scenes_for_group", return_value=([], []), ): result = step._execute() return {"result": result, "captured": captured} def test_execute_selector12_pass_bg_uses_v12_with_plate_prose(): cps = _cps_for_selector_12() out = _run_execute_capturing("12", cps) assert out["result"]["failed_count"] == 0 pass_call = out["captured"]["L05B07"] # pass bg → v12 pack + plate prose surfaced in the user_prompt. assert pass_call["prompt_version"] == "12" assert _PLATE in pass_call["user_prompt"] assert "Anchor projection plate" in pass_call["user_prompt"] # manifest diagnostic records the inject verdict (no plate prose stored). bg_entry = out["result"]["data"]["backgrounds"]["L05B07"] assert bg_entry["projection_card"]["inject"] is True assert "projection_card" in bg_entry assert "plate_prose" not in bg_entry["projection_card"] def test_execute_selector12_fallback_bg_is_byte_identical_to_v10(): cps = _cps_for_selector_12() v12 = _run_execute_capturing("12", cps) v10 = _run_execute_capturing("10", cps) fb12 = v12["captured"]["L09B01"] fb10 = v10["captured"]["L09B01"] # blocked bg under selector "12" must run the real v10 pack/path … assert fb12["prompt_version"] == "10" # … and its user_prompt is byte-identical to the selector-"10" build. assert fb12["user_prompt"] == fb10["user_prompt"] # no projection plate surface leaked into the fallback prompt. assert _PLATE not in fb12["user_prompt"] assert "Anchor projection plate" not in fb12["user_prompt"] # diagnostic still records the fallback reason for audit. fb_entry = v12["result"]["data"]["backgrounds"]["L09B01"] assert fb_entry["projection_card"]["inject"] is False assert fb_entry["projection_card"]["fallback_reason"] == "blocked"