"""W20B: shot_aware_bg_render_plan pure-function tests. Deterministic. LLM / image / VLM / DB / ImageAsset write 0. Tests use a mock ``llm_provider`` to assert the validator contract; production paths NEVER call this module's LLM seam. Covers: - assemble_planner_inputs joins dossier + geometry + overlay + master_plan + shot_staging with exact-ID, surfaces shot_readiness blockers per BG. - validators: DAG, max_refs<=2, two-refs distinct physical_space_id, same-fp, exactly-one anchor at index 0, camera membership, rationale, synthetic readback never production-clear. - llm_provider=None happy path returns structurally-empty plan with diagnostic and llm_call_count=0. - mock llm_provider happy path produces plan + validators=all-passed when readback_status='ok'; production_clear becomes True only then. """ from __future__ import annotations import pytest from app.modules.pipeline.shot_aware_bg_render_plan import ( ALLOWED_LENS_ENUMS, ALLOWED_REF_MODES, MAX_REFS_PER_BG, PLATE_PARTITION_FIELDS, REFERENCE_DAG_FIELDS, RENDER_GUIDANCE_FIELDS, TWO_REFS_DEDUP_TOKEN, ShotAwareBgRenderPlanError, assemble_planner_inputs, build_render_plan_for_fp, route_render_actions, validate_llm_output, ) from app.modules.pipeline import shot_aware_bg_render_plan as _sabrp def _render_guidance(suffix: str = "") -> dict: return { "visible_space_directive": ( "directive describing visible space" + suffix ), "camera_framing_directive": ( "directive describing camera framing" + suffix ), "subject_position_directive": ( "directive describing subject position" + suffix ), "state_cue_directive": ( "directive describing state cue" + suffix ), "negative_continuity_directive": ( "directive describing negative continuity" + suffix ), } # ─────────────────────────────── fixtures ─────────────────────────────── def _dossier_cp_data() -> dict: return { "dossiers": { "fp_a": { "fp_id": "fp_a", "fp_image_path": "/p/c/e/floor_plan_render/fp_a.png", "grid_size": [10, 10], "base_marker_inventory": [ {"number": 1, "label": "u1", "category": "area", "position_hint": "", "base_layer_decision": "base_structural_unit"}, {"number": 2, "label": "u2", "category": "area", "position_hint": "", "base_layer_decision": "base_structural_unit"}, ], "per_bg_render_facts_by_bg_id": { "L01B01": { "bg_id": "L01B01", "fp_id": "fp_a", "target_unit_marker_numbers": [1], "dominant_target_unit_marker_number": 1, "use_numbered_elements": [1], "ignore_numbered_elements": [], "base_marker_numbers_to_reference": [1], "transient_marker_numbers_to_describe": [], "ignored_state_overlay_marker_numbers": [], "clean_background_expected": True, "applies_to_shots": ["S1_Shot1"], "depends_on_bg": [], "diagnostics": [], }, "L01B02": { "bg_id": "L01B02", "fp_id": "fp_a", "target_unit_marker_numbers": [2], "dominant_target_unit_marker_number": 2, "use_numbered_elements": [2], "ignore_numbered_elements": [], "base_marker_numbers_to_reference": [2], "transient_marker_numbers_to_describe": [], "ignored_state_overlay_marker_numbers": [], "clean_background_expected": False, "applies_to_shots": ["S1_Shot2"], "depends_on_bg": ["L01B01"], "diagnostics": [], }, }, # W20E7-C: dossier surfaces the clean anchor candidate set # so build_render_plan_for_fp can pass it through to the # validator. L01B01 has clean_background_expected=True; # L01B02 does not. "anchor_selection_metadata": { "candidate_bg_ids": ["L01B01"], "selection_diagnostics": [], "selected_anchor_bg_id": None, }, } } } def _geometry_cp_data() -> dict: return { "per_fp": { "fp_a": { "fp_id": "fp_a", "readback_status": "synthetic_fixture", "geometry": { "fp_id": "fp_a", "grid_size": [10, 10], "readback_status": "synthetic_fixture", "camera_cell_candidates_per_unit": { "1": [[0, 0]], "2": [[1, 0]], }, "look_at_cell_candidates_per_unit": { "1": [[2, 0], [3, 0], [4, 0]], "2": [[2, 0], [3, 0], [4, 0]], }, "direction_vector_records": [ {"camera_unit": 1, "camera_cell": [0, 0], "look_at_unit": 2, "look_at_cell": [1, 0], "dx": 0, "dy": 1, "length_l1": 1}, {"camera_unit": 2, "camera_cell": [1, 0], "look_at_unit": 1, "look_at_cell": [0, 0], "dx": 0, "dy": -1, "length_l1": 1}, ], }, } } } def _overlay_cp_data() -> dict: return { "overlays": { "L01B01": {"bg_id": "L01B01", "fp_id": "fp_a"}, "L01B02": {"bg_id": "L01B02", "fp_id": "fp_a"}, } } def _master_plan_cp_data() -> dict: return {"plans": {}} def _shot_staging_cp_data() -> dict: return { "shots": [ {"shot_id": "S1_Shot1", "character_angles": []}, {"shot_id": "S1_Shot2", "character_angles": []}, ] } def _valid_llm_output() -> dict: """Spec-compliant graph for the fixture above.""" return { "graph": { "nodes": [ { "bg_id": "L01B01", "node_index": 0, "mode": "fp_seeded_anchor", "is_dwelling_identity_anchor": True, "rationale": "clean primary unit; identity anchor.", "reference_decision": { "selected_refs": [], "rejected_refs": [], "same_physical_space_dedup_decision": "single_ref", "why_single_ref_or_two_refs": ( "anchor mode uses no prior ref" ), "physical_space_id_per_ref": [], "same_physical_space_low_delta_candidate": False, "low_delta_reuse_target_bg_id": "", "low_delta_rationale": "", }, "camera_decision": { "camera_unit": 1, "camera_cell": [0, 0], "look_at_unit": 1, "look_at_cell": [2, 0], "lens_enum": "normal", "fov_deg": 50, "framing_notes": "anchor framing", }, "render_guidance": _render_guidance(" (node 0)"), }, { "bg_id": "L01B02", "node_index": 1, "mode": "reference_derived", "is_dwelling_identity_anchor": False, "rationale": "inherits identity from L01B01.", "reference_decision": { "selected_refs": [ { "ref_bg_id": "L01B01", "physical_space_id": "primary_unit_space", "space_description": None, } ], "rejected_refs": [], "same_physical_space_dedup_decision": "single_ref", "why_single_ref_or_two_refs": ( "only one ref needed (same dwelling identity)" ), "physical_space_id_per_ref": ["primary_unit_space"], "same_physical_space_low_delta_candidate": False, "low_delta_reuse_target_bg_id": "", "low_delta_rationale": "", }, "camera_decision": { "camera_unit": 2, "camera_cell": [1, 0], "look_at_unit": 2, "look_at_cell": [3, 0], "lens_enum": "wide", "fov_deg": 90, "framing_notes": "interior view of secondary unit", }, "render_guidance": _render_guidance(" (node 1)"), }, ] } } # ────────────────────── assemble_planner_inputs ────────────────────── def test_assemble_inputs_emits_fp_bundle_with_shot_readiness_ok(): out = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=_geometry_cp_data(), overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=_shot_staging_cp_data(), ) assert "fp_a" in out bundle = out["fp_a"] assert bundle["fp_id"] == "fp_a" assert bundle["readback_status"] == "synthetic_fixture" assert bundle["shot_readiness"]["ok"] is True assert "L01B01" in bundle["shot_readiness"]["per_bg"] assert bundle["shot_readiness"]["per_bg"]["L01B01"]["ok"] is True assert bundle["candidate_catalog"] == [] def test_assemble_inputs_marks_bg_not_ready_when_shot_missing(): sst = _shot_staging_cp_data() sst["shots"] = [s for s in sst["shots"] if s["shot_id"] != "S1_Shot2"] out = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=_geometry_cp_data(), overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=sst, ) per_bg = out["fp_a"]["shot_readiness"]["per_bg"] assert per_bg["L01B01"]["ok"] is True assert per_bg["L01B02"]["ok"] is False assert any( "S1_Shot2" in b for b in per_bg["L01B02"]["blockers"] ) def test_assemble_inputs_skips_fp_when_geometry_missing(): out = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data={"per_fp": {}}, overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=_shot_staging_cp_data(), ) assert out == {} # ───────────────────────── validators ───────────────────────────── def _validate(nodes): return validate_llm_output( fp_id="fp_a", nodes=nodes, same_fp_bg_ids=frozenset({"L01B01", "L01B02"}), geometry=_geometry_cp_data()["per_fp"]["fp_a"]["geometry"], readback_status="ok", ) def test_validator_all_pass_on_valid_output(): v = _validate(_valid_llm_output()["graph"]["nodes"]) assert v["all_validators_passed"] is True def test_validator_blocks_when_anchor_missing_or_not_first(): nodes = _valid_llm_output()["graph"]["nodes"] nodes[0]["is_dwelling_identity_anchor"] = False v = _validate(nodes) assert v["anchor_exactly_one_ok"] is False assert v["all_validators_passed"] is False # ── W20E7-C: clean anchor candidate set validator ──────────────────── def _validate_with_clean( nodes, clean_anchor_candidate_bg_ids, ): """Helper that exercises the W20E7-C clean-anchor validator branch.""" return validate_llm_output( fp_id="fp_a", nodes=nodes, same_fp_bg_ids=frozenset({"L01B01", "L01B02"}), geometry=_geometry_cp_data()["per_fp"]["fp_a"]["geometry"], readback_status="ok", clean_anchor_candidate_bg_ids=clean_anchor_candidate_bg_ids, ) def test_w20e7c_anchor_in_clean_candidate_set_passes(): """anchor bg_id present in the dossier's clean candidate surface → new validator passes; ``all_validators_passed`` stays True (other validators already green for the valid fixture). """ nodes = _valid_llm_output()["graph"]["nodes"] v = _validate_with_clean( nodes, clean_anchor_candidate_bg_ids=frozenset({"L01B01"}), ) assert v["anchor_in_clean_candidate_set_ok"] is True assert v["all_validators_passed"] is True def test_w20e7c_anchor_not_in_clean_candidate_set_fails(): """anchor bg_id NOT in the clean candidate surface → fail-closed. Diagnostic must name the offending bg_id and the available set so the operator can audit the LLM choice. """ nodes = _valid_llm_output()["graph"]["nodes"] # L01B01 is the anchor in the fixture; pass a candidate set that # deliberately omits it. v = _validate_with_clean( nodes, clean_anchor_candidate_bg_ids=frozenset({"L01B02"}), ) assert v["anchor_in_clean_candidate_set_ok"] is False assert v["all_validators_passed"] is False diag_blob = " ".join(v["diagnostics"]) assert "L01B01" in diag_blob assert "anchor_selection_metadata.candidate_bg_ids" in diag_blob def test_w20e7c_empty_clean_candidate_surface_fails(): """Empty surface → hard fail. The LLM cannot legitimately pick an anchor when the dossier surfaces zero clean candidates. """ nodes = _valid_llm_output()["graph"]["nodes"] v = _validate_with_clean( nodes, clean_anchor_candidate_bg_ids=frozenset(), ) assert v["anchor_in_clean_candidate_set_ok"] is False assert v["all_validators_passed"] is False diag_blob = " ".join(v["diagnostics"]) assert "empty" in diag_blob.lower() def test_w20e7c_validator_no_op_when_none_supplied(): """Back-compat: existing call sites pre-dating the W20E7-C wiring pass no clean_anchor_candidate_bg_ids → validator is a no-op. Production paths NEVER pass None (they always derive from dossier); this test pins the back-compat contract for unit-test call sites. """ nodes = _valid_llm_output()["graph"]["nodes"] v = validate_llm_output( fp_id="fp_a", nodes=nodes, same_fp_bg_ids=frozenset({"L01B01", "L01B02"}), geometry=_geometry_cp_data()["per_fp"]["fp_a"]["geometry"], readback_status="ok", ) assert v["anchor_in_clean_candidate_set_ok"] is True assert v["all_validators_passed"] is True def test_w20e7c_build_plan_passes_clean_candidates_from_dossier(): """``build_render_plan_for_fp`` derives the clean candidate set from the dossier's ``anchor_selection_metadata.candidate_bg_ids`` and forwards it to the validator. A mock LLM that picks an anchor outside that set produces ``status='failed'`` with the W20E7-C diagnostic. """ bundle = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=_geometry_cp_data(), overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=_shot_staging_cp_data(), )["fp_a"] # Inject an explicit clean candidate surface that omits L01B01 so # the mock LLM's anchor choice falls outside the surface. bundle["dossier"]["anchor_selection_metadata"] = { "candidate_bg_ids": ["L01B02"], "selection_diagnostics": [], "selected_anchor_bg_id": None, } bundle["readback_status"] = "ok" bundle["geometry"]["readback_status"] = "ok" def _fake_provider(**_kwargs): return _valid_llm_output() plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=bundle, llm_provider=_fake_provider, ) assert plan["shot_aware_bg_render_plan_status"] == "failed" assert plan["production_clear"] is False diag_blob = " ".join(plan["validators"]["diagnostics"]) assert "L01B01" in diag_blob assert "anchor_selection_metadata.candidate_bg_ids" in diag_blob def test_w20e7c_build_plan_fails_when_dossier_clean_candidate_empty(): """When the dossier surfaces zero clean candidates for an fp, the build_render_plan call fails the validator with the empty-surface diagnostic (no LLM-side workaround possible). """ bundle = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=_geometry_cp_data(), overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=_shot_staging_cp_data(), )["fp_a"] bundle["dossier"]["anchor_selection_metadata"] = { "candidate_bg_ids": [], "selection_diagnostics": [ "no clean_background_expected=True overlay found" ], "selected_anchor_bg_id": None, } bundle["readback_status"] = "ok" bundle["geometry"]["readback_status"] = "ok" def _fake_provider(**_kwargs): return _valid_llm_output() plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=bundle, llm_provider=_fake_provider, ) assert plan["shot_aware_bg_render_plan_status"] == "failed" assert plan["production_clear"] is False diag_blob = " ".join(plan["validators"]["diagnostics"]) assert "empty" in diag_blob.lower() def test_w20e7c_prompt_pack_system_md_declares_clean_anchor_constraint(): """The active prompt pack (``1.202605271800``) must explicitly require the anchor to be drawn from ``anchor_selection_metadata.candidate_bg_ids``. This guards against a regression where the prompt drifts away from the validator. """ import pathlib repo_root = pathlib.Path(__file__).resolve().parents[3] sys_md = ( repo_root / "prompts" / "_base" / "shot_aware_bg_render_plan" / "1.202605271800" / "system.md" ) text = sys_md.read_text(encoding="utf-8") assert "anchor_selection_metadata.candidate_bg_ids" in text assert "clean_background_expected" in text def test_validator_blocks_when_two_refs_share_physical_space_id(): nodes = _valid_llm_output()["graph"]["nodes"] # Force two refs both with the same physical_space_id. nodes[1]["mode"] = "two_refs_distinct_spaces" nodes[1]["reference_decision"]["selected_refs"] = [ {"ref_bg_id": "L01B01", "physical_space_id": "X", "space_description": None}, {"ref_bg_id": "L01B01", "physical_space_id": "X", "space_description": None}, ] nodes[1]["reference_decision"]["physical_space_id_per_ref"] = [ "X", "X", ] nodes[1]["reference_decision"][ "same_physical_space_dedup_decision" ] = TWO_REFS_DEDUP_TOKEN nodes[1]["reference_decision"]["why_single_ref_or_two_refs"] = ( "co-visible through opening" ) v = _validate(nodes) assert v["two_refs_distinct_spaces_ok"] is False def test_validator_blocks_when_two_refs_missing_dedup_token(): nodes = _valid_llm_output()["graph"]["nodes"] nodes[1]["reference_decision"]["selected_refs"] = [ {"ref_bg_id": "L01B01", "physical_space_id": "A", "space_description": None}, {"ref_bg_id": "L01B01", "physical_space_id": "B", "space_description": None}, ] nodes[1]["reference_decision"]["physical_space_id_per_ref"] = [ "A", "B", ] nodes[1]["reference_decision"][ "same_physical_space_dedup_decision" ] = "single_ref" # ← wrong token for 2 refs nodes[1]["reference_decision"]["why_single_ref_or_two_refs"] = ( "co-visible" ) v = _validate(nodes) assert v["two_refs_distinct_spaces_ok"] is False def test_validator_blocks_max_refs_per_bg_exceeded(): nodes = _valid_llm_output()["graph"]["nodes"] nodes[1]["reference_decision"]["selected_refs"] = [ {"ref_bg_id": "L01B01", "physical_space_id": "A"}, {"ref_bg_id": "L01B01", "physical_space_id": "B"}, {"ref_bg_id": "L01B01", "physical_space_id": "C"}, ] v = _validate(nodes) assert v["max_refs_per_bg_ok"] is False def test_validator_blocks_camera_cell_outside_candidates(): nodes = _valid_llm_output()["graph"]["nodes"] nodes[1]["camera_decision"]["camera_cell"] = [9, 9] # outside cand set v = _validate(nodes) assert v["camera_in_candidates_ok"] is False def test_validator_blocks_lens_enum_outside_allowed(): nodes = _valid_llm_output()["graph"]["nodes"] nodes[1]["camera_decision"]["lens_enum"] = "fish_eye" v = _validate(nodes) assert v["camera_in_candidates_ok"] is False def test_validator_blocks_self_loop_ref(): nodes = _valid_llm_output()["graph"]["nodes"] nodes[1]["reference_decision"]["selected_refs"] = [ {"ref_bg_id": "L01B02", "physical_space_id": "S"} ] nodes[1]["reference_decision"]["physical_space_id_per_ref"] = ["S"] v = _validate(nodes) assert v["dag_ok"] is False def test_validator_blocks_dangling_parent_reference(): nodes = _valid_llm_output()["graph"]["nodes"] nodes[1]["reference_decision"]["selected_refs"] = [ {"ref_bg_id": "ghost_bg", "physical_space_id": "G"} ] nodes[1]["reference_decision"]["physical_space_id_per_ref"] = ["G"] v = _validate(nodes) assert v["dag_ok"] is False or v["same_fp_only_ok"] is False def test_validator_blocks_cross_fp_reference(): nodes = _valid_llm_output()["graph"]["nodes"] nodes[1]["reference_decision"]["selected_refs"] = [ {"ref_bg_id": "L02B01", "physical_space_id": "X"} ] nodes[1]["reference_decision"]["physical_space_id_per_ref"] = ["X"] v = _validate(nodes) assert v["same_fp_only_ok"] is False def test_validator_blocks_empty_rationale(): nodes = _valid_llm_output()["graph"]["nodes"] nodes[1]["rationale"] = " " v = _validate(nodes) assert v["rationale_and_mode_ok"] is False def test_validator_blocks_unknown_mode_enum(): nodes = _valid_llm_output()["graph"]["nodes"] nodes[1]["mode"] = "unknown_mode" v = _validate(nodes) assert v["rationale_and_mode_ok"] is False def test_validator_marks_synthetic_readback_not_production_clear(): v = validate_llm_output( fp_id="fp_a", nodes=_valid_llm_output()["graph"]["nodes"], same_fp_bg_ids=frozenset({"L01B01", "L01B02"}), geometry=_geometry_cp_data()["per_fp"]["fp_a"]["geometry"], readback_status="synthetic_fixture", ) assert v["synthetic_readback_production_clear"] is False assert any( "synthetic" in d for d in v["diagnostics"] ) # ─────────── narrow patch validators (Codex W20B review #1-#5) ─────────── def test_validator_blocks_graph_with_missing_bg(): """#1: dwelling-scoped graph must include every same-fp bg.""" nodes = _valid_llm_output()["graph"]["nodes"][:1] # drop L01B02 v = _validate(nodes) assert v["graph_completeness_ok"] is False assert any("L01B02" in d for d in v["diagnostics"]) assert v["all_validators_passed"] is False def test_validator_blocks_graph_with_extra_bg(): nodes = _valid_llm_output()["graph"]["nodes"] # Inject an extra non-same-fp bg. nodes.append({ "bg_id": "L99B99", "node_index": 2, "mode": "fp_seeded_anchor", "is_dwelling_identity_anchor": False, "rationale": "stray", "reference_decision": { "selected_refs": [], "rejected_refs": [], "same_physical_space_dedup_decision": "single_ref", "why_single_ref_or_two_refs": "", "physical_space_id_per_ref": [], "same_physical_space_low_delta_candidate": False, "low_delta_reuse_target_bg_id": "", "low_delta_rationale": "", }, "camera_decision": { "camera_unit": 1, "camera_cell": [0, 0], "look_at_unit": 1, "look_at_cell": [2, 0], "lens_enum": "normal", "fov_deg": 50, "framing_notes": "", }, "render_guidance": _render_guidance(" (extra)"), }) v = _validate(nodes) assert v["graph_completeness_ok"] is False def test_validator_blocks_missing_reference_decision(): """#2: required shape — reference_decision must be a dict.""" nodes = _valid_llm_output()["graph"]["nodes"] nodes[0].pop("reference_decision") v = _validate(nodes) assert v["required_shape_ok"] is False assert v["all_validators_passed"] is False def test_validator_blocks_missing_low_delta_candidate_fields(): """W21B-w3 Commit 1: v2 schema-required reuse candidate signals must be present in production validator (schema-shape mirror).""" for fname in ( "same_physical_space_low_delta_candidate", "low_delta_reuse_target_bg_id", "low_delta_rationale", ): nodes = _valid_llm_output()["graph"]["nodes"] nodes[0]["reference_decision"].pop(fname) v = _validate(nodes) assert v["required_shape_ok"] is False, fname assert v["all_validators_passed"] is False, fname def test_validator_blocks_low_delta_candidate_wrong_type(): """W21B-w3 Commit 1: candidate flag must be bool, target/rationale must be string (type-only; empty sentinel allowed).""" nodes = _valid_llm_output()["graph"]["nodes"] nodes[0]["reference_decision"][ "same_physical_space_low_delta_candidate" ] = "true" # string, not bool v = _validate(nodes) assert v["required_shape_ok"] is False nodes = _valid_llm_output()["graph"]["nodes"] nodes[0]["reference_decision"]["low_delta_reuse_target_bg_id"] = 0 # int v = _validate(nodes) assert v["required_shape_ok"] is False def test_validator_passes_empty_sentinel_low_delta_fields(): """W21B-w3 Commit 1: candidate=false with empty-string target/rationale sentinel is the contract for non-candidate nodes — must pass shape.""" nodes = _valid_llm_output()["graph"]["nodes"] # _valid_llm_output already uses candidate=false + "" sentinels. v = _validate(nodes) assert v["required_shape_ok"] is True def test_validator_blocks_missing_camera_decision_field(): """#2: required shape — camera_decision.camera_unit must be int.""" nodes = _valid_llm_output()["graph"]["nodes"] nodes[1]["camera_decision"].pop("camera_unit") v = _validate(nodes) assert v["required_shape_ok"] is False def test_validator_blocks_camera_cell_wrong_type(): nodes = _valid_llm_output()["graph"]["nodes"] nodes[0]["camera_decision"]["camera_cell"] = [0, "x"] v = _validate(nodes) assert v["required_shape_ok"] is False def test_validator_blocks_invalid_node_index_type(): nodes = _valid_llm_output()["graph"]["nodes"] nodes[0]["node_index"] = "0" v = _validate(nodes) assert v["required_shape_ok"] is False def test_validator_blocks_two_refs_mode_with_one_ref(): """#3: mode==two_refs_distinct_spaces requires exactly 2 refs.""" nodes = _valid_llm_output()["graph"]["nodes"] nodes[1]["mode"] = "two_refs_distinct_spaces" # selected_refs is 1, mismatch with the mode. v = _validate(nodes) assert v["two_refs_distinct_spaces_ok"] is False def test_validator_blocks_reference_derived_mode_with_two_refs(): """#3: 2 selected refs require mode='two_refs_distinct_spaces'.""" nodes = _valid_llm_output()["graph"]["nodes"] nodes[1]["reference_decision"]["selected_refs"] = [ {"ref_bg_id": "L01B01", "physical_space_id": "A"}, {"ref_bg_id": "L01B01", "physical_space_id": "B"}, ] nodes[1]["reference_decision"]["physical_space_id_per_ref"] = ["A", "B"] nodes[1]["reference_decision"][ "same_physical_space_dedup_decision" ] = TWO_REFS_DEDUP_TOKEN nodes[1]["reference_decision"]["why_single_ref_or_two_refs"] = "co-visible" # mode remains 'reference_derived' which is wrong for 2 refs. v = _validate(nodes) assert v["two_refs_distinct_spaces_ok"] is False def test_validator_blocks_two_refs_share_ref_bg_id(): """#3: two selected_refs must have exact-string distinct ref_bg_id.""" nodes = _valid_llm_output()["graph"]["nodes"] nodes[1]["mode"] = "two_refs_distinct_spaces" nodes[1]["reference_decision"]["selected_refs"] = [ {"ref_bg_id": "L01B01", "physical_space_id": "A"}, {"ref_bg_id": "L01B01", "physical_space_id": "B"}, # same ref_bg_id ] nodes[1]["reference_decision"]["physical_space_id_per_ref"] = ["A", "B"] nodes[1]["reference_decision"][ "same_physical_space_dedup_decision" ] = TWO_REFS_DEDUP_TOKEN nodes[1]["reference_decision"]["why_single_ref_or_two_refs"] = "x" v = _validate(nodes) assert v["two_refs_distinct_spaces_ok"] is False def test_validator_blocks_psi_length_mismatch_under_single_ref(): """#3: psi length must equal selected_refs length for ALL counts.""" nodes = _valid_llm_output()["graph"]["nodes"] # node[1] has 1 selected_ref; force psi list of length 2. nodes[1]["reference_decision"]["physical_space_id_per_ref"] = ["A", "B"] v = _validate(nodes) assert v["two_refs_distinct_spaces_ok"] is False def test_validator_blocks_selected_refs_psi_exact_id_mismatch(): """#3: selected_refs[i].physical_space_id must equal psi[i] exact.""" nodes = _valid_llm_output()["graph"]["nodes"] nodes[1]["reference_decision"]["selected_refs"] = [ {"ref_bg_id": "L01B01", "physical_space_id": "A"} ] nodes[1]["reference_decision"]["physical_space_id_per_ref"] = ["B"] # mismatch v = _validate(nodes) assert v["two_refs_distinct_spaces_ok"] is False def test_validator_blocks_fov_deg_mismatch_with_lens(): """#4: fov_deg must match the geometry-derived lens→fov map. Fixture geometry has no view_cone_records → fallback map used: wide=90 / normal=50 / telephoto=25. """ nodes = _valid_llm_output()["graph"]["nodes"] nodes[1]["camera_decision"]["lens_enum"] = "normal" nodes[1]["camera_decision"]["fov_deg"] = 90 # wrong for normal v = _validate(nodes) assert v["camera_in_candidates_ok"] is False def test_validator_uses_geometry_view_cone_records_when_present(): """#4: when view_cone_records carry custom fov values, those win.""" from app.modules.pipeline.shot_aware_bg_render_plan import ( validate_llm_output as _v, ) geometry = dict(_geometry_cp_data()["per_fp"]["fp_a"]["geometry"]) # Inject custom records — normal lens fov = 60 (not default 50). geometry["view_cone_records"] = [ {"camera_unit": 1, "camera_cell": [0, 0], "look_at_unit": 2, "look_at_cell": [1, 0], "dx": 0, "dy": 1, "lens_enum": "normal", "fov_deg": 60}, {"camera_unit": 1, "camera_cell": [0, 0], "look_at_unit": 2, "look_at_cell": [1, 0], "dx": 0, "dy": 1, "lens_enum": "wide", "fov_deg": 90}, ] nodes = _valid_llm_output()["graph"]["nodes"] # node[0] used normal/50 — must now fail (50 != 60). result = _v( fp_id="fp_a", nodes=nodes, same_fp_bg_ids=frozenset({"L01B01", "L01B02"}), geometry=geometry, readback_status="ok", ) assert result["camera_in_candidates_ok"] is False def test_validator_blocks_geometry_conflicting_fov(): """#4: geometry self-conflict (same lens, two different fov values) fails closed via the derived map.""" from app.modules.pipeline.shot_aware_bg_render_plan import ( validate_llm_output as _v, ) geometry = dict(_geometry_cp_data()["per_fp"]["fp_a"]["geometry"]) geometry["view_cone_records"] = [ {"camera_unit": 1, "camera_cell": [0, 0], "look_at_unit": 2, "look_at_cell": [1, 0], "dx": 0, "dy": 1, "lens_enum": "normal", "fov_deg": 50}, {"camera_unit": 2, "camera_cell": [1, 0], "look_at_unit": 1, "look_at_cell": [0, 0], "dx": 0, "dy": -1, "lens_enum": "normal", "fov_deg": 60}, # conflict ] result = _v( fp_id="fp_a", nodes=_valid_llm_output()["graph"]["nodes"], same_fp_bg_ids=frozenset({"L01B01", "L01B02"}), geometry=geometry, readback_status="ok", ) assert result["camera_in_candidates_ok"] is False assert any("conflicting" in d for d in result["diagnostics"]) def test_validator_blocks_node_index_swap(): """#5: node_index must equal the entry's list position.""" nodes = _valid_llm_output()["graph"]["nodes"] nodes[0]["node_index"] = 1 nodes[1]["node_index"] = 0 v = _validate(nodes) assert v["node_index_order_ok"] is False def test_validator_blocks_duplicate_node_index(): nodes = _valid_llm_output()["graph"]["nodes"] nodes[1]["node_index"] = 0 # same as node[0] v = _validate(nodes) assert v["node_index_order_ok"] is False # ─────────────────── build_render_plan_for_fp ─────────────────── def test_build_plan_with_no_llm_provider_returns_empty_plan_zero_llm(): bundle = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=_geometry_cp_data(), overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=_shot_staging_cp_data(), )["fp_a"] plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=bundle, llm_provider=None ) assert plan["graph"]["nodes"] == [] assert plan["real_api_call_counts"] == { "image": 0, "llm": 0, "vlm": 0 } assert plan["production_clear"] is False assert plan["shot_aware_bg_render_plan_status"] == "not_applicable" def test_build_plan_with_mock_llm_passes_validators_for_valid_output(): bundle = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=_geometry_cp_data(), overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=_shot_staging_cp_data(), )["fp_a"] # Force readback_status='ok' (a real VLM would produce this). bundle["readback_status"] = "ok" def provider(**_kwargs): return _valid_llm_output() plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=bundle, llm_provider=provider ) assert plan["shot_aware_bg_render_plan_status"] == "ok" assert plan["validators"]["all_validators_passed"] is True assert plan["production_clear"] is True assert plan["real_api_call_counts"]["llm"] == 1 def test_build_plan_synthetic_readback_never_production_clear(): bundle = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=_geometry_cp_data(), overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=_shot_staging_cp_data(), )["fp_a"] # readback_status='synthetic_fixture' is the default in fixture. assert bundle["readback_status"] == "synthetic_fixture" def provider(**_kwargs): return _valid_llm_output() plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=bundle, llm_provider=provider ) # validators may still pass structurally; production_clear MUST be False. assert plan["production_clear"] is False def test_build_plan_default_stamps_not_available_projection_fields(): """W21B-w4 C3: with no projection_card_index (card subsystem OFF / default), every plan node still carries the additive projection-card fields stamped ``not_available`` — existing plan semantics unchanged.""" bundle = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=_geometry_cp_data(), overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=_shot_staging_cp_data(), )["fp_a"] bundle["readback_status"] = "ok" plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=bundle, llm_provider=lambda **_k: _valid_llm_output() ) nodes = plan["graph"]["nodes"] assert nodes for n in nodes: assert n["projection_card_state"] == "not_available" assert n["projection_card_source_bg_id"] == n["bg_id"] # canonical render_action fields untouched (additive enrichment). assert n["render_action"] in ("render_new_plate", "reuse_existing_plate") def test_build_plan_routes_through_zone_map_when_usable(): """W21B Phase 2 wiring: a usable dwelling zone_plan routes the ⑧ canonical surface through the one-plate-per-zone helper (render_action_source= dwelling_zone_map) instead of the partition normalizer. The byte-identical no-zone_plan path is pinned by every other build test (zone_plan default None). Consolidation semantics (many bgs → one zone plate) are covered by the pure apply_dwelling_zone_map_plan unit tests; this asserts the wiring.""" bundle = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=_geometry_cp_data(), overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=_shot_staging_cp_data(), )["fp_a"] bundle["readback_status"] = "ok" zone_plan = { "fp_id": "fp_a", "synthetic": False, "bg_zone_assignments": { "L01B01": {"zone_id": "Z01", "confidence": 0.9}, "L01B02": {"zone_id": "Z02", "confidence": 0.9}, }, } plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=bundle, llm_provider=lambda **_k: _valid_llm_output(), zone_plan=zone_plan, ) nodes = plan["graph"]["nodes"] assert nodes for n in nodes: assert n["render_action_source"] == "dwelling_zone_map" # each bg alone in its zone → its own anchor plate; alias, never derive. assert n["zone_plate_bg_id"] == n["bg_id"] assert n["ref_tree_parents"] == [] def test_build_plan_zone_map_surfaces_no_clean_diagnostic_in_validators(): """W21B Phase 2 (B2): the zone path's no-clean-candidate diagnostic MUST reach the final plan validators (visual gate), not vanish after the ⑧ summary copy. dossier candidate_bg_ids = [L01B01] (Z01 clean), so Z02 (L01B02) has no clean candidate → a diagnostic must surface.""" bundle = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=_geometry_cp_data(), overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=_shot_staging_cp_data(), )["fp_a"] bundle["readback_status"] = "ok" zone_plan = { "fp_id": "fp_a", "synthetic": False, "bg_zone_assignments": { "L01B01": {"zone_id": "Z01", "confidence": 0.9}, "L01B02": {"zone_id": "Z02", "confidence": 0.9}, }, } plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=bundle, llm_provider=lambda **_k: _valid_llm_output(), zone_plan=zone_plan, ) v = plan["validators"] assert v.get("zone_map_applied") is True assert "zone_anchor_no_clean_candidate:Z02" in (v.get("zone_anchor_diagnostics") or []) # also surfaced in the human-facing diagnostics blob. assert "zone_anchor_no_clean_candidate:Z02" in v.get("diagnostics", []) def test_build_plan_consumes_projection_card_index_when_provided(): """W21B-w4 C3: a supplied projection_card_index stamps the per-node card state/provenance after render-action routing.""" bundle = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=_geometry_cp_data(), overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=_shot_staging_cp_data(), )["fp_a"] bundle["readback_status"] = "ok" card_index = { "L01B01": {"S1_Shot1": {"card_state": "pass", "card_id": "c1", "fallback_reason": ""}}, "L01B02": {"S1_Shot2": {"card_state": "needs_review", "card_id": "c2", "fallback_reason": "low_confidence"}}, } plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=bundle, llm_provider=lambda **_k: _valid_llm_output(), projection_card_index=card_index, ) by_bg = {n["bg_id"]: n for n in plan["graph"]["nodes"]} assert by_bg["L01B01"]["projection_card_state"] == "pass" assert by_bg["L01B01"]["projection_card_id"] == "c1" assert by_bg["L01B01"]["anchor_shot_id"] == "S1_Shot1" def test_build_plan_marks_not_applicable_when_no_shot_readiness(): """W20E6-B nano: an FP with zero staged consuming shots across every BG is ``not_applicable`` -- not ``failed``. No upstream contract was violated; the LLM planner simply has nothing to plan for. Provider must not be invoked. """ bundle = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=_geometry_cp_data(), overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data={"shots": []}, # nothing committed )["fp_a"] plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=bundle, llm_provider=lambda **_kw: _valid_llm_output() ) assert plan["shot_aware_bg_render_plan_status"] == "not_applicable" assert plan["graph"]["nodes"] == [] assert plan["real_api_call_counts"]["llm"] == 0 def test_build_plan_records_llm_failure_without_raising(): bundle = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=_geometry_cp_data(), overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=_shot_staging_cp_data(), )["fp_a"] def bad_provider(**_kw): raise RuntimeError("upstream LLM down") plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=bundle, llm_provider=bad_provider ) assert plan["shot_aware_bg_render_plan_status"] == "failed" assert plan["real_api_call_counts"]["llm"] == 1 # attempted once # ─────────────────────── boundary surface ─────────────────────── def test_module_constants_are_locked(): assert MAX_REFS_PER_BG == 2 # W21B-w3 Commit 1 (2026-05-29): legacy style_reference_new_space 제거 + # same_physical_space_view / related_style_new_space 추가 (v2 pack 정렬). assert ALLOWED_REF_MODES == frozenset({ "fp_seeded_anchor", "reference_derived", "same_physical_space_view", "related_style_new_space", "two_refs_distinct_spaces", }) assert ALLOWED_LENS_ENUMS == frozenset({"wide", "normal", "telephoto"}) assert TWO_REFS_DEDUP_TOKEN == "distinct_visible_spaces" assert RENDER_GUIDANCE_FIELDS == ( "visible_space_directive", "camera_framing_directive", "subject_position_directive", "state_cue_directive", "negative_continuity_directive", ) # ───────────── W20D render_guidance required-shape validators ───────────── def test_validator_blocks_missing_render_guidance(): """W20D: required shape — render_guidance must be a dict.""" nodes = _valid_llm_output()["graph"]["nodes"] nodes[0].pop("render_guidance") v = _validate(nodes) assert v["required_shape_ok"] is False assert v["all_validators_passed"] is False assert any( "render_guidance" in d for d in v["diagnostics"] ) def test_validator_blocks_render_guidance_not_dict(): nodes = _valid_llm_output()["graph"]["nodes"] nodes[0]["render_guidance"] = "not a dict" v = _validate(nodes) assert v["required_shape_ok"] is False @pytest.mark.parametrize("field", list(RENDER_GUIDANCE_FIELDS)) def test_validator_blocks_render_guidance_field_missing(field): nodes = _valid_llm_output()["graph"]["nodes"] nodes[0]["render_guidance"].pop(field) v = _validate(nodes) assert v["required_shape_ok"] is False assert any(field in d for d in v["diagnostics"]) @pytest.mark.parametrize("field", list(RENDER_GUIDANCE_FIELDS)) def test_validator_blocks_render_guidance_field_empty_string(field): nodes = _valid_llm_output()["graph"]["nodes"] nodes[0]["render_guidance"][field] = " " v = _validate(nodes) assert v["required_shape_ok"] is False assert any(field in d for d in v["diagnostics"]) @pytest.mark.parametrize("field", list(RENDER_GUIDANCE_FIELDS)) def test_validator_blocks_render_guidance_field_wrong_type(field): nodes = _valid_llm_output()["graph"]["nodes"] nodes[0]["render_guidance"][field] = 123 v = _validate(nodes) assert v["required_shape_ok"] is False def test_validator_blocks_render_guidance_unknown_key(): nodes = _valid_llm_output()["graph"]["nodes"] nodes[0]["render_guidance"]["wild_directive"] = "x" v = _validate(nodes) assert v["required_shape_ok"] is False assert any("unknown keys" in d for d in v["diagnostics"]) def test_validator_passes_when_render_guidance_present_on_every_node(): """Sanity: the full fixture with render_guidance on every node clears the required-shape validator (also clears every other validator with readback_status='ok').""" nodes = _valid_llm_output()["graph"]["nodes"] v = _validate(nodes) assert v["required_shape_ok"] is True assert v["all_validators_passed"] is True # ───────────── W20E6-B: staged-shot-subset readiness contract ───────────── def _dossier_with_partially_unstaged_bg() -> dict: """A dossier where L01B01 has 2 applies_to_shots (one staged, one not). The other BG L01B02 keeps its single staged shot. """ d = _dossier_cp_data() bg1 = d["dossiers"]["fp_a"]["per_bg_render_facts_by_bg_id"]["L01B01"] bg1["applies_to_shots"] = ["S1_Shot1", "S1_Shot_unstaged"] return d def test_assemble_inputs_does_not_block_bg_with_at_least_one_staged_shot(): """W20E6-B: a BG with one staged + one unstaged shot must remain renderable. The unstaged shot surfaces as ``omitted_unstaged_shots``, NOT as a structural blocker. """ out = assemble_planner_inputs( dossier_cp_data=_dossier_with_partially_unstaged_bg(), geometry_cp_data=_geometry_cp_data(), overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=_shot_staging_cp_data(), ) bg_entry = out["fp_a"]["shot_readiness"]["per_bg"]["L01B01"] assert bg_entry["ok"] is True assert bg_entry["blockers"] == [] assert "S1_Shot_unstaged" in bg_entry["omitted_unstaged_shots"] surfaced_ids = [s["shot_id"] for s in bg_entry["surfaced_shots"]] assert surfaced_ids == ["S1_Shot1"] def test_assemble_inputs_emits_renderable_bg_ids_subset(): """W20E6-B: FP-level shot_readiness exposes the staged-shot renderable subset for downstream graph completeness.""" sst = _shot_staging_cp_data() sst["shots"] = [s for s in sst["shots"] if s["shot_id"] != "S1_Shot2"] out = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=_geometry_cp_data(), overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=sst, ) readiness = out["fp_a"]["shot_readiness"] assert readiness["renderable_bg_ids"] == ["L01B01"] assert readiness["per_bg"]["L01B02"]["omitted_unstaged_shots"] == [ "S1_Shot2" ] def test_assemble_inputs_marks_bg_with_zero_staged_shots_not_renderable(): """W20E6-B: a BG whose every applies_to_shot is missing from staging is not renderable; its omitted_unstaged_shots holds the full list.""" sst = _shot_staging_cp_data() sst["shots"] = [s for s in sst["shots"] if s["shot_id"] != "S1_Shot2"] out = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=_geometry_cp_data(), overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=sst, ) entry = out["fp_a"]["shot_readiness"]["per_bg"]["L01B02"] assert entry["ok"] is False assert entry["omitted_unstaged_shots"] == ["S1_Shot2"] assert any( "no staged consuming shots" in b for b in entry["blockers"] ) def test_validator_completeness_uses_renderable_bg_ids_subset(): """W20E6-B: graph completeness must validate against the renderable subset. A graph that emits only the renderable BG's node must pass completeness (the omitted non-renderable BG is NOT a missing entry). """ nodes = [_valid_llm_output()["graph"]["nodes"][0]] v = validate_llm_output( fp_id="fp_a", nodes=nodes, same_fp_bg_ids=frozenset({"L01B01", "L01B02"}), renderable_bg_ids=frozenset({"L01B01"}), geometry=_geometry_cp_data()["per_fp"]["fp_a"]["geometry"], readback_status="ok", ) assert v["graph_completeness_ok"] is True def test_validator_completeness_blocks_node_for_non_renderable_bg(): """W20E6-B: emitting a graph node for a BG outside the renderable set is a hard fail (LLM contract violation). """ nodes = _valid_llm_output()["graph"]["nodes"] v = validate_llm_output( fp_id="fp_a", nodes=nodes, same_fp_bg_ids=frozenset({"L01B01", "L01B02"}), renderable_bg_ids=frozenset({"L01B01"}), geometry=_geometry_cp_data()["per_fp"]["fp_a"]["geometry"], readback_status="ok", ) assert v["graph_completeness_ok"] is False assert any("extra bg_ids" in d for d in v["diagnostics"]) def test_validator_renderable_bg_ids_omitted_falls_back_to_same_fp(): """W20E6-B backward compatibility: when renderable_bg_ids is not provided, completeness falls back to same_fp_bg_ids (preserves legacy callers). """ nodes = _valid_llm_output()["graph"]["nodes"] v = validate_llm_output( fp_id="fp_a", nodes=nodes, same_fp_bg_ids=frozenset({"L01B01", "L01B02"}), geometry=_geometry_cp_data()["per_fp"]["fp_a"]["geometry"], readback_status="ok", ) assert v["graph_completeness_ok"] is True def test_build_plan_skips_provider_when_no_renderable_bg(): """W20E6-B + nano: an FP whose every BG is missing every staged shot is ``not_applicable``. The LLM provider must never be invoked and the result must not contribute to the step-level failed_count. """ bundle = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=_geometry_cp_data(), overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data={"shots": []}, # nothing committed )["fp_a"] invocations = {"count": 0} def provider(**_kw): invocations["count"] += 1 return _valid_llm_output() plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=bundle, llm_provider=provider ) assert plan["shot_aware_bg_render_plan_status"] == "not_applicable" assert plan["real_api_call_counts"]["llm"] == 0 assert invocations["count"] == 0 def test_build_plan_renderable_subset_passes_validators(): """W20E6-B integration: when only L01B01 is renderable, a graph containing just L01B01's node clears completeness and the plan status is 'ok' (readback_status='ok' supplied so production_clear can flip True downstream). """ geom = _geometry_cp_data() geom["per_fp"]["fp_a"]["readback_status"] = "ok" geom["per_fp"]["fp_a"]["geometry"]["readback_status"] = "ok" sst = _shot_staging_cp_data() sst["shots"] = [s for s in sst["shots"] if s["shot_id"] != "S1_Shot2"] bundle = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=geom, overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=sst, )["fp_a"] assert bundle["shot_readiness"]["renderable_bg_ids"] == ["L01B01"] def provider(**_kw): only_anchor = _valid_llm_output()["graph"]["nodes"][0] return {"graph": {"nodes": [only_anchor]}} plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=bundle, llm_provider=provider ) assert plan["shot_aware_bg_render_plan_status"] == "ok" assert plan["validators"]["graph_completeness_ok"] is True # ───────────── Commit 2a — render_action router (W21B-w3) ───────────── # # Deterministic router that assigns canonical ``render_action`` / # ``reuse_target_bg_id`` after LLM-output validation. reuse requires BOTH # the LLM low-delta candidate signal AND deterministic same-space # corroboration (camera_unit + look_at_unit / target-marker overlap); # neither alone may trigger reuse. Tested as a pure unit (no LLM). def _rnode( bg_id, node_index, mode, *, anchor=False, candidate=False, target="", ld_rationale="", camera_unit=1, look_at_unit=1, selected_refs=None, ): """Minimal node carrying exactly the fields the router reads.""" if selected_refs is None: # Eligible modes carry a ref by nature; forbidden modes default # to empty so HF4 (eligible + empty refs) does not mask intent. selected_refs = ( [{"ref_bg_id": "ref", "physical_space_id": "S"}] if mode in ("same_physical_space_view", "reference_derived") else [] ) return { "bg_id": bg_id, "node_index": node_index, "mode": mode, "is_dwelling_identity_anchor": anchor, "rationale": "r", "reference_decision": { "selected_refs": selected_refs, "rejected_refs": [], "same_physical_space_dedup_decision": "single_ref", "why_single_ref_or_two_refs": "", "physical_space_id_per_ref": [], "same_physical_space_low_delta_candidate": candidate, "low_delta_reuse_target_bg_id": target, "low_delta_rationale": ld_rationale, }, "camera_decision": { "camera_unit": camera_unit, "camera_cell": [0, 0], "look_at_unit": look_at_unit, "look_at_cell": [2, 0], "lens_enum": "normal", "fov_deg": 50, "framing_notes": "", }, "render_guidance": _render_guidance(), } def _facts(**bg_to_markers): """Build a dossier carrying only per_bg target-unit markers. Usage: _facts(L01B01=[1], L01B02=[1]) → markers + dominant derived. """ pbf = {} for bg_id, markers in bg_to_markers.items(): pbf[bg_id] = { "bg_id": bg_id, "target_unit_marker_numbers": list(markers), "dominant_target_unit_marker_number": ( markers[0] if len(markers) == 1 else None ), } return {"per_bg_render_facts_by_bg_id": pbf} def _route(nodes, dossier): return route_render_actions(nodes=nodes, dossier=dossier) def test_router_reuse_when_candidate_and_corroborated(): """Case 1: candidate true + valid earlier render_new_plate target + same camera_unit + marker overlap → reuse_existing_plate.""" nodes = [ _rnode("A", 0, "fp_seeded_anchor", anchor=True, camera_unit=1), _rnode( "B", 1, "reference_derived", candidate=True, target="A", ld_rationale="same view, low delta", camera_unit=1, look_at_unit=1, ), ] routed, res = _route(nodes, _facts(A=[1], B=[1])) assert res["render_action_routing_ok"] is True assert res["reuse_count"] == 1 assert routed[0]["render_action"] == "render_new_plate" assert routed[0]["reuse_target_bg_id"] == "" assert routed[1]["render_action"] == "reuse_existing_plate" assert routed[1]["reuse_target_bg_id"] == "A" def test_router_downgrade_when_camera_matches_but_no_look_or_overlap(): """Case 2: candidate true + valid target + same camera_unit but different look_at_unit AND no marker overlap → render_new_plate + non-fail diagnostic (LLM/code reconcile, not malformed).""" nodes = [ _rnode("A", 0, "fp_seeded_anchor", anchor=True, camera_unit=1), _rnode( "B", 1, "reference_derived", candidate=True, target="A", ld_rationale="maybe", camera_unit=1, look_at_unit=2, ), ] routed, res = _route(nodes, _facts(A=[1], B=[2])) assert res["render_action_routing_ok"] is True # non-fail assert res["reuse_count"] == 0 assert routed[1]["render_action"] == "render_new_plate" assert routed[1]["reuse_target_bg_id"] == "" assert any("corrobor" in d.lower() for d in res["diagnostics"]) def test_router_no_code_only_reuse_when_candidate_false(): """Case 3: candidate false but full deterministic overlap present → still render_new_plate. Code alone may never trigger reuse.""" nodes = [ _rnode("A", 0, "fp_seeded_anchor", anchor=True, camera_unit=1), _rnode( "B", 1, "reference_derived", candidate=False, target="", camera_unit=1, look_at_unit=1, ), ] routed, res = _route(nodes, _facts(A=[1], B=[1])) assert res["render_action_routing_ok"] is True assert res["reuse_count"] == 0 assert routed[1]["render_action"] == "render_new_plate" def test_router_hard_fail_candidate_true_empty_target(): """Case 4: candidate true but empty target id → malformed, hard fail.""" nodes = [ _rnode("A", 0, "fp_seeded_anchor", anchor=True, camera_unit=1), _rnode( "B", 1, "reference_derived", candidate=True, target="", ld_rationale="x", camera_unit=1, look_at_unit=1, ), ] _routed, res = _route(nodes, _facts(A=[1], B=[1])) assert res["render_action_routing_ok"] is False def test_router_hard_fail_candidate_false_but_target_nonempty(): """v2 semantic coupling: candidate false MUST carry empty target/rationale. Non-empty → hard fail (router validator area).""" nodes = [ _rnode("A", 0, "fp_seeded_anchor", anchor=True, camera_unit=1), _rnode( "B", 1, "reference_derived", candidate=False, target="A", camera_unit=1, look_at_unit=1, ), ] _routed, res = _route(nodes, _facts(A=[1], B=[1])) assert res["render_action_routing_ok"] is False def test_router_downgrades_when_the_target_is_itself_a_reuse_node(): """Case 5 — reuse-of-reuse 는 **내림**이지 실패가 아니다 (2026-09-19). 앞서는 이 한 줄이 `routing_ok=False` 를 놓아 **fp 전체를 죽였다.** 실측(컨트리로드 2판): fp 44개 중 `fp_container_interior` 만 이것으로 failed 였고(ok 42), provider 재시도는 routing 이전 원시 출력만 보므로 재질문도 안 열렸다 — 사람이 손대기 전에는 영영 안 풀린다. reuse-of-reuse 는 malformed 가 아니라 **허용 안 되는 요청**이고, 보수적으로 읽는 길이 하나뿐이다: 그 노드는 제 참조를 들고 새 plate 를 그린다. 그래서 다른 두 내림(부적격 노드·same-space 미확증)과 같이 둔다. """ nodes = [ _rnode("A", 0, "fp_seeded_anchor", anchor=True, camera_unit=1), _rnode( "B", 1, "reference_derived", candidate=True, target="A", ld_rationale="ok", camera_unit=1, look_at_unit=1, ), _rnode( "C", 2, "reference_derived", candidate=True, target="B", ld_rationale="chain", camera_unit=1, look_at_unit=1, ), ] routed, res = _route(nodes, _facts(A=[1], B=[1], C=[1])) assert res["render_action_routing_ok"] is True, "fp 를 죽이면 안 된다" by_bg = {n["bg_id"]: n for n in routed} # A 는 anchor → 새 plate. B 는 A 를 제대로 재사용. C 만 내려간다. assert by_bg["A"]["render_action"] == "render_new_plate" assert by_bg["B"]["render_action"] == "reuse_existing_plate" assert by_bg["B"]["reuse_target_bg_id"] == "A" assert by_bg["C"]["render_action"] == "render_new_plate" assert by_bg["C"]["reuse_target_bg_id"] == "" # ★내린 것을 **임의로 A 재사용으로 바꾸지 않는다** (Codex 조건). assert by_bg["C"]["reuse_target_bg_id"] != "A" # ★참조는 그대로 남는다 — 새로 그리되 무엇을 보고 그릴지는 안 잃는다. assert by_bg["C"]["reference_decision"]["selected_refs"], by_bg["C"] # 새 plate 는 둘(A·C), reuse 는 하나(B). assert res["reuse_plate_count"] == 1 assert sum(1 for n in routed if n["render_action"] == "render_new_plate") == 2 # 무슨 일이 있었는지는 진단에 남는다 — 조용히 고치면 안 된다. assert any("reuse-of-reuse" in d and "downgraded" in d for d in res["diagnostics"]), res["diagnostics"] # reuse 그래프는 여전히 비순환이고 깊이 1 이다. for n in routed: tgt = n.get("reuse_target_bg_id") if tgt: assert by_bg[tgt]["render_action"] == "render_new_plate" def test_router_hard_fail_target_future_or_self(): """A target must be an EARLIER node. self / future target → hard fail.""" nodes = [ _rnode("A", 0, "fp_seeded_anchor", anchor=True, camera_unit=1), _rnode( "B", 1, "reference_derived", candidate=True, target="B", ld_rationale="self", camera_unit=1, look_at_unit=1, ), ] _routed, res = _route(nodes, _facts(A=[1], B=[1])) assert res["render_action_routing_ok"] is False def test_router_hard_fail_target_unknown_node(): nodes = [ _rnode("A", 0, "fp_seeded_anchor", anchor=True, camera_unit=1), _rnode( "B", 1, "reference_derived", candidate=True, target="ZZZ", ld_rationale="ghost", camera_unit=1, look_at_unit=1, ), ] _routed, res = _route(nodes, _facts(A=[1], B=[1])) assert res["render_action_routing_ok"] is False def test_router_hard_fail_eligible_mode_empty_selected_refs(): """Case 6: eligible mode (here reference_derived) with empty selected_refs is a text-only same-space follow-up → hard fail, even on the render_new_plate path.""" nodes = [ _rnode("A", 0, "fp_seeded_anchor", anchor=True, camera_unit=1), _rnode( "B", 1, "reference_derived", candidate=False, camera_unit=1, look_at_unit=1, selected_refs=[], ), ] _routed, res = _route(nodes, _facts(A=[1], B=[1])) assert res["render_action_routing_ok"] is False def test_router_downgrade_forbidden_mode_candidate_true(): """Case 7: a reuse-ineligible mode (related_style_new_space) that the LLM flagged as a candidate is downgraded to render_new_plate with a non-fail diagnostic, never reused.""" nodes = [ _rnode("A", 0, "fp_seeded_anchor", anchor=True, camera_unit=1), _rnode( "B", 1, "related_style_new_space", candidate=True, target="A", ld_rationale="over-signal", camera_unit=1, look_at_unit=1, ), ] routed, res = _route(nodes, _facts(A=[1], B=[1])) assert res["render_action_routing_ok"] is True # non-fail downgrade assert res["reuse_count"] == 0 assert routed[1]["render_action"] == "render_new_plate" assert any("ineligible" in d.lower() or "mode" in d.lower() for d in res["diagnostics"]) def test_router_anchor_always_render_new_plate(): """The dwelling identity anchor is never a reuse node.""" nodes = [ _rnode("A", 0, "fp_seeded_anchor", anchor=True, camera_unit=1), ] routed, res = _route(nodes, _facts(A=[1])) assert routed[0]["render_action"] == "render_new_plate" assert routed[0]["reuse_target_bg_id"] == "" def test_router_reuse_on_same_look_alone_without_marker_overlap(): """corroboration = same_camera AND (same_look OR target_overlap). Lock the same_look-alone branch: same camera_unit + same look_at_unit but DISJOINT target markers must still reuse.""" nodes = [ _rnode("A", 0, "fp_seeded_anchor", anchor=True, camera_unit=1, look_at_unit=5), _rnode( "B", 1, "reference_derived", candidate=True, target="A", ld_rationale="same view", camera_unit=1, look_at_unit=5, ), ] routed, res = _route(nodes, _facts(A=[1], B=[2])) # disjoint markers assert res["reuse_count"] == 1 assert routed[1]["render_action"] == "reuse_existing_plate" assert routed[1]["reuse_target_bg_id"] == "A" def test_router_camera_unit_is_hard_floor_even_with_look_and_overlap(): """camera_unit equality is the hard floor: a different camera_unit may never reuse, even when look_at_unit matches AND markers overlap → render_new_plate + non-fail diagnostic.""" nodes = [ _rnode("A", 0, "fp_seeded_anchor", anchor=True, camera_unit=1, look_at_unit=5), _rnode( "B", 1, "reference_derived", candidate=True, target="A", ld_rationale="looks similar", camera_unit=2, look_at_unit=5, # different camera_unit ), ] routed, res = _route(nodes, _facts(A=[7], B=[7])) # full marker overlap assert res["render_action_routing_ok"] is True # non-fail assert res["reuse_count"] == 0 assert routed[1]["render_action"] == "render_new_plate" assert any("corrobor" in d.lower() for d in res["diagnostics"]) def test_router_same_physical_space_view_reuse_path(): """same_physical_space_view is reuse-eligible; corroborated candidate reuses the earlier plate.""" nodes = [ _rnode("A", 0, "fp_seeded_anchor", anchor=True, camera_unit=3), _rnode( "B", 1, "same_physical_space_view", candidate=True, target="A", ld_rationale="same room, new angle", camera_unit=3, look_at_unit=9, # different look, but overlap ), ] routed, res = _route(nodes, _facts(A=[7], B=[7])) # marker overlap assert routed[1]["render_action"] == "reuse_existing_plate" assert routed[1]["reuse_target_bg_id"] == "A" assert res["reuse_count"] == 1 def test_build_plan_attaches_canonical_render_action_fields(): """Integration: canonical render_action / reuse_target_bg_id land on every graph node written downstream, and a clean graph stays ok.""" geom = _geometry_cp_data() geom["per_fp"]["fp_a"]["readback_status"] = "ok" geom["per_fp"]["fp_a"]["geometry"]["readback_status"] = "ok" bundle = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=geom, overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=_shot_staging_cp_data(), )["fp_a"] def provider(**_kw): return _valid_llm_output() plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=bundle, llm_provider=provider ) assert plan["shot_aware_bg_render_plan_status"] == "ok" for node in plan["graph"]["nodes"]: assert node["render_action"] in ( "render_new_plate", "reuse_existing_plate" ) assert "reuse_target_bg_id" in node assert plan["validators"]["render_action_routing_ok"] is True # ────────────── W21B-w4 3a/3b — partition + reference-DAG wiring ────────────── def _wiring_bundle(): geom = _geometry_cp_data() geom["per_fp"]["fp_a"]["readback_status"] = "ok" geom["per_fp"]["fp_a"]["geometry"]["readback_status"] = "ok" return assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=geom, overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=_shot_staging_cp_data(), )["fp_a"] def test_build_plan_stamps_partition_and_dag_fields_when_card_off(): """(A) decision — ⑥ mirror_plate_partition + ⑦ build_reference_dag run UNCONDITIONALLY (card subsystem OFF / card_content_index=None). Every node carries all PLATE_PARTITION_FIELDS; every fresh-plate node carries all REFERENCE_DAG_FIELDS. This is the core regression guard for SCHEMA 9 — the persisted node shape always includes the partition/DAG fields, so downstream never branches on field presence.""" bundle = _wiring_bundle() plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=bundle, llm_provider=lambda **_kw: _valid_llm_output(), card_content_index=None, ) assert plan["shot_aware_bg_render_plan_status"] == "ok" nodes = plan["graph"]["nodes"] assert nodes for node in nodes: for f in PLATE_PARTITION_FIELDS: assert f in node, f"missing partition field {f}" if node.get("needs_new_plate"): for f in REFERENCE_DAG_FIELDS: assert f in node, f"missing dag field {f}" # ⑥/⑦ summaries surface on validators; withhold is 0 with no cards. assert "plate_shareability_counts" in plan["validators"] assert plan["validators"]["dag_node_count"] >= 1 assert plan["validators"]["card_withhold_count"] == 0 def test_build_plan_card_off_router_is_byte_identical_to_baseline(): """no-op proof — with card_content_index=None the wired route_v2 produces the SAME canonical render_action / reuse counts as the baseline route_render_actions (brief §10 default byte-identical). Only the additive partition/DAG mirror fields differ.""" bundle = _wiring_bundle() nodes = _valid_llm_output()["graph"]["nodes"] base_nodes, base_routing = route_render_actions( nodes=[dict(n) for n in nodes], dossier=bundle["dossier"] ) base_actions = {n["bg_id"]: n["render_action"] for n in base_nodes} plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=bundle, llm_provider=lambda **_kw: _valid_llm_output(), card_content_index=None, ) wired_actions = { n["bg_id"]: n["render_action"] for n in plan["graph"]["nodes"] } assert wired_actions == base_actions assert ( plan["validators"]["reuse_plate_count"] == base_routing["reuse_plate_count"] ) def test_build_plan_threads_card_content_index_into_router(monkeypatch): """wiring proof — build_render_plan_for_fp passes the envelope-bearing card_content_index AND per_bg_readiness through to route_render_actions_v2 (so a horizontal-band withhold can fire in production). The withhold mechanics themselves are covered by test_..._route_v2.""" bundle = _wiring_bundle() seen = {} real_v2 = _sabrp.route_render_actions_v2 def _spy(**kwargs): seen.update(kwargs) return real_v2(**kwargs) monkeypatch.setattr(_sabrp, "route_render_actions_v2", _spy) sentinel_index = {"L01B01": {"S1_Shot1": {"card_state": "blocked"}}} build_render_plan_for_fp( fp_id="fp_a", planner_input=bundle, llm_provider=lambda **_kw: _valid_llm_output(), card_content_index=sentinel_index, ) assert seen.get("card_content_index") is sentinel_index assert "L01B01" in (seen.get("per_bg_readiness") or {}) # ───────── camera candidate deterministic snap (final repair) ───────── # # W21B: the LLM emits camera_cell / look_at_cell as a verbatim copy of a # geometry candidate, but occasionally hallucinates an out-of-set cell # (e.g. a structural-unit cell that is deliberately NOT a look target). # After the W20F9 semantic re-prompt, a deterministic snap to the nearest # valid candidate is the final fallback so a constrained-choice field can # never hard-fail the fp. The snap is PURE; the provider gates it behind # _is_camera_only_failure so non-camera defects are never masked. def _snap_geometry() -> dict: """Inner geometry dict (validator shape) for snap tests.""" return { "camera_cell_candidates_per_unit": {"1": [[0, 0]], "2": [[1, 0]]}, "look_at_cell_candidates_per_unit": { "1": [[2, 0], [3, 0], [4, 0]], "2": [[2, 0], [3, 0], [4, 0]], }, } def _snap_node(bg_id, camera_unit, camera_cell, look_at_cell) -> dict: return { "bg_id": bg_id, "camera_decision": { "camera_unit": camera_unit, "camera_cell": list(camera_cell), "look_at_unit": camera_unit, "look_at_cell": list(look_at_cell), "lens_enum": "normal", "fov_deg": 50, }, } def test_snap_repairs_out_of_candidate_look_at_cell_to_nearest(): # look_at [5,0] invalid for unit 1; nearest in [[2,0],[3,0],[4,0]] is # [4,0] (Manhattan distance 1). nodes = [_snap_node("L01B01", 1, [0, 0], [5, 0])] repaired, repairs = _sabrp.snap_camera_cells_to_candidates( nodes=nodes, geometry=_snap_geometry() ) assert repaired[0]["camera_decision"]["look_at_cell"] == [4, 0] assert len(repairs) == 1 r = repairs[0] assert r["bg_id"] == "L01B01" assert r["field"] == "look_at_cell" assert r["unit"] == 1 assert r["from"] == [5, 0] assert r["to"] == [4, 0] assert r["distance"] == 1 # pure: input not mutated assert nodes[0]["camera_decision"]["look_at_cell"] == [5, 0] def test_snap_tie_break_prefers_lower_row_then_col(): # [3,0] is equidistant (Manhattan 1) to [2,0] and [4,0]; tie-break # (distance, row, col) picks the lower-row [2,0]. geom = { "camera_cell_candidates_per_unit": {"1": [[0, 0]]}, "look_at_cell_candidates_per_unit": {"1": [[2, 0], [4, 0]]}, } nodes = [_snap_node("L01B01", 1, [0, 0], [3, 0])] repaired, repairs = _sabrp.snap_camera_cells_to_candidates( nodes=nodes, geometry=geom ) assert repaired[0]["camera_decision"]["look_at_cell"] == [2, 0] assert repairs[0]["to"] == [2, 0] assert repairs[0]["distance"] == 1 def test_snap_camera_cell_repaired_independently_of_look_at(): # Only camera_cell is invalid (unit 2 expects [1,0]); look_at is valid. nodes = [_snap_node("L01B02", 2, [9, 9], [3, 0])] repaired, repairs = _sabrp.snap_camera_cells_to_candidates( nodes=nodes, geometry=_snap_geometry() ) assert repaired[0]["camera_decision"]["camera_cell"] == [1, 0] assert repaired[0]["camera_decision"]["look_at_cell"] == [3, 0] # untouched assert len(repairs) == 1 assert repairs[0]["field"] == "camera_cell" assert repairs[0]["unit"] == 2 def test_snap_no_op_when_cell_already_valid(): nodes = [_snap_node("L01B01", 1, [0, 0], [3, 0])] # both already valid repaired, repairs = _sabrp.snap_camera_cells_to_candidates( nodes=nodes, geometry=_snap_geometry() ) assert repairs == [] assert repaired[0]["camera_decision"]["look_at_cell"] == [3, 0] def test_snap_skips_malformed_cell_fail_fast(): # Non-int cell must NOT be snapped — left for the validator to fail-fast. nodes = [_snap_node("L01B01", 1, [0, 0], [0, "x"])] repaired, repairs = _sabrp.snap_camera_cells_to_candidates( nodes=nodes, geometry=_snap_geometry() ) assert repairs == [] assert repaired[0]["camera_decision"]["look_at_cell"] == [0, "x"] def test_snap_skips_when_candidate_set_empty(): geom = { "camera_cell_candidates_per_unit": {"1": [[0, 0]]}, "look_at_cell_candidates_per_unit": {"1": []}, # empty } nodes = [_snap_node("L01B01", 1, [0, 0], [5, 0])] repaired, repairs = _sabrp.snap_camera_cells_to_candidates( nodes=nodes, geometry=geom ) assert repairs == [] assert repaired[0]["camera_decision"]["look_at_cell"] == [5, 0] def test_snap_skips_invalid_camera_unit(): nodes = [_snap_node("L01B01", "not-an-int", [9, 9], [9, 9])] repaired, repairs = _sabrp.snap_camera_cells_to_candidates( nodes=nodes, geometry=_snap_geometry() ) assert repairs == [] assert repaired[0]["camera_decision"]["look_at_cell"] == [9, 9] def test_snap_ignores_malformed_candidate_entries(): # A malformed candidate entry must be skipped (not crash); the snap # still finds the nearest well-formed candidate. geom = { "camera_cell_candidates_per_unit": {"1": [[0, 0]]}, "look_at_cell_candidates_per_unit": {"1": [[2, 0], [9, "x"], [3, 0]]}, } nodes = [_snap_node("L01B01", 1, [0, 0], [5, 0])] repaired, repairs = _sabrp.snap_camera_cells_to_candidates( nodes=nodes, geometry=geom ) # [3,0] is the nearest WELL-FORMED candidate to [5,0] (Manhattan 2). assert repaired[0]["camera_decision"]["look_at_cell"] == [3, 0] assert repairs[0]["to"] == [3, 0] def test_build_plan_surfaces_camera_snap_repairs_in_diagnostics(): """A provider that applied a camera snap surfaces the repairs as a per-fp diagnostic (no silent repair) and bumps the llm call count.""" bundle = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=_geometry_cp_data(), overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=_shot_staging_cp_data(), )["fp_a"] bundle["readback_status"] = "ok" bundle["geometry"]["readback_status"] = "ok" def _provider_with_snap(**_kwargs): out = _valid_llm_output() out["_w20f9_retry_metadata"] = { "camera_validator_retry_attempted": 1, "camera_validator_retry_succeeded": 0, "camera_candidate_snap_repairs": [ { "bg_id": "L01B02", "field": "look_at_cell", "unit": 2, "from": [9, 9], "to": [3, 0], "distance": 6, } ], } return out plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=bundle, llm_provider=_provider_with_snap, ) surfaced = [ d["w20f9_camera_validator_retry"] for d in plan["diagnostics"] if isinstance(d, dict) and "w20f9_camera_validator_retry" in d ] assert len(surfaced) == 1 repairs = surfaced[0]["camera_candidate_snap_repairs"] assert repairs[0]["field"] == "look_at_cell" assert repairs[0]["to"] == [3, 0] assert plan["real_api_call_counts"]["llm"] == 2 def test_build_plan_surfaces_w20f10_graph_anchor_retry_and_llm_count(): """W20F10 sentinel = per-fp diagnostics 로 surface + llm 콜 2 집계 (카메라 메타 키 재사용 금지 — 별도 키).""" bundle = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=_geometry_cp_data(), overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=_shot_staging_cp_data(), )["fp_a"] bundle["readback_status"] = "ok" bundle["geometry"]["readback_status"] = "ok" def _provider_with_ga_retry(**_kwargs): out = _valid_llm_output() out["_graph_anchor_retry_metadata"] = { "graph_anchor_validator_retry_attempted": 1, "graph_anchor_validator_retry_succeeded": 1, "first_pass_diagnostics": ["graph.nodes must be a non-empty list"], } return out plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=bundle, llm_provider=_provider_with_ga_retry, ) surfaced = [ d["w20f10_graph_anchor_validator_retry"] for d in plan["diagnostics"] if isinstance(d, dict) and "w20f10_graph_anchor_validator_retry" in d ] assert len(surfaced) == 1 assert surfaced[0]["graph_anchor_validator_retry_succeeded"] == 1 assert plan["real_api_call_counts"]["llm"] == 2 # sentinel 은 영속 산출에서 strip assert "_graph_anchor_retry_metadata" not in plan def test_build_plan_failure_preserves_provider_completion_count(): """W20F10 실패(2콜) 예외의 completion_call_count 가 실패 감사의 실비용(llm=2)으로 보존 (Codex 조건 5).""" bundle = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=_geometry_cp_data(), overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=_shot_staging_cp_data(), )["fp_a"] bundle["readback_status"] = "ok" bundle["geometry"]["readback_status"] = "ok" def _provider_two_call_failure(**_kwargs): err = RuntimeError("W20F10 retry still failed (SAMPLE)") err.completion_call_count = 2 raise err plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=bundle, llm_provider=_provider_two_call_failure, ) assert plan["shot_aware_bg_render_plan_status"] == "failed" assert plan["real_api_call_counts"]["llm"] == 2 # ─────────── W20F11 — camera/look_at viability preflight (2026-07-23) ─────────── def _viability_bundle(camera=None, look_at=None): bundle = assemble_planner_inputs( dossier_cp_data=_dossier_cp_data(), geometry_cp_data=_geometry_cp_data(), overlay_cp_data=_overlay_cp_data(), master_plan_cp_data=_master_plan_cp_data(), shot_staging_cp_data=_shot_staging_cp_data(), )["fp_a"] bundle["readback_status"] = "ok" bundle["geometry"]["readback_status"] = "ok" if camera is not None: bundle["geometry"]["camera_cell_candidates_per_unit"] = camera if look_at is not None: bundle["geometry"]["look_at_cell_candidates_per_unit"] = look_at return bundle def _sentinel_provider(): from unittest.mock import MagicMock return MagicMock(side_effect=AssertionError( "provider must NOT be called for a non-viable fp")) def _assert_camera_na(plan): assert plan["shot_aware_bg_render_plan_status"] == "not_applicable" assert plan["not_applicable_reason_code"] == ( "no_viable_camera_look_at_pair") assert plan["real_api_call_counts"]["llm"] == 0 v = plan["camera_lookat_viability"] assert "camera_candidate_counts" in v assert "look_at_candidate_counts" in v def test_viability_all_look_at_empty_not_applicable_zero_calls(): """슬라이스 E fp_open_sea 실측 재현: look_at 전 유닛 빈 배열.""" provider = _sentinel_provider() plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=_viability_bundle( look_at={"1": [], "2": []}), llm_provider=provider, ) provider.assert_not_called() _assert_camera_na(plan) assert plan["camera_lookat_viability"][ "look_at_candidate_counts"] == {"1": 0, "2": 0} def test_viability_all_camera_empty_not_applicable_zero_calls(): provider = _sentinel_provider() plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=_viability_bundle(camera={"1": [], "2": []}), llm_provider=provider, ) provider.assert_not_called() _assert_camera_na(plan) def test_viability_disjoint_units_not_applicable(): """전역 합계는 nonzero 여도 같은 unit pair 가 없으면 구조적 불가.""" provider = _sentinel_provider() plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=_viability_bundle( camera={"1": [[0, 0]], "2": []}, look_at={"1": [], "2": [[2, 0]]}, ), llm_provider=provider, ) provider.assert_not_called() _assert_camera_na(plan) def test_viability_one_viable_unit_keeps_llm_path_unchanged(): """viable unit 존재=기존 provider 경로 그대로 (산출 무변).""" calls = [] def _provider(**kwargs): calls.append(kwargs["fp_id"]) return _valid_llm_output() plan = build_render_plan_for_fp( fp_id="fp_a", planner_input=_viability_bundle(), # fixture 후보 그대로=viable llm_provider=_provider, ) assert calls == ["fp_a"] assert plan["shot_aware_bg_render_plan_status"] == "ok" assert "not_applicable_reason_code" not in plan # 경계: 단 하나의 same-unit pair 만 있어도 viable (순수 함수 SOT) from app.modules.pipeline.shot_aware_bg_render_plan import ( assess_camera_lookat_viability, ) v = assess_camera_lookat_viability({ "camera_cell_candidates_per_unit": {"1": [[0, 0]], "2": []}, "look_at_cell_candidates_per_unit": {"1": [[2, 0]], "2": []}, }) assert v["viable"] is True and v["viable_units"] == ["1"] def test_viability_malformed_geometry_fail_closed_not_na(): """malformed=상류 계약 위반 — not_applicable 세탁 금지, raise.""" import pytest as _pytest for camera, look_at in ( ("not-a-dict", None), (None, "not-a-dict"), ({"1": "not-a-list"}, None), ({"1": [[0, 0, 0]]}, None), ({"1": [[0, "x"]]}, None), ({"1": [[True, 0]]}, None), # NARROW-1: validator str(camera_unit) 조회로 도달 불가한 key / # 응답 schema minimum=0 이라 선택 불가한 음수 cell ({1: [[0, 0]]}, None), # int key ({"01": [[0, 0]]}, None), # 비정규 key ({"--1": [[0, 0]]}, None), # raw ValueError 누출 경계 (total 가드) ({"1": [[-1, 0]]}, None), # 음수 cell ): provider = _sentinel_provider() with _pytest.raises(ShotAwareBgRenderPlanError): build_render_plan_for_fp( fp_id="fp_a", planner_input=_viability_bundle( camera=camera, look_at=look_at), llm_provider=provider, ) provider.assert_not_called() def test_viability_geometry_root_not_dict_fail_closed(): """root 자체가 dict 가 아니면 fail-closed (계약 완결).""" import pytest as _pytest from app.modules.pipeline.shot_aware_bg_render_plan import ( assess_camera_lookat_viability, ) with _pytest.raises(ShotAwareBgRenderPlanError): assess_camera_lookat_viability(None) with _pytest.raises(ShotAwareBgRenderPlanError): assess_camera_lookat_viability("not-a-dict")