"""W20B: shot-aware background render plan (LLM planner contract). Per-fp dwelling-scoped reference graph DAG + per-bg camera + reference decisions, all LLM-emitted, code-validated. **No image / VLM / DB / ImageAsset writes here.** The module exposes pure helpers; LLM calls are dependency-injected via an ``llm_provider`` callable so default production paths run with zero external API calls. Inputs (consumed from upstream checkpoints, all exact-ID joined): - base_location_dossier (W20A) — dwelling identity + per-bg facts. - floor_plan_geometry_readback (W20A2) — readback status + cell candidates. - floor_plan_overlay_payload (W19B-1) — already folded into the dossier; kept available for sanity. - background_master_plan — bg DAG + applies_to_shots. - shot_staging — committed shot staging payload. Output (one entry per fp_id with ≥1 renderable bg): - dwelling-scoped graph DAG (ordered nodes). - per-bg ``reference_decision`` (LLM mode + selected_refs + physical_space_id_per_ref + ...). - per-bg ``camera_decision`` (LLM cell + lens picks, validated against the W20A2 geometry candidate sets). - validators block enumerating which contracts passed/failed + structured blockers. - real_api_call_counts (image / llm / vlm) — always 0 unless an llm_provider is wired AND invoked. W20 boundaries enforced in this module: - exact-ID only — every join uses exact integer / string equality. - no semantic label / description parsing. - no deterministic top-K reference selection (LLM-owned). - no same-space judgement by code (LLM emits ``physical_space_id_per_ref``; code only checks set cardinality). - max_refs_per_bg = 2. - exactly one ``is_dwelling_identity_anchor = True`` per fp_id. - synthetic readback never promotes to ``production_ready = True``. - shot readiness is a structural gate; **no readiness LLM classifier** — if the staging cp does not carry the required fields, fail-closed with a precise blocker. """ from __future__ import annotations from typing import Any, Callable, Dict, FrozenSet, List, Optional, Tuple # W21B-w3 Commit 1 (2026-05-29): legacy ``style_reference_new_space`` # 제거 + ``same_physical_space_view`` / ``related_style_new_space`` 추가. # v2 prompt pack 이 emit 하는 mode set 과 정렬. 두 신규 mode 는 # ``reference_derived`` 와 동일하게 단일-ref (≤1) 로 취급되며, mode-keyed # 검증 로직은 ``TWO_REFS_MODE`` / anchor 두 곳뿐이라 그대로 호환된다. # canonical render_action / reuse 정규화는 Commit 2a router 영역. ALLOWED_REF_MODES: FrozenSet[str] = frozenset( { "fp_seeded_anchor", "reference_derived", "same_physical_space_view", "related_style_new_space", "two_refs_distinct_spaces", } ) TWO_REFS_MODE: str = "two_refs_distinct_spaces" MAX_REFS_PER_BG: int = 2 TWO_REFS_DEDUP_TOKEN: str = "distinct_visible_spaces" # W21B-w3 Commit 2a (2026-05-29): canonical render_action enum the # deterministic router stamps onto every plan node AFTER LLM-output # validation. The LLM never emits these — they are derived from the # low-delta candidate signal AND deterministic same-space corroboration. RENDER_ACTION_NEW: str = "render_new_plate" RENDER_ACTION_REUSE: str = "reuse_existing_plate" ALLOWED_RENDER_ACTIONS: FrozenSet[str] = frozenset( {RENDER_ACTION_NEW, RENDER_ACTION_REUSE} ) # Only these modes may be routed to reuse. ``fp_seeded_anchor`` (the # identity anchor), ``related_style_new_space`` (deliberately a *new* # room), and ``two_refs_distinct_spaces`` (two distinct spaces) are # never reuse candidates; an LLM candidate flag on them is downgraded. REUSE_ELIGIBLE_MODES: FrozenSet[str] = frozenset( {"same_physical_space_view", "reference_derived"} ) # Allowed lens enums must match the W20A2 geometry candidate set # verbatim. Re-declared here to avoid creating a reverse dependency. ALLOWED_LENS_ENUMS: FrozenSet[str] = frozenset({"wide", "normal", "telephoto"}) # W20D renderer-facing guidance fields (every field is a required # non-empty string, exact shape only — no semantic parsing). RENDER_GUIDANCE_FIELDS: Tuple[str, ...] = ( "visible_space_directive", "camera_framing_directive", "subject_position_directive", "state_cue_directive", "negative_continuity_directive", ) # W20A2 default lens→fov mapping fallback. Used only when the geometry # checkpoint did NOT carry view_cone_records (e.g. zero / one unit # observed); in normal flow the geometry's own records are SOT. _FALLBACK_LENS_FOV_MAP: Dict[str, int] = { "wide": 90, "normal": 50, "telephoto": 25, } class ShotAwareBgRenderPlanError(Exception): """Fail-closed signal for the planner builder.""" # ───────────────────────────────────────────────────────────────────── # Input assembly # ───────────────────────────────────────────────────────────────────── def _index_dossiers(dossier_cp_data: Dict[str, Any]) -> Dict[str, Dict[str, Any]]: return dict((dossier_cp_data or {}).get("dossiers") or {}) def _index_geometry( geometry_cp_data: Dict[str, Any], ) -> Dict[str, Dict[str, Any]]: out: Dict[str, Dict[str, Any]] = {} per_fp = (geometry_cp_data or {}).get("per_fp") or {} for fp_id, entry in per_fp.items(): if isinstance(entry, dict) and "geometry" in entry: out[fp_id] = entry return out def _index_shot_staging( shot_staging_cp_data: Dict[str, Any], *, diagnostics: Optional[Dict[str, Any]] = None, ) -> Dict[str, Dict[str, Any]]: """Build ``shot_id → staging entry`` exact-ID map. W20F1 — shot_staging cp 가 ``shot_id`` / ``id`` 를 carry 하지 않고 ``scene_index`` + ``shot_index`` 정수만 가지는 합법 입력 경로를 지원한다. fallback 합성 ID 형식 = ``f"S{scene_index}_Shot{shot_index}"`` (shot_extract / scene_detail violation message 의 명명과 동일, Codex 2026-05-28 confirmed). entry 가 명시 ``shot_id`` / ``id`` 를 들고 있으면 그 값을 그대로 우선 사용. bool 은 int subclass 라 ``isinstance(x, bool)`` 별도 reject. 결손/타입불량 entry 는 silent drop 금지 — ``diagnostics`` dict 가 주어졌으면 카운터에 누적, 없으면 logger.warning 으로 surface 한다. """ out: Dict[str, Dict[str, Any]] = {} counter_explicit_id = 0 counter_synth_id = 0 counter_dropped_no_id_no_index: List[str] = [] counter_dropped_bad_index_type: List[str] = [] counter_collision: List[str] = [] for raw_idx, entry in enumerate((shot_staging_cp_data or {}).get("shots") or []): if not isinstance(entry, dict): counter_dropped_no_id_no_index.append(f"") continue explicit_sid = entry.get("shot_id") or entry.get("id") if isinstance(explicit_sid, str) and explicit_sid: sid = explicit_sid counter_explicit_id += 1 else: scene_index = entry.get("scene_index") shot_index = entry.get("shot_index") if ( isinstance(scene_index, int) and not isinstance(scene_index, bool) and isinstance(shot_index, int) and not isinstance(shot_index, bool) ): sid = f"S{scene_index}_Shot{shot_index}" counter_synth_id += 1 else: counter_dropped_bad_index_type.append( f"raw_idx={raw_idx} scene_index={scene_index!r} " f"shot_index={shot_index!r}" ) continue if sid in out: counter_collision.append(sid) continue out[sid] = entry if diagnostics is not None: diagnostics["shot_staging_index_explicit_id"] = counter_explicit_id diagnostics["shot_staging_index_synthesized_id"] = counter_synth_id diagnostics["shot_staging_index_dropped_no_id"] = counter_dropped_no_id_no_index diagnostics["shot_staging_index_dropped_bad_index_type"] = counter_dropped_bad_index_type diagnostics["shot_staging_index_collisions"] = counter_collision else: if counter_dropped_no_id_no_index or counter_dropped_bad_index_type or counter_collision: import logging as _logging _logging.getLogger(__name__).warning( "_index_shot_staging: explicit=%d synth=%d dropped_no_id=%r " "dropped_bad_index=%r collisions=%r", counter_explicit_id, counter_synth_id, counter_dropped_no_id_no_index, counter_dropped_bad_index_type, counter_collision, ) return out def _consuming_shot_readiness( *, applies_to_shots: List[str], staging_index: Dict[str, Dict[str, Any]] ) -> Tuple[bool, List[str], List[Dict[str, Any]], List[str]]: """Structural readiness gate — code only verifies presence of the bg's consuming shot_ids in the staging cp. NO LLM classifier. W20E6-B contract: ``shot_staging`` is the staged/selected subset of the project's shots, not the full ``applies_to_shots`` universe. A BG with ANY staged shot is renderable; missing-from-staging shots are surfaced as ``omitted_unstaged_shots`` for diagnostic context, **not** as structural blockers. Only the empty / fully-unstaged cases carry structural blockers (the BG cannot be planned without at least one staged consuming shot). Returns ``(ok, blockers, surfaced_shot_entries, omitted_unstaged_shots)``. """ blockers: List[str] = [] surfaced: List[Dict[str, Any]] = [] omitted: List[str] = [] if not applies_to_shots: blockers.append("bg has empty applies_to_shots") return False, blockers, surfaced, omitted for sid in applies_to_shots: entry = staging_index.get(sid) if entry is None: omitted.append(sid) continue # Minimal structural check: shot_staging must carry at least the # shot's identifier. Deeper field checks belong to the LLM # prompt's contract, not to a code-side classifier. surfaced.append( { "shot_id": sid, # Pass-through verbatim — runtime data, not contract. "staging": entry, } ) if not surfaced: # No staged shot for this BG -> not renderable. Surface as a # structural blocker so the FP-level renderable_bg_ids set can # exclude it cleanly, while keeping the omitted list visible # to the LLM prompt and downstream audit. blockers.append( f"bg has no staged consuming shots " f"(omitted_unstaged_shots={omitted!r})" ) return False, blockers, surfaced, omitted return True, blockers, surfaced, omitted def assemble_planner_inputs( *, dossier_cp_data: Dict[str, Any], geometry_cp_data: Dict[str, Any], overlay_cp_data: Dict[str, Any], master_plan_cp_data: Dict[str, Any], shot_staging_cp_data: Dict[str, Any], shot_staging_diagnostics: Optional[Dict[str, Any]] = None, ) -> Dict[str, Dict[str, Any]]: """Build per-fp planner input bundles. Each fp_id present in the dossier checkpoint and the geometry checkpoint produces one bundle: { "fp_id": str, "dossier": , "geometry": , "readback_status": str, "shot_readiness": { "ok": bool, "blockers": [str], "per_bg": { bg_id: { "applies_to_shots": [str], "surfaced_shots": [{"shot_id": str, "staging": {...}}], "ok": bool, "blockers": [str], } } }, "candidate_catalog": [], # W20B: empty seed catalog; future wave may grow. "raw_overlay": , } fp_ids missing from either dossier or geometry are not surfaced (the planner step turns that into a fp-level skip + diagnostic). """ dossiers = _index_dossiers(dossier_cp_data) geometries = _index_geometry(geometry_cp_data) staging_index = _index_shot_staging( shot_staging_cp_data, diagnostics=shot_staging_diagnostics, ) out: Dict[str, Dict[str, Any]] = {} for fp_id, dossier in dossiers.items(): if fp_id not in geometries: continue geom_entry = geometries[fp_id] geometry = geom_entry.get("geometry") or {} readback_status = ( geom_entry.get("readback_status") or geom_entry.get("readback", {}).get("status", "unknown") ) per_bg_facts = ( dossier.get("per_bg_render_facts_by_bg_id") or {} ) per_bg_readiness: Dict[str, Dict[str, Any]] = {} fp_blockers: List[str] = [] any_ok = False renderable_bg_ids_list: List[str] = [] for bg_id, facts in per_bg_facts.items(): applies = list(facts.get("applies_to_shots") or []) ok, bg_blockers, surfaced, omitted = _consuming_shot_readiness( applies_to_shots=applies, staging_index=staging_index ) per_bg_readiness[bg_id] = { "applies_to_shots": applies, "surfaced_shots": surfaced, # W20E6-B: missing-from-staging shots are diagnostic # context for the LLM. They are NOT blockers. "omitted_unstaged_shots": omitted, "ok": ok, "blockers": bg_blockers, } if ok: any_ok = True renderable_bg_ids_list.append(bg_id) else: # Surface BG-level blockers under the FP banner so the # consuming reporter can audit per-BG drop reasons. fp_blockers.extend( f"bg_id={bg_id!r}: {b}" for b in bg_blockers ) overlay_subset = { bg_id: payload for bg_id, payload in ( (overlay_cp_data or {}).get("overlays") or {} ).items() if isinstance(payload, dict) and payload.get("fp_id") == fp_id } out[fp_id] = { "fp_id": fp_id, "dossier": dossier, "geometry": geometry, "readback_status": readback_status, "shot_readiness": { "ok": any_ok, "blockers": fp_blockers, "per_bg": per_bg_readiness, # W20E6-B: the staged-shot-only renderable set drives # graph completeness downstream (LLM only emits nodes # for these bg_ids). "renderable_bg_ids": sorted(renderable_bg_ids_list), }, "candidate_catalog": [], "raw_overlay": overlay_subset, } return out # ───────────────────────────────────────────────────────────────────── # Validators # ───────────────────────────────────────────────────────────────────── def _is_int_pair_list(value: Any) -> bool: """``[row, col]`` shape with two non-negative ints (no bools).""" if not isinstance(value, list) or len(value) != 2: return False for v in value: if isinstance(v, bool) or not isinstance(v, int): return False return True def _validate_required_shape( *, nodes: List[Dict[str, Any]] ) -> Tuple[bool, List[str]]: """Structural / type-only schema-lite check. The W20B prompt pack carries a JSON Schema, but the production validator must also fail-closed on missing / wrong-typed fields so a malformed LLM output cannot quietly default through the other validators. **No semantic parsing** — code checks only that the keys exist and the types match. Allowed types mirror the ``schema.json`` shape exactly. """ blockers: List[str] = [] if not isinstance(nodes, list) or not nodes: blockers.append("graph.nodes must be a non-empty list") return False, blockers for idx, node in enumerate(nodes): prefix = f"graph.nodes[{idx}]" if not isinstance(node, dict): blockers.append(f"{prefix} not a dict") continue # top-level fields bg_id = node.get("bg_id") if not isinstance(bg_id, str) or not bg_id: blockers.append(f"{prefix}.bg_id must be non-empty string") ni = node.get("node_index") if isinstance(ni, bool) or not isinstance(ni, int) or ni < 0: blockers.append(f"{prefix}.node_index must be non-negative int") mode = node.get("mode") if not isinstance(mode, str) or not mode: blockers.append(f"{prefix}.mode must be non-empty string") is_anchor = node.get("is_dwelling_identity_anchor") if not isinstance(is_anchor, bool): blockers.append( f"{prefix}.is_dwelling_identity_anchor must be bool" ) rationale = node.get("rationale") if not isinstance(rationale, str): blockers.append(f"{prefix}.rationale must be string") ref_dec = node.get("reference_decision") if not isinstance(ref_dec, dict): blockers.append(f"{prefix}.reference_decision must be dict") else: sel = ref_dec.get("selected_refs") if not isinstance(sel, list): blockers.append( f"{prefix}.reference_decision.selected_refs must be list" ) else: for s_idx, entry in enumerate(sel): s_prefix = f"{prefix}.reference_decision.selected_refs[{s_idx}]" if not isinstance(entry, dict): blockers.append(f"{s_prefix} not a dict") continue if not isinstance(entry.get("ref_bg_id"), str) or not entry.get("ref_bg_id"): blockers.append( f"{s_prefix}.ref_bg_id must be non-empty string" ) if not isinstance(entry.get("physical_space_id"), str) or not entry.get("physical_space_id"): blockers.append( f"{s_prefix}.physical_space_id must be non-empty string" ) rej = ref_dec.get("rejected_refs") if not isinstance(rej, list): blockers.append( f"{prefix}.reference_decision.rejected_refs must be list" ) for fname in ( "same_physical_space_dedup_decision", "why_single_ref_or_two_refs", # W21B-w3 Commit 1: same-space low-delta reuse candidate # 신호. empty sentinel 허용 (candidate=false 시 ""). semantic # coupling (candidate↔target/rationale) 은 Commit 2a router. "low_delta_reuse_target_bg_id", "low_delta_rationale", ): if not isinstance(ref_dec.get(fname), str): blockers.append( f"{prefix}.reference_decision.{fname} must be string" ) # W21B-w3 Commit 1: candidate flag bool required (type/shape only). if not isinstance( ref_dec.get("same_physical_space_low_delta_candidate"), bool ): blockers.append( f"{prefix}.reference_decision." f"same_physical_space_low_delta_candidate must be bool" ) psi = ref_dec.get("physical_space_id_per_ref") if not isinstance(psi, list): blockers.append( f"{prefix}.reference_decision.physical_space_id_per_ref " f"must be list" ) else: for p_idx, p in enumerate(psi): if not isinstance(p, str) or not p: blockers.append( f"{prefix}.reference_decision." f"physical_space_id_per_ref[{p_idx}] " f"must be non-empty string" ) cam = node.get("camera_decision") if not isinstance(cam, dict): blockers.append(f"{prefix}.camera_decision must be dict") else: for fname in ("camera_unit", "look_at_unit"): v = cam.get(fname) if isinstance(v, bool) or not isinstance(v, int): blockers.append( f"{prefix}.camera_decision.{fname} must be int" ) for fname in ("camera_cell", "look_at_cell"): if not _is_int_pair_list(cam.get(fname)): blockers.append( f"{prefix}.camera_decision.{fname} must be " f"[int, int]" ) lens = cam.get("lens_enum") if not isinstance(lens, str) or not lens: blockers.append( f"{prefix}.camera_decision.lens_enum must be string" ) fov = cam.get("fov_deg") if isinstance(fov, bool) or not isinstance(fov, int) or fov <= 0: blockers.append( f"{prefix}.camera_decision.fov_deg must be positive int" ) rg = node.get("render_guidance") if not isinstance(rg, dict): blockers.append(f"{prefix}.render_guidance must be dict") else: extra_keys = sorted(set(rg.keys()) - set(RENDER_GUIDANCE_FIELDS)) if extra_keys: blockers.append( f"{prefix}.render_guidance has unknown keys " f"{extra_keys}; allowed={list(RENDER_GUIDANCE_FIELDS)}" ) for fname in RENDER_GUIDANCE_FIELDS: value = rg.get(fname) if not isinstance(value, str) or not value.strip(): blockers.append( f"{prefix}.render_guidance.{fname} must be " f"non-empty string" ) return (not blockers), blockers def _validate_node_index_order( *, nodes: List[Dict[str, Any]] ) -> Tuple[bool, List[str]]: """``node_index`` must equal the entry's position in ``nodes``.""" blockers: List[str] = [] seen: set = set() for idx, node in enumerate(nodes): ni = node.get("node_index") if isinstance(ni, bool) or not isinstance(ni, int): blockers.append( f"node[{idx}].node_index missing or non-int — order check skipped" ) continue if ni != idx: blockers.append( f"node[{idx}].node_index={ni} does not equal its list " f"position {idx}" ) if ni in seen: blockers.append( f"node_index={ni} appears more than once in graph" ) seen.add(ni) return (not blockers), blockers def _validate_graph_completeness( *, nodes: List[Dict[str, Any]], renderable_bg_ids: FrozenSet[str] ) -> Tuple[bool, List[str]]: """Exact-string set equality between graph node bg_ids and the fp's **renderable** bg_ids (W20E6-B). Renderable = the BG has at least one staged consuming shot in ``shot_staging``. Non-renderable BGs (no staged shots) are deliberately excluded from the LLM-emitted graph; this validator enforces that contract -- any missing renderable bg_id, or any emitted bg_id outside the renderable set, is a hard fail. """ blockers: List[str] = [] seen: List[str] = [] for node in nodes: bg_id = node.get("bg_id") if isinstance(bg_id, str) and bg_id: seen.append(bg_id) seen_set = set(seen) missing = sorted(renderable_bg_ids - seen_set) extra = sorted(seen_set - renderable_bg_ids) if missing: blockers.append( f"graph completeness: missing renderable bg_ids {missing} " f"(dwelling-scoped graph must include every staged-shot " f"renderable BG of the fp)" ) if extra: blockers.append( f"graph completeness: extra bg_ids {extra} " f"(graph emitted bg_ids that are not in the staged-shot " f"renderable set)" ) return (not blockers), blockers def _validate_dag(*, nodes: List[Dict[str, Any]]) -> Tuple[bool, List[str]]: """No cycles, every parent appears earlier in the order.""" blockers: List[str] = [] bg_seen: Dict[str, int] = {} for idx, node in enumerate(nodes): bg_id = node.get("bg_id") if not isinstance(bg_id, str): blockers.append(f"node[{idx}].bg_id missing or non-string") continue if bg_id in bg_seen: blockers.append(f"bg_id={bg_id!r} appears twice in graph") continue bg_seen[bg_id] = idx ref_dec = node.get("reference_decision") or {} for ref in ref_dec.get("selected_refs") or []: ref_bg = ref.get("ref_bg_id") if not isinstance(ref_bg, str): blockers.append( f"bg_id={bg_id!r} ref entry missing ref_bg_id" ) continue if ref_bg == bg_id: blockers.append( f"bg_id={bg_id!r} references itself (self-loop)" ) continue parent_idx = bg_seen.get(ref_bg) if parent_idx is None: blockers.append( f"bg_id={bg_id!r} references {ref_bg!r} which has " f"not appeared earlier in the graph (DAG violation)" ) elif parent_idx >= idx: blockers.append( f"bg_id={bg_id!r} references {ref_bg!r} that is " f"not earlier in walk order (DAG violation)" ) return (not blockers), blockers def _validate_max_refs_per_bg( *, nodes: List[Dict[str, Any]] ) -> Tuple[bool, List[str]]: blockers: List[str] = [] for node in nodes: ref_dec = node.get("reference_decision") or {} selected = ref_dec.get("selected_refs") or [] if len(selected) > MAX_REFS_PER_BG: blockers.append( f"bg_id={node.get('bg_id')!r} selected_refs count " f"{len(selected)} > {MAX_REFS_PER_BG}" ) return (not blockers), blockers def _validate_two_refs_distinct_spaces( *, nodes: List[Dict[str, Any]] ) -> Tuple[bool, List[str]]: """Reference-decision consistency. Codex W20B narrow patch #3: - ``physical_space_id_per_ref`` length must equal ``selected_refs`` length for ALL counts (0 / 1 / 2). - ``selected_refs[i].physical_space_id`` must equal ``physical_space_id_per_ref[i]`` exact-string. - ``mode == 'two_refs_distinct_spaces'`` ⇔ ``len(selected_refs) == 2``. - When 2 refs: * the two ``ref_bg_id`` values are exact-string distinct; * the two ``physical_space_id_per_ref`` values are exact-string distinct; * ``same_physical_space_dedup_decision`` = ``distinct_visible_spaces``; * ``why_single_ref_or_two_refs`` is non-empty. Code never lexically inspects descriptions; only exact-string set / equality checks. """ blockers: List[str] = [] for node in nodes: bg_id = node.get("bg_id") mode = node.get("mode") ref_dec = node.get("reference_decision") or {} selected = ref_dec.get("selected_refs") psi = ref_dec.get("physical_space_id_per_ref") # Robust to lists/non-lists. Required-shape validator already # reports a hard error when they are not list; we still guard # below so this validator doesn't crash on bad input. sel_list = list(selected) if isinstance(selected, list) else [] psi_list = list(psi) if isinstance(psi, list) else [] # 1) length agreement (0/1/2). if isinstance(selected, list) and isinstance(psi, list): if len(psi_list) != len(sel_list): blockers.append( f"bg_id={bg_id!r} physical_space_id_per_ref length " f"({len(psi_list)}) != selected_refs length " f"({len(sel_list)})" ) # 2) per-index exact equality between selected_refs[i].physical_space_id # and physical_space_id_per_ref[i]. for i, (sel_entry, psi_id) in enumerate(zip(sel_list, psi_list)): if not isinstance(sel_entry, dict): continue sel_psi = sel_entry.get("physical_space_id") if isinstance(sel_psi, str) and isinstance(psi_id, str): if sel_psi != psi_id: blockers.append( f"bg_id={bg_id!r} selected_refs[{i}]." f"physical_space_id={sel_psi!r} does not equal " f"physical_space_id_per_ref[{i}]={psi_id!r}" ) # 3) mode ↔ selected_refs count contract. if mode == TWO_REFS_MODE: if len(sel_list) != 2: blockers.append( f"bg_id={bg_id!r} mode={TWO_REFS_MODE!r} requires " f"exactly 2 selected_refs (got {len(sel_list)})" ) elif len(sel_list) == 2 and isinstance(mode, str): blockers.append( f"bg_id={bg_id!r} has 2 selected_refs but mode={mode!r} " f"(must be {TWO_REFS_MODE!r})" ) # 4) Two-ref distinctness checks. if len(sel_list) == 2: ref_bg_ids = [ e.get("ref_bg_id") if isinstance(e, dict) else None for e in sel_list ] if all(isinstance(r, str) and r for r in ref_bg_ids): if ref_bg_ids[0] == ref_bg_ids[1]: blockers.append( f"bg_id={bg_id!r} two selected_refs share " f"ref_bg_id={ref_bg_ids[0]!r} (must be exact-string distinct)" ) if not ( len(psi_list) == 2 and all(isinstance(s, str) and s for s in psi_list) ): blockers.append( f"bg_id={bg_id!r} two-ref mode requires exactly 2 " f"non-empty physical_space_id_per_ref entries " f"(got {psi_list!r})" ) elif len(set(psi_list)) != 2: blockers.append( f"bg_id={bg_id!r} physical_space_id_per_ref entries " f"are not exact-string distinct (got {psi_list!r})" ) dedup = ref_dec.get("same_physical_space_dedup_decision") if dedup != TWO_REFS_DEDUP_TOKEN: blockers.append( f"bg_id={bg_id!r} two-ref mode requires " f"same_physical_space_dedup_decision=" f"{TWO_REFS_DEDUP_TOKEN!r} (got {dedup!r})" ) why = ref_dec.get("why_single_ref_or_two_refs") if not isinstance(why, str) or not why.strip(): blockers.append( f"bg_id={bg_id!r} two-ref mode requires non-empty " f"why_single_ref_or_two_refs" ) return (not blockers), blockers def _derive_lens_fov_map( *, geometry: Dict[str, Any] ) -> Tuple[Dict[str, int], List[str]]: """Build a ``lens_enum → fov_deg`` map from the geometry checkpoint. Source-of-truth is ``geometry.view_cone_records`` (W20A2 carries one record per ``(direction, lens)`` pair). When a lens appears in multiple records with **disagreeing** fov_deg, that is a hard fail-closed signal — geometry shape is broken upstream. When ``view_cone_records`` is missing or empty (e.g. fewer than two units observed), the fallback ``_FALLBACK_LENS_FOV_MAP`` is returned with a diagnostic so the consumer knows the map is a default, not an observation. """ blockers: List[str] = [] records = geometry.get("view_cone_records") or [] by_lens: Dict[str, int] = {} for r in records: if not isinstance(r, dict): continue lens = r.get("lens_enum") fov = r.get("fov_deg") if not isinstance(lens, str) or not isinstance(fov, int): continue prior = by_lens.get(lens) if prior is None: by_lens[lens] = fov elif prior != fov: blockers.append( f"geometry.view_cone_records carries conflicting " f"fov_deg for lens={lens!r}: {prior} vs {fov}" ) if not by_lens: # No records → use fallback. Mark via empty diagnostic; caller # will surface it only if a node references a lens not in # ALLOWED_LENS_ENUMS. return dict(_FALLBACK_LENS_FOV_MAP), blockers return by_lens, blockers def _validate_same_fp_only( *, fp_id: str, nodes: List[Dict[str, Any]], same_fp_bg_ids: FrozenSet[str], ) -> Tuple[bool, List[str]]: blockers: List[str] = [] for node in nodes: bg_id = node.get("bg_id") if isinstance(bg_id, str) and bg_id not in same_fp_bg_ids: blockers.append( f"bg_id={bg_id!r} in graph is not a same-fp bg of " f"fp_id={fp_id!r}" ) ref_dec = node.get("reference_decision") or {} for ref in ref_dec.get("selected_refs") or []: ref_bg = ref.get("ref_bg_id") if isinstance(ref_bg, str) and ref_bg not in same_fp_bg_ids: blockers.append( f"bg_id={bg_id!r} references ref_bg_id={ref_bg!r} " f"outside same-fp set" ) return (not blockers), blockers def _validate_anchor_exactly_one( *, nodes: List[Dict[str, Any]] ) -> Tuple[bool, List[str]]: blockers: List[str] = [] anchors = [n for n in nodes if n.get("is_dwelling_identity_anchor")] if len(anchors) != 1: blockers.append( f"graph requires exactly one anchor node " f"(found {len(anchors)})" ) else: first_node = nodes[0] if nodes else None if anchors[0] is not first_node: blockers.append( "anchor node must be the first node in graph walk order" ) anchor_mode = anchors[0].get("mode") if anchor_mode != "fp_seeded_anchor": blockers.append( f"anchor mode must be fp_seeded_anchor " f"(got {anchor_mode!r})" ) return (not blockers), blockers def _validate_anchor_in_clean_candidate_set( *, nodes: List[Dict[str, Any]], clean_anchor_candidate_bg_ids: Optional[FrozenSet[str]], ) -> Tuple[bool, List[str]]: """W20E7-C: anchor BG must be drawn from the dossier's clean anchor candidate surface (``anchor_selection_metadata.candidate_bg_ids`` — every BG in that list satisfies ``clean_background_expected=True``). Caller contract: - ``clean_anchor_candidate_bg_ids is None``: validator is a no-op (back-compat for unit-test call sites pre-dating the W20E7-C wiring). Production paths NEVER pass ``None`` — they always derive the set from the dossier. - ``clean_anchor_candidate_bg_ids == frozenset()``: empty surface is a hard fail (no clean anchor candidate exists for this fp; the LLM cannot legitimately pick an anchor). - non-empty set: the anchor's ``bg_id`` MUST be an exact-string member of the set. Cross-validator interaction: when ``_validate_anchor_exactly_one`` already fails (zero / multiple anchors, or anchor not at index 0), this validator no-ops so the operator sees the primary anchor shape error without a redundant diagnostic chain. """ if clean_anchor_candidate_bg_ids is None: return True, [] blockers: List[str] = [] anchors = [n for n in nodes if n.get("is_dwelling_identity_anchor")] if len(anchors) != 1: # Delegated to _validate_anchor_exactly_one — do not duplicate. return True, [] anchor_bg_id = anchors[0].get("bg_id") if not isinstance(anchor_bg_id, str) or not anchor_bg_id: # Required-shape validator already reports this. return True, [] if not clean_anchor_candidate_bg_ids: blockers.append( "clean anchor candidate set is empty -- dossier surfaced no " "clean_background_expected=True bg_ids for this fp; the LLM " "cannot pick an anchor without a clean candidate." ) return False, blockers if anchor_bg_id not in clean_anchor_candidate_bg_ids: blockers.append( f"anchor bg_id={anchor_bg_id!r} is not in the clean anchor " f"candidate set {sorted(clean_anchor_candidate_bg_ids)!r}; " f"the anchor MUST be drawn from the dossier's " f"anchor_selection_metadata.candidate_bg_ids subset " f"(clean_background_expected=True only)." ) return False, blockers return True, blockers def _validate_camera_in_candidates( *, nodes: List[Dict[str, Any]], geometry: Dict[str, Any], ) -> Tuple[bool, List[str]]: blockers: List[str] = [] cam_cands = geometry.get("camera_cell_candidates_per_unit") or {} look_cands = geometry.get("look_at_cell_candidates_per_unit") or {} lens_fov_map, lens_map_blockers = _derive_lens_fov_map(geometry=geometry) blockers.extend(lens_map_blockers) for node in nodes: cam = node.get("camera_decision") or {} bg_id = node.get("bg_id") cam_unit = cam.get("camera_unit") cam_cell = cam.get("camera_cell") look_cell = cam.get("look_at_cell") lens_enum = cam.get("lens_enum") fov_deg = cam.get("fov_deg") if isinstance(cam_unit, bool) or not isinstance(cam_unit, int): blockers.append( f"bg_id={bg_id!r} camera_decision.camera_unit not int" ) continue unit_key = str(cam_unit) if unit_key not in cam_cands: blockers.append( f"bg_id={bg_id!r} camera_unit={cam_unit} not in " f"camera_cell_candidates_per_unit" ) continue # cells are emitted as [row, col] lists. valid_cam_cells = [tuple(c) for c in cam_cands[unit_key] or []] if not ( isinstance(cam_cell, list) and len(cam_cell) == 2 and tuple(cam_cell) in valid_cam_cells ): blockers.append( f"bg_id={bg_id!r} camera_cell={cam_cell!r} not in " f"camera_cell_candidates_per_unit[{unit_key}]" ) valid_look_cells = [tuple(c) for c in look_cands.get(unit_key) or []] if not ( isinstance(look_cell, list) and len(look_cell) == 2 and tuple(look_cell) in valid_look_cells ): blockers.append( f"bg_id={bg_id!r} look_at_cell={look_cell!r} not in " f"look_at_cell_candidates_per_unit[{unit_key}]" ) if lens_enum not in ALLOWED_LENS_ENUMS: blockers.append( f"bg_id={bg_id!r} lens_enum={lens_enum!r} not in " f"{sorted(ALLOWED_LENS_ENUMS)}" ) else: expected_fov = lens_fov_map.get(lens_enum) if expected_fov is None: blockers.append( f"bg_id={bg_id!r} lens_enum={lens_enum!r} has no " f"fov mapping in geometry.view_cone_records or fallback" ) elif not ( isinstance(fov_deg, int) and not isinstance(fov_deg, bool) ): blockers.append( f"bg_id={bg_id!r} fov_deg={fov_deg!r} not int " f"(expected {expected_fov} for lens={lens_enum!r})" ) elif fov_deg != expected_fov: blockers.append( f"bg_id={bg_id!r} fov_deg={fov_deg} does not match " f"expected {expected_fov} for lens_enum={lens_enum!r}" ) return (not blockers), blockers def _is_int_cell(cell: Any) -> bool: """A grid cell is a ``[int, int]`` pair (bool excluded — bool is an int).""" return ( isinstance(cell, list) and len(cell) == 2 and all(isinstance(v, int) and not isinstance(v, bool) for v in cell) ) def _snap_cell_to_nearest( cell: Any, candidates: Any ) -> Optional[List[int]]: """Nearest valid candidate to ``cell`` (Manhattan distance, stable tie-break), or ``None`` when no snap should happen. Returns ``None`` (no repair — left for the validator to fail-fast) when: - ``cell`` is not a ``[int, int]`` pair (malformed), - the candidate set is empty, - ``cell`` is already a valid candidate (no-op). Tie-break is ``(distance, row, col)`` so the choice is deterministic. """ if not _is_int_cell(cell): return None valid = [ (c[0], c[1]) for c in (candidates or []) if isinstance(c, (list, tuple)) and len(c) == 2 and all(isinstance(v, int) and not isinstance(v, bool) for v in c) ] if not valid: return None cell_t = (cell[0], cell[1]) if cell_t in valid: return None best = min( valid, key=lambda cand: ( abs(cand[0] - cell_t[0]) + abs(cand[1] - cell_t[1]), cand[0], cand[1], ), ) return [best[0], best[1]] def snap_camera_cells_to_candidates( *, nodes: List[Dict[str, Any]], geometry: Dict[str, Any], ) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]: """Deterministically repair out-of-candidate ``camera_cell`` / ``look_at_cell`` to the nearest valid candidate. PURE: returns ``(repaired_nodes, repairs)`` and never mutates the input nodes. ``repairs`` carries one entry per snapped field:: {"bg_id", "field", "unit", "from", "to", "distance"} The function repairs ONLY the constrained-choice cell fields and ONLY when a snap is unambiguous (valid ``camera_unit``, non-empty candidate set, ``[int, int]`` cell). Malformed cells / invalid units / empty candidate sets are left untouched so the validator fail-fasts. It is the caller's responsibility to gate this behind a camera-only failure so non-camera defects are never masked. """ cam_cands = geometry.get("camera_cell_candidates_per_unit") or {} look_cands = geometry.get("look_at_cell_candidates_per_unit") or {} repaired: List[Dict[str, Any]] = [] repairs: List[Dict[str, Any]] = [] for node in nodes: new_node = dict(node) cam = dict(node.get("camera_decision") or {}) cam_unit = cam.get("camera_unit") if isinstance(cam_unit, int) and not isinstance(cam_unit, bool): unit_key = str(cam_unit) for field, cand_map in ( ("camera_cell", cam_cands), ("look_at_cell", look_cands), ): cell = cam.get(field) snapped = _snap_cell_to_nearest(cell, cand_map.get(unit_key)) if snapped is not None: repairs.append( { "bg_id": node.get("bg_id"), "field": field, "unit": cam_unit, "from": [cell[0], cell[1]], "to": snapped, "distance": ( abs(snapped[0] - cell[0]) + abs(snapped[1] - cell[1]) ), } ) cam[field] = snapped new_node["camera_decision"] = cam repaired.append(new_node) return repaired, repairs def _validate_rationale_nonempty( *, nodes: List[Dict[str, Any]] ) -> Tuple[bool, List[str]]: blockers: List[str] = [] for node in nodes: rationale = node.get("rationale") if not isinstance(rationale, str) or not rationale.strip(): blockers.append( f"bg_id={node.get('bg_id')!r} rationale missing or empty" ) mode = node.get("mode") if mode not in ALLOWED_REF_MODES: blockers.append( f"bg_id={node.get('bg_id')!r} mode={mode!r} not in " f"{sorted(ALLOWED_REF_MODES)}" ) return (not blockers), blockers def _validate_synthetic_readback_block( *, readback_status: str, ) -> Tuple[bool, List[str]]: """When readback is synthetic, the plan must NOT be marked production-clear. We surface that as a *constraint diagnostic* — the plan can still be authored, but a real smoke is blocked.""" if readback_status != "ok": return False, [ f"readback_status={readback_status!r} — plan cannot be " f"promoted to production_ready until a real VLM readback " f"(status='ok') is wired up" ] return True, [] def validate_llm_output( *, fp_id: str, nodes: List[Dict[str, Any]], same_fp_bg_ids: FrozenSet[str], geometry: Dict[str, Any], readback_status: str, renderable_bg_ids: Optional[FrozenSet[str]] = None, clean_anchor_candidate_bg_ids: Optional[FrozenSet[str]] = None, ) -> Dict[str, Any]: """Validate the LLM-emitted shot-aware bg render plan graph. W20E6-B: ``renderable_bg_ids`` is the staged-shot subset of ``same_fp_bg_ids`` and drives graph-completeness. When omitted (no caller supplied a subset), the broader ``same_fp_bg_ids`` is used as a backward-compatible default -- preserves call sites that have not yet been updated. W20E7-C: ``clean_anchor_candidate_bg_ids`` is the dossier's ``anchor_selection_metadata.candidate_bg_ids`` (the ``clean_background_expected=True`` subset). When provided, the anchor's bg_id MUST be a member; an empty set fails closed (no clean anchor available). When ``None``, the validator is a no-op (back-compat for unit tests pre-dating the wiring). Production paths always pass a frozenset, possibly empty. """ diagnostics: List[str] = [] shape_ok, b = _validate_required_shape(nodes=nodes) diagnostics.extend(b) completeness_target = ( renderable_bg_ids if renderable_bg_ids is not None else same_fp_bg_ids ) completeness_ok, b = _validate_graph_completeness( nodes=nodes, renderable_bg_ids=completeness_target ) diagnostics.extend(b) node_index_ok, b = _validate_node_index_order(nodes=nodes) diagnostics.extend(b) dag_ok, b = _validate_dag(nodes=nodes) diagnostics.extend(b) refs_ok, b = _validate_max_refs_per_bg(nodes=nodes) diagnostics.extend(b) two_ok, b = _validate_two_refs_distinct_spaces(nodes=nodes) diagnostics.extend(b) fp_ok, b = _validate_same_fp_only( fp_id=fp_id, nodes=nodes, same_fp_bg_ids=same_fp_bg_ids ) diagnostics.extend(b) anchor_ok, b = _validate_anchor_exactly_one(nodes=nodes) diagnostics.extend(b) anchor_clean_ok, b = _validate_anchor_in_clean_candidate_set( nodes=nodes, clean_anchor_candidate_bg_ids=clean_anchor_candidate_bg_ids, ) diagnostics.extend(b) cam_ok, b = _validate_camera_in_candidates( nodes=nodes, geometry=geometry ) diagnostics.extend(b) rat_ok, b = _validate_rationale_nonempty(nodes=nodes) diagnostics.extend(b) synth_ok, b = _validate_synthetic_readback_block( readback_status=readback_status ) diagnostics.extend(b) all_ok = all( [ shape_ok, completeness_ok, node_index_ok, dag_ok, refs_ok, two_ok, fp_ok, anchor_ok, anchor_clean_ok, cam_ok, rat_ok, ] ) return { "required_shape_ok": shape_ok, "graph_completeness_ok": completeness_ok, "node_index_order_ok": node_index_ok, "dag_ok": dag_ok, "max_refs_per_bg_ok": refs_ok, "two_refs_distinct_spaces_ok": two_ok, "same_fp_only_ok": fp_ok, "anchor_exactly_one_ok": anchor_ok, "anchor_in_clean_candidate_set_ok": anchor_clean_ok, "camera_in_candidates_ok": cam_ok, "rationale_and_mode_ok": rat_ok, "synthetic_readback_production_clear": synth_ok, "all_validators_passed": all_ok, "diagnostics": diagnostics, } # ───────────────────────────────────────────────────────────────────── # Commit 2a — render_action router (deterministic, post-validation) # ───────────────────────────────────────────────────────────────────── def _low_delta_marker_set( per_bg_facts: Dict[str, Any], bg_id: Any ) -> FrozenSet[int]: """Deterministic target-unit marker fingerprint for a bg, drawn from the dossier's ``per_bg_render_facts_by_bg_id`` (the SOT). LLM ``physical_space_id`` is diagnostic only and never consulted here.""" if not isinstance(bg_id, str): return frozenset() facts = per_bg_facts.get(bg_id) or {} out: set = set() for n in facts.get("target_unit_marker_numbers") or []: if isinstance(n, int) and not isinstance(n, bool): out.add(n) dom = facts.get("dominant_target_unit_marker_number") if isinstance(dom, int) and not isinstance(dom, bool): out.add(dom) return frozenset(out) def _corroborates_same_space( *, current: Dict[str, Any], target: Dict[str, Any], per_bg_facts: Dict[str, Any], ) -> bool: """Deterministic same-physical-space corroboration between a reuse candidate and its target plate. ``camera_unit`` equality is the hard floor — a reuse may never cross units. On top of that we require EITHER the same ``look_at_unit`` OR an overlap in the dossier target-unit markers, so that two plates that sit in the same unit but face genuinely different walls/targets (the "same space, very different view" failure mode) are NOT collapsed. """ cur_cam = (current.get("camera_decision") or {}).get("camera_unit") tgt_cam = (target.get("camera_decision") or {}).get("camera_unit") if ( isinstance(cur_cam, bool) or isinstance(tgt_cam, bool) or not isinstance(cur_cam, int) or not isinstance(tgt_cam, int) or cur_cam != tgt_cam ): return False cur_look = (current.get("camera_decision") or {}).get("look_at_unit") tgt_look = (target.get("camera_decision") or {}).get("look_at_unit") same_look = ( isinstance(cur_look, int) and not isinstance(cur_look, bool) and cur_look == tgt_look ) overlap = bool( _low_delta_marker_set(per_bg_facts, current.get("bg_id")) & _low_delta_marker_set(per_bg_facts, target.get("bg_id")) ) return same_look or overlap def route_render_actions( *, nodes: List[Dict[str, Any]], dossier: Dict[str, Any] ) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]: """Stamp canonical ``render_action`` / ``reuse_target_bg_id`` onto each plan node after LLM-output validation, before the checkpoint write. A node is routed to ``reuse_existing_plate`` ONLY when ALL hold: - it is not the dwelling identity anchor, - its ``mode`` is in :data:`REUSE_ELIGIBLE_MODES`, - the LLM set ``same_physical_space_low_delta_candidate=true`` with a non-empty ``low_delta_reuse_target_bg_id`` (LLM signal — required), - that target is an EARLIER same-fp node whose own canonical ``render_action`` is ``render_new_plate`` (no reuse-of-reuse; the earlier-only rule makes the reuse graph acyclic by construction), - deterministic same-space corroboration passes (code signal — required). Neither the LLM flag nor the code overlap may trigger reuse on its own. Otherwise the node stays ``render_new_plate`` with an empty ``reuse_target_bg_id``. Contract/structure violations (candidate↔target coupling, invalid target, text-only same-space follow-up) fail the routing (``render_action_routing_ok=False``); a valid-but-uncorroborated candidate, an over-signalled ineligible node, **or a reuse-of-reuse request** is a non-fail downgrade with a diagnostic — the node keeps its references and renders a new plate. """ per_bg_facts = (dossier or {}).get("per_bg_render_facts_by_bg_id") or {} diagnostics: List[str] = [] routing_ok = True reuse_count = 0 # node_index ascending → a reuse target is always decided before the # node that points at it. def _idx_key(i: int) -> Any: ni = nodes[i].get("node_index") return ni if isinstance(ni, int) and not isinstance(ni, bool) else i order = sorted(range(len(nodes)), key=_idx_key) routed: List[Optional[Dict[str, Any]]] = [None] * len(nodes) decided_by_bg: Dict[str, Dict[str, Any]] = {} for i in order: node = dict(nodes[i]) ref = node.get("reference_decision") or {} bg_id = node.get("bg_id") mode = node.get("mode") is_anchor = node.get("is_dwelling_identity_anchor") is True candidate = ( ref.get("same_physical_space_low_delta_candidate") is True ) target_id = ref.get("low_delta_reuse_target_bg_id") ld_rationale = ref.get("low_delta_rationale") target_id = target_id if isinstance(target_id, str) else "" ld_rationale = ld_rationale if isinstance(ld_rationale, str) else "" node["render_action"] = RENDER_ACTION_NEW node["reuse_target_bg_id"] = "" # v2 semantic coupling — mode-independent hard fails. if candidate and not target_id: routing_ok = False diagnostics.append( f"bg_id={bg_id!r} candidate=true but " f"low_delta_reuse_target_bg_id is empty (malformed)" ) if (not candidate) and (target_id or ld_rationale): routing_ok = False diagnostics.append( f"bg_id={bg_id!r} candidate=false but low_delta target/" f"rationale non-empty (v2 empty-sentinel contract " f"violation)" ) # Eligible modes must carry a reference — no text-only same-space # follow-up, on either render path. Hard fail. if mode in REUSE_ELIGIBLE_MODES: sel = ref.get("selected_refs") if not isinstance(sel, list) or not sel: routing_ok = False diagnostics.append( f"bg_id={bg_id!r} mode={mode!r} requires a non-empty " f"selected_refs (text-only same-space follow-up " f"forbidden)" ) # Routing decision (only meaningful when the LLM signalled a # candidate with a target id). if candidate and target_id: if is_anchor or mode not in REUSE_ELIGIBLE_MODES: diagnostics.append( f"bg_id={bg_id!r} low-delta candidate on reuse-" f"ineligible node (anchor={is_anchor}, mode={mode!r}) " f"→ downgraded to render_new_plate" ) elif target_id == bg_id: routing_ok = False diagnostics.append( f"bg_id={bg_id!r} reuse target is self (must be an " f"earlier node)" ) else: tgt = decided_by_bg.get(target_id) if tgt is None: routing_ok = False diagnostics.append( f"bg_id={bg_id!r} reuse target {target_id!r} is " f"not an earlier same-fp node" ) elif tgt.get("render_action") != RENDER_ACTION_NEW: # ★2026-09-19 — **내림이지 실패가 아니다.** 앞서 두 갈래 # (reuse 부적격 노드 · same-space 미확증)와 같은 처리다. # reuse-of-reuse 는 malformed 가 아니라 **허용 안 되는 # 요청**이고, 보수적으로 읽는 길이 하나뿐이다 — 이 노드는 # 제 참조를 그대로 들고 **새 plate 를 그린다**. # 실측: fp 44개 중 이 한 줄 때문에 fp_container_interior # 가 통째로 죽었고(ok 42·failed 1), provider 재시도는 # routing 이전 원시 출력만 보므로 재질문도 안 열렸다. # ★자기 참조·없는 대상·필수 참조 결손은 **그대로 실패**다 # (무슨 뜻인지 알 수 없으니 고를 수가 없다). diagnostics.append( f"bg_id={bg_id!r} reuse target {target_id!r} is " f"itself a reuse node (reuse-of-reuse forbidden) " f"→ downgraded to render_new_plate" ) elif not _corroborates_same_space( current=node, target=tgt, per_bg_facts=per_bg_facts ): diagnostics.append( f"bg_id={bg_id!r} LLM low-delta candidate not " f"corroborated by deterministic same-space overlap " f"(camera_unit + look_at_unit / target-marker) " f"→ downgraded to render_new_plate" ) else: node["render_action"] = RENDER_ACTION_REUSE node["reuse_target_bg_id"] = target_id reuse_count += 1 routed[i] = node if isinstance(bg_id, str): decided_by_bg[bg_id] = node routed_nodes = [ n if n is not None else dict(nodes[idx]) for idx, n in enumerate(routed) ] return routed_nodes, { "render_action_routing_ok": routing_ok, "reuse_plate_count": reuse_count, "reuse_count": reuse_count, "diagnostics": diagnostics, } # ───────────────────────────────────────────────────────────────────── # W21B-w4 C3 — projection-card enrichment (deterministic, post-routing) # ───────────────────────────────────────────────────────────────────── # # After ``route_render_actions`` stamps canonical render_action / # reuse_target_bg_id, the plan vNext stamps the per-node projection-card # fields the BG (background_prompt vNext, C4) and scene (scene_detail # vNext, C5+) consumers read so they follow one shot contract (wiring # brief §5/§8/§9). This is a DETERMINISTIC read of the # ``shot_projection_card`` checkpoint index the step assembles — NO LLM / # VLM / image / DB. The C2 gate already decided each card's state; this # pass never re-derives card content. # # ``camera_decision`` is DEMOTED to a compat / audit field by this wave # (NOT deleted — the validator/router/legacy consumers still read it). # These projection-card fields are the forward projection SOT. # plan-level ``projection_card_state`` vocabulary (Codex C3 consensus #1): # pass / needs_review / blocked — a real card resolved to this gate state # not_available — projection subsystem unavailable (card step OFF / # checkpoint absent / card_index None), OR a reuse target # whose projection could not be resolved (fail-closed) # not_applicable — this bg is not a projection target (no card for any of # its shots; e.g. fp-less / direct-plate BG) _CARD_STATE_PRIORITY: Dict[str, int] = {"pass": 3, "needs_review": 2, "blocked": 1} # Additive fields stamped onto every node (existing node shape untouched). PROJECTION_CARD_FIELDS: Tuple[str, ...] = ( "anchor_shot_id", "projection_card_id", "projection_card_hash", "projection_card_state", "projection_card_fallback_reason", "projection_card_source_bg_id", "projection_card_inherited", ) def _select_anchor_card( *, bg_id: str, card_index: Dict[str, Dict[str, Dict[str, Any]]], per_bg_readiness: Dict[str, Dict[str, Any]], ) -> Tuple[Optional[str], Optional[Dict[str, Any]]]: """Pick the anchor ``(shot_id, card)`` for one bg from its shot cards. Among the bg's ``applies_to_shots`` that carry a card entry, choose the best ``card_state`` (pass > needs_review > blocked); ties break on ``applies_to_shots`` order (earliest wins). Returns ``(None, None)`` when the bg has no card target. """ by_shot = (card_index or {}).get(bg_id) or {} applies = (per_bg_readiness.get(bg_id) or {}).get("applies_to_shots") or [] best_shot: Optional[str] = None best_card: Optional[Dict[str, Any]] = None best_pri = -1 for shot_id in applies: card = by_shot.get(str(shot_id)) if not isinstance(card, dict): continue pri = _CARD_STATE_PRIORITY.get(card.get("card_state"), 0) if pri > best_pri: best_pri = pri best_shot = str(shot_id) best_card = card return best_shot, best_card def _resolve_self_projection( *, bg_id: str, card_index: Optional[Dict[str, Dict[str, Dict[str, Any]]]], per_bg_readiness: Dict[str, Dict[str, Any]], ) -> Dict[str, Any]: """Resolve the non-reuse (self) projection fields for one bg node.""" if card_index is None: return { "anchor_shot_id": "", "projection_card_id": "", "projection_card_hash": "", "projection_card_state": "not_available", "projection_card_fallback_reason": "projection_subsystem_unavailable", "projection_card_source_bg_id": bg_id, "projection_card_inherited": False, } anchor_shot_id, card = _select_anchor_card( bg_id=bg_id, card_index=card_index, per_bg_readiness=per_bg_readiness ) if card is None: return { "anchor_shot_id": "", "projection_card_id": "", "projection_card_hash": "", "projection_card_state": "not_applicable", "projection_card_fallback_reason": "no_projection_card_target", "projection_card_source_bg_id": bg_id, "projection_card_inherited": False, } state = card.get("card_state") card_id = str(card.get("card_id") or "") raw_fallback = str(card.get("fallback_reason") or "") if state not in _CARD_STATE_PRIORITY: # An error / shape-broken card entry fails closed to blocked; keep a # non-empty reason so downstream never sees a blocked card with no # explanation. state = "blocked" fallback = raw_fallback or "projection_card_invalid" elif state in ("pass", "needs_review") and not card_id: # FAIL-CLOSED (Codex C3 review Required 1): a pass / needs_review # card with no content-addressed id cannot be cross-checked # downstream (C4 uses id == hash == card_id). An empty id is a false # pass, so the card is not trustworthy → blocked. state = "blocked" fallback = "missing_projection_card_id" elif state == "pass": fallback = "" else: # needs_review / blocked with a valid id — keep the card's reason. fallback = raw_fallback return { "anchor_shot_id": anchor_shot_id or "", "projection_card_id": card_id, # content-addressed card_id doubles as the integrity hash in v0 # (id == hash); a distinct prose digest is a later schema bump. "projection_card_hash": card_id, "projection_card_state": state, "projection_card_fallback_reason": fallback, "projection_card_source_bg_id": bg_id, "projection_card_inherited": False, } def enrich_nodes_with_projection_cards( *, nodes: List[Dict[str, Any]], card_index: Optional[Dict[str, Dict[str, Dict[str, Any]]]], per_bg_readiness: Dict[str, Dict[str, Any]], ) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]: """Stamp per-node projection-card fields onto a routed plan (C3). ``card_index``: ``{bg_id: {shot_id: {card_state, card_id, fallback_reason}}}`` assembled by the step from the ``shot_projection_card`` checkpoint, or ``None`` when the projection subsystem is unavailable (card step OFF / checkpoint absent). No card content is re-derived here — the C2 gate already ran. Reuse inheritance (wiring brief §8 decision-lock, Codex C3 consensus #2) is FAIL-CLOSED: a ``reuse_existing_plate`` node copies the resolved projection fields of its render target (the bg that actually produced the pixels), then re-points provenance with ``projection_card_source_bg_id=reuse_target_bg_id`` and ``projection_card_inherited=true``. It NEVER forges a ``pass`` — a blocked / needs_review target is inherited verbatim, and a target whose projection cannot be resolved yields ``not_available``. Returns ``(nodes, summary)``. Nodes are mutated in place — additive fields only; the existing node shape is untouched. Iteration runs in ``node_index`` order so a reuse target (always an earlier render_new_plate node, per the router) is resolved before any child that inherits from it. """ resolved_by_bg: Dict[str, Dict[str, Any]] = {} counts: Dict[str, int] = {} for node in sorted(nodes, key=lambda n: n.get("node_index", 0)): bg_id = str(node.get("bg_id") or "") render_action = node.get("render_action") reuse_target = str(node.get("reuse_target_bg_id") or "") if render_action == RENDER_ACTION_REUSE and reuse_target: target = resolved_by_bg.get(reuse_target) if target is None: # Target absent from the resolved nodes entirely. fields = { "anchor_shot_id": "", "projection_card_id": "", "projection_card_hash": "", "projection_card_state": "not_available", "projection_card_fallback_reason": "reuse_target_missing", "projection_card_source_bg_id": reuse_target, "projection_card_inherited": True, } elif target["projection_card_state"] not in _CARD_STATE_PRIORITY: # Target exists but carries NO usable card (its own state is # not_available / not_applicable). The reused pixels have no # projection to inherit → fail-closed not_available, never a # silently-inherited not_applicable (Codex C3 review Required # 2). fields = { "anchor_shot_id": "", "projection_card_id": "", "projection_card_hash": "", "projection_card_state": "not_available", "projection_card_fallback_reason": "reuse_target_projection_missing", "projection_card_source_bg_id": reuse_target, "projection_card_inherited": True, } else: # Inherit the target's RESOLVED projection verbatim (no # forged pass), then re-point provenance at the reuse target. fields = dict(target) fields["projection_card_source_bg_id"] = reuse_target fields["projection_card_inherited"] = True else: fields = _resolve_self_projection( bg_id=bg_id, card_index=card_index, per_bg_readiness=per_bg_readiness, ) for key in PROJECTION_CARD_FIELDS: node[key] = fields[key] resolved_by_bg[bg_id] = fields state = fields["projection_card_state"] counts[state] = counts.get(state, 0) + 1 return nodes, {"projection_card_state_counts": counts} # ───────────────────────────────────────────────────────────────────── # 3a — self-card base-band prep + card-aware route_render_actions_v2 # # brief docs/w21b-wave4-bg-plate-partition-refdag-brief-20260531 (APPROVED v0.1): # ③ self-card base-band prep (reuse 무관) → ④ card-aware route_render_actions_v2. # D5 FINAL lock (dry §10-2 real-card 실측): v0 hard-withhold = horizontal # left↔right only (both visibility∈{visible,partial}, both confidence>=THRESHOLD, # same base marker present in both cards). depth foreground↔background 모순은 # DIAGNOSTIC-ONLY (canonical render_action 불변) — 같은 bg 두 샷이 conf 0.80~0.88 # 에도 depth 만 갈리는 VLM depth-band instability 가 dry 에서 관측됨. # ───────────────────────────────────────────────────────────────────── def extract_base_bands(card_entry: Dict[str, Any]) -> Dict[int, Tuple[Any, Any, Any, Any]]: """{marker_number: (horizontal_band, depth_band, visibility, confidence)} for the ``base`` markers of a C2 card envelope. transient / ignored_state_overlay markers are dropped; only integer marker_number entries are kept.""" vlm = ((card_entry or {}).get("card") or {}).get("vlm_output") or {} out: Dict[int, Tuple[Any, Any, Any, Any]] = {} for it in (vlm.get("visible_items") or []): if it.get("marker_layer") != "base": continue mn = it.get("marker_number") if not isinstance(mn, int) or isinstance(mn, bool): continue out[mn] = ( it.get("horizontal_band"), it.get("depth_band"), it.get("visibility"), it.get("confidence"), ) return out # D5 FINAL lock parameters (dry-calibrated; surfaced via config at the step). BAND_CONFIDENCE_THRESHOLD: float = 0.6 _BAND_VISIBILITY_OK: FrozenSet[str] = frozenset({"visible", "partial"}) _HORIZONTAL_OPPOSITE: FrozenSet[Tuple[str, str]] = frozenset( {("left", "right"), ("right", "left")} ) _DEPTH_OPPOSITE: FrozenSet[Tuple[str, str]] = frozenset( {("foreground", "background"), ("background", "foreground")} ) def base_band_contradictions( cur_bands: Dict[int, Tuple[Any, Any, Any, Any]], tgt_bands: Dict[int, Tuple[Any, Any, Any, Any]], conf_threshold: float = BAND_CONFIDENCE_THRESHOLD, ) -> Tuple[List[Tuple], List[Tuple], List[Tuple]]: """D5 corroboration between two cards' base bands. Returns ``(hard, depth_diagnostics, guarded)``: - ``hard``: horizontal ``left↔right`` contradictions that pass the D5 gate (same base marker both sides, both visibility∈{visible,partial}, both confidence>=threshold). These DO withhold a reuse. - ``depth_diagnostics``: depth ``foreground↔background`` contradictions that pass the same gate. DIAGNOSTIC-ONLY — canonical render_action unchanged. - ``guarded``: opposite-band markers excluded by visibility / confidence. """ shared = set(cur_bands) & set(tgt_bands) hard: List[Tuple] = [] depth_diag: List[Tuple] = [] guarded: List[Tuple] = [] for mn in sorted(shared): h1, d1, v1, c1 = cur_bands[mn] h2, d2, v2, c2 = tgt_bands[mn] opp_h = (h1, h2) in _HORIZONTAL_OPPOSITE opp_d = (d1, d2) in _DEPTH_OPPOSITE if not (opp_h or opp_d): continue # D5 gate (applies to both hard and diagnostic). if v1 not in _BAND_VISIBILITY_OK or v2 not in _BAND_VISIBILITY_OK: guarded.append((mn, "visibility", v1, v2)) continue if ( not isinstance(c1, (int, float)) or isinstance(c1, bool) or not isinstance(c2, (int, float)) or isinstance(c2, bool) or c1 < conf_threshold or c2 < conf_threshold ): guarded.append((mn, "confidence", c1, c2)) continue if opp_h: # horizontal left↔right → v0 HARD withhold hard.append((mn, "horizontal", (h1, d1), (h2, d2), (c1, c2))) else: # depth foreground↔background → diagnostic only depth_diag.append((mn, "depth", (h1, d1), (h2, d2), (c1, c2))) return hard, depth_diag, guarded def count_corroborating_base_markers( cur_bands: Dict[int, Tuple[Any, Any, Any, Any]], tgt_bands: Dict[int, Tuple[Any, Any, Any, Any]], conf_threshold: float = BAND_CONFIDENCE_THRESHOLD, ) -> int: """Count base markers that POSITIVELY corroborate two cards' shared POV. A marker counts only when it is present on both sides, both visibility∈{visible,partial}, both confidence>=threshold, and its ``horizontal_band`` does NOT conflict (left↔right). This is the positive evidence ``shareable_card_corroborated`` requires (brief §5) — two pass cards with no shared base marker, or only guarded markers, yield 0 (geometry-only). depth agreement is not required (depth is diagnostic-only). """ count = 0 for mn in set(cur_bands) & set(tgt_bands): h1, _d1, v1, c1 = cur_bands[mn] h2, _d2, v2, c2 = tgt_bands[mn] if v1 not in _BAND_VISIBILITY_OK or v2 not in _BAND_VISIBILITY_OK: continue if ( not isinstance(c1, (int, float)) or isinstance(c1, bool) or not isinstance(c2, (int, float)) or isinstance(c2, bool) or c1 < conf_threshold or c2 < conf_threshold ): continue if (h1, h2) in _HORIZONTAL_OPPOSITE: continue # a conflict is not corroboration count += 1 return count def _self_card_base_bands( *, bg_id: str, card_content_index: Dict[str, Dict[str, Dict[str, Any]]], per_bg_readiness: Dict[str, Dict[str, Any]], ) -> Optional[Dict[int, Tuple[Any, Any, Any, Any]]]: """③ self-card band prep — base bands of one bg's anchor card, or None. Returns ``None`` (band evidence unavailable → withhold impossible, fail-safe to reuse) UNLESS the anchor card is a trustworthy ``pass`` card whose envelope ``card_id`` re-joins the index entry's ``card_id`` (brief §7-C, D2: pass only, no forged-pass band). Reuse is decided BEFORE C3 enrich, so this selects the anchor itself rather than reading a stamped ``anchor_shot_id``. """ _shot, entry = _select_anchor_card( bg_id=bg_id, card_index=card_content_index, per_bg_readiness=per_bg_readiness ) if not isinstance(entry, dict): return None if entry.get("card_state") != "pass": return None # D2 — only a pass card's bands are trusted card_id = entry.get("card_id") envelope = entry.get("card") or {} # re-join: index pointer must match the content-addressed envelope id. if not card_id or envelope.get("card_id") != card_id: return None bands = extract_base_bands(entry) return bands or None def route_render_actions_v2( *, nodes: List[Dict[str, Any]], dossier: Dict[str, Any], card_content_index: Optional[Dict[str, Dict[str, Dict[str, Any]]]] = None, per_bg_readiness: Optional[Dict[str, Dict[str, Any]]] = None, conf_threshold: float = BAND_CONFIDENCE_THRESHOLD, ) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]: """④ card-aware router — geometry route + horizontal-band withhold. Runs the deterministic geometry/LLM-signal router first, then (only when a projection card content index is available) re-decides each ``reuse_existing_plate`` node: if the reuse node's anchor card and its render target's anchor card carry a D5 horizontal ``left↔right`` contradiction, the canonical render_action is WITHHELD to ``render_new_plate`` (the withhold is the canonical SOT — brief §4 order ④). depth conflicts are recorded as a diagnostic only. With no card index this is a behavioral no-op == baseline ``route_render_actions`` (brief §10 default byte-identical). A reuse node is never the reuse target of another node (router earlier-only rule), so withholding it never invalidates a downstream inheritance. WITHHOLD-ONLY (Finding 2, 2a lock): this pass only CLOSES reuses; it never re-opens a reuse the baseline router already rejected. If A→new, B reuses A but is card-withheld to new, and C low-delta-targets B, the baseline router has already downgraded C→B as reuse-of-reuse (render_new_plate + diagnostic, 2026-09-19: **내림이지 실패가 아니다**); that is PRESERVED here — C stays render_new_plate, fail-closed. Re-opening a withheld target's downstream reuse is future route_v3 / 2-pass router work. """ routed, summary = route_render_actions(nodes=nodes, dossier=dossier) summary["card_withhold_count"] = 0 if not card_content_index: return routed, summary readiness = per_bg_readiness or {} band_by_bg: Dict[str, Optional[Dict[int, Tuple[Any, Any, Any, Any]]]] = {} def _bands(bg: Optional[str]) -> Optional[Dict[int, Tuple[Any, Any, Any, Any]]]: if not isinstance(bg, str) or not bg: return None if bg not in band_by_bg: band_by_bg[bg] = _self_card_base_bands( bg_id=bg, card_content_index=card_content_index, per_bg_readiness=readiness, ) return band_by_bg[bg] events: List[Dict[str, Any]] = [] for n in routed: if n.get("render_action") != RENDER_ACTION_REUSE: continue cur_b = _bands(n.get("bg_id")) tgt_b = _bands(n.get("reuse_target_bg_id")) if not cur_b or not tgt_b: continue hard, depth_diag, _guarded = base_band_contradictions( cur_b, tgt_b, conf_threshold ) if hard: tgt = n.get("reuse_target_bg_id") n["render_action"] = RENDER_ACTION_NEW n["reuse_target_bg_id"] = "" n["card_withhold_reason"] = "horizontal_band_conflict" events.append({ "bg_id": n.get("bg_id"), "withheld_reuse_target_bg_id": tgt, "markers": [h[0] for h in hard], }) else: # reuse kept. Mark card-corroborated ONLY when there is positive # shared-base evidence (Finding 1) — no shared marker / all-guarded # stays geometry-only in ⑥ mirror. if count_corroborating_base_markers(cur_b, tgt_b, conf_threshold) > 0: n["plate_corroboration"] = "card_corroborated" if depth_diag: # depth conflict is diagnostic only (D5 depth = diagnostic-only). n["depth_band_conflict_diagnostic"] = [d[0] for d in depth_diag] summary["card_withhold_count"] = len(events) if events: summary["card_withhold_events"] = events # A withhold flips a node reuse→new, so the baseline reuse counts are now # stale. Recompute from the post-withhold canonical render_action — these # feed validators["reuse_plate_count"] in build_render_plan_for_fp. final_reuse = sum( 1 for n in routed if n.get("render_action") == RENDER_ACTION_REUSE ) summary["reuse_plate_count"] = final_reuse summary["reuse_count"] = final_reuse return routed, summary # ───────────────────────────────────────────────────────────────────── # 3a — plate partition mirror (⑥) — PLATE_PARTITION_FIELDS # # brief §5: a deterministic, additive MIRROR of the canonical render_action / # reuse_target_bg_id (the SOT stays canonical render_action; these fields are a # higher-level view of it). Runs after C3 enrich so plate_anchor_shot_id can read # the group anchor's stamped anchor_shot_id. # ───────────────────────────────────────────────────────────────────── PLATE_PARTITION_FIELDS: Tuple[str, ...] = ( "plate_group_id", "plate_anchor_bg_id", "plate_anchor_shot_id", "plate_shareability", "needs_new_plate", ) def mirror_plate_partition( nodes: List[Dict[str, Any]], ) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]: """Stamp PLATE_PARTITION_FIELDS as a mirror of the canonical render_action. ``plate_group_id`` / ``plate_anchor_bg_id`` = the bg that produces the plate pixels (self for ``render_new_plate``, the reuse target for ``reuse_existing_plate``). ``needs_new_plate`` = ``render_action == render_new_plate``. ``plate_shareability`` (brief §5 vocab): - ``not_shareable_card_withheld`` — a node carrying ``card_withhold_reason`` (a reuse that horizontal-band conflict flipped to render_new_plate), - ``exclusive`` — a solo render_new_plate group (no reuse member), - ``shareable_card_corroborated`` / ``shareable_geometry_only`` — a group with ≥1 reuse member; ``card_corroborated`` iff any member resolved its reuse via a passing card on both sides (``plate_corroboration`` stamped by ``route_render_actions_v2``), else geometry-only. Nodes are mutated in place (additive). The render_action SOT is untouched. """ # Pass 1: per-node group id + needs_new_plate; index anchors by group. anchor_by_group: Dict[str, Dict[str, Any]] = {} members_by_group: Dict[str, List[Dict[str, Any]]] = {} for n in nodes: bg_id = n.get("bg_id") is_reuse = n.get("render_action") == RENDER_ACTION_REUSE group = n.get("reuse_target_bg_id") if is_reuse else bg_id group = group if isinstance(group, str) and group else bg_id n["plate_group_id"] = group n["plate_anchor_bg_id"] = group n["needs_new_plate"] = not is_reuse if is_reuse: members_by_group.setdefault(group, []).append(n) elif not n.get("card_withhold_reason"): # the fresh-plate node whose bg IS the group is the group anchor. if group == bg_id: anchor_by_group[group] = n # Pass 2: shareability + plate_anchor_shot_id (group anchor's anchor_shot_id). for n in nodes: group = n["plate_group_id"] anchor = anchor_by_group.get(group) n["plate_anchor_shot_id"] = ( str(anchor.get("anchor_shot_id") or "") if anchor else str(n.get("anchor_shot_id") or "") ) if n.get("card_withhold_reason"): n["plate_shareability"] = "not_shareable_card_withheld" continue members = members_by_group.get(group) or [] if n["needs_new_plate"]: if not members: n["plate_shareability"] = "exclusive" else: corroborated = any( m.get("plate_corroboration") == "card_corroborated" for m in members ) n["plate_shareability"] = ( "shareable_card_corroborated" if corroborated else "shareable_geometry_only" ) else: # reuse member n["plate_shareability"] = ( "shareable_card_corroborated" if n.get("plate_corroboration") == "card_corroborated" else "shareable_geometry_only" ) from collections import Counter as _Counter summary = { "plate_group_count": len({n["plate_group_id"] for n in nodes}), "plate_shareability_counts": dict( _Counter(n["plate_shareability"] for n in nodes) ), } return nodes, summary # ───────────────────────────────────────────────────────────────────── # 3b — reference DAG construction (⑦) — REFERENCE_DAG_FIELDS # # brief §6: over the fresh-plate set (needs_new_plate True), make the render # DAG explicit — which earlier plates each fresh plate conditions on, each ref's # role, and a topological render order. Reuse aliases (needs_new_plate False) are # NOT DAG nodes. The earlier-only router rule already guarantees acyclicity, so # render order is the node_index topological order. # ───────────────────────────────────────────────────────────────────── REFERENCE_DAG_FIELDS: Tuple[str, ...] = ( "ref_tree_parents", "ref_role_per_parent", "render_order_index", "max_refs", ) MAX_REFS_PER_BG: int = 2 _MODE_REF_ROLE: Dict[str, str] = { "same_physical_space_view": "space_continuity", "reference_derived": "space_continuity", "related_style_new_space": "style", "two_refs_distinct_spaces": "style", } def build_reference_dag( nodes: List[Dict[str, Any]], ) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]: """Stamp REFERENCE_DAG_FIELDS onto the fresh-plate nodes (⑦). Only ``needs_new_plate`` nodes are DAG nodes. ``ref_tree_parents`` are the ``selected_refs`` canonicalized to their rendered-plate anchor bg ids (each ref mapped through ⑥ ``plate_anchor_bg_id`` so a reuse alias resolves to the fresh plate it copies), deduped and capped at ``MAX_REFS_PER_BG`` distinct parents; ``ref_role_per_parent`` maps each parent to ``style`` / ``space_continuity`` derived deterministically from the node ``mode``; ``render_order_index`` is the node_index topological order among fresh plates. """ fresh = [n for n in nodes if n.get("needs_new_plate")] def _idx(n: Dict[str, Any]) -> int: ni = n.get("node_index") return ni if isinstance(ni, int) and not isinstance(ni, bool) else 0 for order_i, n in enumerate(sorted(fresh, key=_idx)): n["render_order_index"] = order_i # ⑦ parents must point at RENDERED plates, never reuse aliases. ``selected_refs`` # is pre-route LLM output, so a ref_bg_id can name a bg that the router later # turned into a ``reuse_existing_plate`` alias (which produces no pixels and is # not a DAG node). ⑥ mirror already stamped ``plate_anchor_bg_id`` = the bg # whose pixels each plate uses (self for fresh, the reuse target for an alias), # so canonicalize every ref through it, then dedupe (two refs may collapse to # the same anchor) and cap at MAX_REFS_PER_BG DISTINCT rendered-plate parents. # (Codex wiring-review Required, 2026-05-31.) plate_anchor_by_bg: Dict[str, str] = { n["bg_id"]: str(n.get("plate_anchor_bg_id") or n["bg_id"]) for n in nodes if isinstance(n.get("bg_id"), str) and n["bg_id"] } for n in fresh: ref_dec = n.get("reference_decision") or {} selected = ref_dec.get("selected_refs") or [] role = _MODE_REF_ROLE.get(n.get("mode")) self_bg = n.get("bg_id") parents: List[str] = [] roles: Dict[str, str] = {} for ref in selected: rb = ref.get("ref_bg_id") if not isinstance(rb, str) or not rb: continue canon = plate_anchor_by_bg.get(rb, rb) # an alias whose plate is this very node would be a self-edge — the # router's earlier-only rule makes this impossible, but guard anyway. if canon == self_bg or canon in parents: continue parents.append(canon) if role: roles[canon] = role if len(parents) >= MAX_REFS_PER_BG: break n["ref_tree_parents"] = parents n["ref_role_per_parent"] = roles n["max_refs"] = MAX_REFS_PER_BG return nodes, {"dag_node_count": len(fresh)} # ───────────────────────────────────────────────────────────────────── # W21B-w5 STEP4 (A wiring) — space_partition_plan canonical normalizer (⑧) # # The bg_space_partition step (order 21.594) produces the LLM space-partition # SOT: which bgs share a physical space, their anchor, the render_action per bg, # and the bounded reference tree. This normalizer is the FINAL authority over the # canonical render surface — when a usable plan is present it OVERRIDES every # downstream-read field from the partition SOT, demoting the wave4 geometry route # / mirror / DAG output to ``geometry_*_diagnostic`` (never read downstream). # There is then exactly ONE canonical SOT (Codex A2 + 5-condition lock, w5). # # fail-closed: a missing / unusable plan (or one that does not cover every node) # is a NO-OP — the wave4 geometry surface is left exactly as produced and every # node records a ``partition_fallback_reason``. NEVER half-consumes per node. # ───────────────────────────────────────────────────────────────────── #: canonical surface fields the normalizer overrides AND mirrors to a #: ``geometry__diagnostic`` key before overriding (Codex condition 1+2). _PARTITION_CANONICAL_FIELDS: Tuple[str, ...] = ( "render_action", "reuse_target_bg_id", "needs_new_plate", "ref_tree_parents", "ref_role_per_parent", "render_order_index", "max_refs", "plate_group_id", "plate_anchor_bg_id", "plate_shareability", ) RENDER_ACTION_SOURCE_PARTITION: str = "space_partition_plan" RENDER_ACTION_SOURCE_GEOMETRY: str = "geometry_route" #: partition ref parents are within-group same-space anchors. _PARTITION_REF_ROLE: str = "space_continuity" def _partition_plan_is_usable( space_partition_plan: Optional[Dict[str, Any]], node_bg_ids: FrozenSet[str] ) -> Tuple[bool, str]: """Decide whether the plan can drive the canonical surface (fail-closed). Usable iff it is a dict whose ``render_actions`` AND ``node_assignments`` both cover EVERY plan-node bg, AND ``plate_groups`` provides an anchor for every group those bgs are assigned to. The override consumes all three, so checking only ``render_actions`` would let a partial plan proceed with silent fallback-like defaults (``gid=bg`` / ``anchor=bg``) — a half-consume that can mis-group a reuse node (Codex review hardening). Returns ``(usable, fallback_reason)``; the reason is empty when usable. """ if not isinstance(space_partition_plan, dict): return False, "partition_plan_missing" render_actions = space_partition_plan.get("render_actions") if not isinstance(render_actions, dict) or not render_actions: return False, "partition_plan_no_render_actions" uncovered = [bg for bg in node_bg_ids if bg not in render_actions] if uncovered: return False, f"partition_plan_uncovered_bgs:{','.join(sorted(uncovered))}"[:200] node_assignments = space_partition_plan.get("node_assignments") if not isinstance(node_assignments, dict): return False, "partition_plan_no_node_assignments" unassigned = [bg for bg in node_bg_ids if bg not in node_assignments] if unassigned: return False, f"partition_plan_unassigned_bgs:{','.join(sorted(unassigned))}"[:200] # every group a plan-node bg is assigned to must have an anchor in plate_groups. anchored_groups = { g.get("plate_group_id") for g in (space_partition_plan.get("plate_groups") or []) if isinstance(g, dict) and g.get("anchor_bg_id") } missing_groups = sorted( {node_assignments[bg] for bg in node_bg_ids} - anchored_groups ) if missing_groups: return False, f"partition_plan_missing_group_anchor:{','.join(map(str, missing_groups))}"[:200] return True, "" def apply_space_partition_plan( nodes: List[Dict[str, Any]], space_partition_plan: Optional[Dict[str, Any]], ) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]: """⑧ Apply the space-partition plan. Partition owns grouping / anchor lineage / cross-group reuse veto; shot_aware (camera-aware) owns the render_action. When ``space_partition_plan`` is usable (covers every node), per node it preserves the prior geometry/wave4 value of each ``_PARTITION_CANONICAL_FIELDS`` field under ``geometry__diagnostic`` and then — * ``render_action`` / ``reuse_target_bg_id`` / ``needs_new_plate`` KEEP the shot_aware camera-aware decision; partition NEVER promotes ``render_new_plate`` -> reuse. It only vetoes a reuse whose target lands in a DIFFERENT partition group (cross-zone false reuse) back to render_new_plate. Same-group reuse is preserved; same-group NEW stays NEW with anchor lineage (= derive: same physical space, different angle, fresh plate referencing the anchor). ``partition_cross_group_reuse_vetoed`` flags a vetoed node. * ``plate_group_id`` from ``plan.node_assignments``; ``plate_anchor_bg_id`` = the group's anchor, * ``ref_tree_parents`` from ``plan.ref_tree_parents`` (already capped at max 2 anchors, copy-less reuse aliases excluded), ``ref_role_per_parent`` = ``space_continuity`` per parent, ``max_refs`` = 2, * ``plate_shareability`` = ``exclusive`` (solo group) / ``shareable_partition`` (grouped), * ``render_order_index`` recomputed topologically over the NEW fresh set. Each node is stamped ``render_action_source = space_partition_plan``. fail-closed: an unusable / uncovered plan leaves every field exactly as the wave4 pipeline produced it and stamps ``render_action_source = geometry_route`` + ``partition_fallback_reason`` (Codex condition 4 — no half-consume). Returns ``(nodes, summary)`` where summary carries the post-partition canonical counts (Codex condition 3) plus the applied/fallback decision. """ node_bg_ids: FrozenSet[str] = frozenset( n.get("bg_id") for n in nodes if isinstance(n.get("bg_id"), str) and n.get("bg_id") ) usable, fallback_reason = _partition_plan_is_usable( space_partition_plan, node_bg_ids ) if not usable: # NO-OP: legacy geometry surface untouched; record why (visual-gate # traceability — Codex). render_action_source = geometry_route. for n in nodes: n["render_action_source"] = RENDER_ACTION_SOURCE_GEOMETRY n["partition_fallback_reason"] = fallback_reason return nodes, _partition_summary(nodes, applied=False, fallback_reason=fallback_reason) plan = space_partition_plan # narrowed # NOTE: plan["render_actions"] is validated by _partition_plan_is_usable but is # no longer consumed here — shot_aware render_action is the SOT (partition only # vetoes cross-group reuse). Kept in the plan for audit / diagnostics. plan_ref_parents: Dict[str, Any] = plan.get("ref_tree_parents") or {} node_assignments: Dict[str, Any] = plan.get("node_assignments") or {} # group_id -> anchor_bg_id (from plate_groups); group_id -> member count. anchor_by_group: Dict[str, str] = {} size_by_group: Dict[str, int] = {} for g in plan.get("plate_groups") or []: gid = g.get("plate_group_id") if not isinstance(gid, str): continue anchor_by_group[gid] = str(g.get("anchor_bg_id") or "") size_by_group[gid] = 1 + len(g.get("member_bg_ids") or []) for n in nodes: bg = n.get("bg_id") # shot_aware (camera-aware) render_action is the SOT for reuse-vs-derive. # Capture it BEFORE the diagnostic snapshot, then let partition apply # grouping / anchor lineage / cross-group veto — but NEVER a new->reuse # promote (that false promote tiled same-space different-angle shots onto a # single anchor plate; the camera-aware router already decided reuse only # when camera_unit + look_at_unit corroborate the same framing). sa_action = n.get("render_action") or RENDER_ACTION_NEW sa_target = str(n.get("reuse_target_bg_id") or "") for f in _PARTITION_CANONICAL_FIELDS: n[f"geometry_{f}_diagnostic"] = n.get(f) gid = node_assignments.get(bg) gid = gid if isinstance(gid, str) and gid else bg anchor = anchor_by_group.get(gid, bg) # render_action = shot_aware decision + one partition override: a reuse whose # target sits in a DIFFERENT partition group is a cross-zone false reuse → # veto back to a fresh own plate (the reason this module exists). A same-group # reuse is preserved; a same-group NEW stays NEW (derive, not promote). veto_cross_group = False if sa_action == RENDER_ACTION_REUSE and sa_target: target_gid = node_assignments.get(sa_target) if isinstance(target_gid, str) and target_gid == gid: final_action, final_target = RENDER_ACTION_REUSE, sa_target else: final_action, final_target = RENDER_ACTION_NEW, "" veto_cross_group = True else: final_action, final_target = RENDER_ACTION_NEW, "" is_reuse = final_action == RENDER_ACTION_REUSE n["render_action"] = final_action n["reuse_target_bg_id"] = final_target n["needs_new_plate"] = not is_reuse n["plate_group_id"] = gid n["plate_anchor_bg_id"] = anchor # anchor lineage from partition: a non-anchor member derives from its anchor. # ref_tree_parents drives the derive even when the plate is rendered fresh # (same space, different angle = new plate that references the anchor). parents = [p for p in (plan_ref_parents.get(bg) or []) if isinstance(p, str) and p][:MAX_REFS_PER_BG] n["ref_tree_parents"] = parents n["ref_role_per_parent"] = {p: _PARTITION_REF_ROLE for p in parents} n["max_refs"] = MAX_REFS_PER_BG n["plate_shareability"] = ( "shareable_partition" if size_by_group.get(gid, 1) > 1 else "exclusive" ) n["render_action_source"] = RENDER_ACTION_SOURCE_PARTITION n["partition_fallback_reason"] = "" n["partition_cross_group_reuse_vetoed"] = veto_cross_group # render_order_index recomputed topologically over the NEW fresh set only # (a partition flip changes which nodes are fresh — Codex condition 3). fresh = [n for n in nodes if n.get("needs_new_plate")] def _idx(n: Dict[str, Any]) -> int: ni = n.get("node_index") return ni if isinstance(ni, int) and not isinstance(ni, bool) else 0 # reuse aliases are not DAG nodes → no render_order_index. for n in nodes: if not n.get("needs_new_plate"): n["render_order_index"] = None for order_i, n in enumerate(sorted(fresh, key=_idx)): n["render_order_index"] = order_i return nodes, _partition_summary(nodes, applied=True, fallback_reason="") def _partition_summary( nodes: List[Dict[str, Any]], *, applied: bool, fallback_reason: str ) -> Dict[str, Any]: """Post-partition canonical counts (Codex condition 3) recomputed from the FINAL nodes — never the legacy basis.""" from collections import Counter as _Counter reuse = sum(1 for n in nodes if n.get("render_action") == RENDER_ACTION_REUSE) fresh = sum(1 for n in nodes if n.get("needs_new_plate")) return { "space_partition_applied": applied, "space_partition_fallback_reason": fallback_reason, "reuse_plate_count": reuse, "reuse_count": reuse, "plate_group_count": len({n.get("plate_group_id") for n in nodes}), "dag_node_count": fresh, "plate_shareability_counts": dict( _Counter(n.get("plate_shareability") for n in nodes) ), } # ───────────────────────────────────────────────────────────────────── # W21B Phase 2 (2026-06-08): dwelling zone-map → zone-1-plate application. # # The dwelling_zone_map step (FP image + VLM) groups same-dwelling bgs across # camera angles the edge-judge (bg_space_partition) could not. Phase 2 turns # that grouping into the canonical render surface under the user's HARD RULE: # EXACTLY ONE plate per physical space. The zone anchor renders a fresh plate; # every same-zone non-anchor bg ALIASES that plate via reuse_existing_plate. # # This is a SEPARATE lightweight application path from apply_space_partition_plan # (Option b): the partition normalizer is left UNCHANGED (its 06-07 camera-aware # derive / cross-group veto is preserved for the bg_space_partition route). The # alias is NOT a derive — it carries NO ref_tree_parents (the derive signal); # per-shot angle / character variety is Phase 3 (scene_image i2i references the # one zone plate). select() between the two paths happens in # build_render_plan_for_fp; this path runs ONLY when a usable zone plan applies. # ───────────────────────────────────────────────────────────────────── RENDER_ACTION_SOURCE_ZONE_MAP: str = "dwelling_zone_map" def _zone_plan_is_usable( zone_plan: Optional[Dict[str, Any]], node_bg_ids: FrozenSet[str] ) -> Tuple[bool, str]: """Decide whether the dwelling zone map can drive the canonical surface. Usable iff ``zone_plan`` is a NON-synthetic dict whose ``bg_zone_assignments`` cover EVERY plan-node bg with a non-None ``zone_id``, AND those node bgs span >= 2 DISTINCT zones. A single zone has nothing to consolidate, so the legacy ``apply_space_partition_plan`` path stays byte-identical (Codex: single_zone skip). The synthetic fixture asserts nothing and must never drive the surface. Judged over the node bgs only — a zone_map may carry bgs with no staged shot. Returns ``(usable, fallback_reason)``; the reason is empty when usable. """ if not isinstance(zone_plan, dict): return False, "zone_plan_missing" if zone_plan.get("synthetic"): return False, "zone_plan_synthetic" bg_assign = zone_plan.get("bg_zone_assignments") if not isinstance(bg_assign, dict) or not bg_assign: return False, "zone_plan_no_assignments" uncovered = [bg for bg in node_bg_ids if bg not in bg_assign] if uncovered: return False, f"zone_plan_uncovered_bgs:{','.join(sorted(uncovered))}"[:200] null_zone = [ bg for bg in node_bg_ids if not ( isinstance(bg_assign.get(bg), dict) and isinstance(bg_assign[bg].get("zone_id"), str) and bg_assign[bg].get("zone_id") ) ] if null_zone: return False, f"zone_plan_null_zone:{','.join(sorted(null_zone))}"[:200] distinct_zones = {bg_assign[bg]["zone_id"] for bg in node_bg_ids} if len(distinct_zones) < 2: return False, "zone_plan_single_zone" return True, "" def select_zone_anchor( *, members: List[Dict[str, Any]], zone_id: str, clean_anchor_candidate_bg_ids: FrozenSet[str], bg_zone_assignments: Dict[str, Any], diagnostics: List[str], ) -> str: """Pick the single plate anchor for one zone from its member nodes. Preference (each tier breaks the prior tie), per the Phase 2 design: 1. a dossier clean-background candidate (the anchor plate must be clean); 2. a node already routed to a fresh plate — an identity anchor or a ``render_new_plate`` render_action — over one routed to reuse; 3. lower ``node_index`` (stable, earlier in the graph walk); 4. higher VLM zone-assignment confidence; 5. stable ``bg_id``. When NO member is a clean candidate this does NOT fail — it still returns a stable anchor and appends ``zone_anchor_no_clean_candidate:`` to ``diagnostics`` so the visual gate sees it (never hidden like a synthetic). """ def _conf(bg: str) -> float: c = (bg_zone_assignments.get(bg) or {}).get("confidence") return float(c) if isinstance(c, (int, float)) and not isinstance(c, bool) else 0.0 def _rank(n: Dict[str, Any]) -> Tuple: bg = n.get("bg_id") in_clean = bg in clean_anchor_candidate_bg_ids is_fresh = bool(n.get("is_dwelling_identity_anchor")) or ( n.get("render_action") == RENDER_ACTION_NEW ) ni = n.get("node_index") ni = ni if isinstance(ni, int) and not isinstance(ni, bool) else 0 return ( 0 if in_clean else 1, 0 if is_fresh else 1, ni, -_conf(str(bg)), str(bg), ) if not any( n.get("bg_id") in clean_anchor_candidate_bg_ids for n in members ): diagnostics.append(f"zone_anchor_no_clean_candidate:{zone_id}") return str(sorted(members, key=_rank)[0].get("bg_id")) def _zone_summary( nodes: List[Dict[str, Any]], *, applied: bool, fallback_reason: str, anchor_diagnostics: Optional[List[str]] = None, ) -> Dict[str, Any]: """Post-application counts — keys mirror ``_partition_summary`` so the build_render_plan_for_fp consumer reads either path uniformly, plus the zone-specific ``zone_map_applied`` / ``zone_anchor_diagnostics``.""" from collections import Counter as _Counter reuse = sum(1 for n in nodes if n.get("render_action") == RENDER_ACTION_REUSE) fresh = sum(1 for n in nodes if n.get("needs_new_plate")) return { "space_partition_applied": applied, "space_partition_fallback_reason": fallback_reason, "reuse_plate_count": reuse, "reuse_count": reuse, "plate_group_count": len({n.get("plate_group_id") for n in nodes}), "dag_node_count": fresh, "plate_shareability_counts": dict( _Counter(n.get("plate_shareability") for n in nodes) ), "zone_map_applied": applied, "zone_anchor_diagnostics": list(anchor_diagnostics or []), } def apply_dwelling_zone_map_plan( nodes: List[Dict[str, Any]], zone_plan: Optional[Dict[str, Any]], *, clean_anchor_candidate_bg_ids: FrozenSet[str] = frozenset(), ) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]: """Apply the dwelling zone map: EXACTLY ONE plate per zone (Phase 2). Per zone, ``select_zone_anchor`` picks the single plate anchor. The anchor renders a fresh plate (``render_new_plate`` / ``needs_new_plate=True``); every same-zone non-anchor bg ALIASES it (``reuse_existing_plate``, ``reuse_target_bg_id`` = anchor) and carries NO ``ref_tree_parents`` — the alias is a reference, not a derive (the user's hard rule: one physical space = one plate; per-shot angle/character is Phase 3 i2i off the one plate). Each node snapshots its prior canonical fields under ``geometry_ _diagnostic`` (same visual-gate traceability as the partition path) and is stamped ``render_action_source = dwelling_zone_map``. Additionally each node gets the Phase 3 contract ``zone_id`` + ``zone_plate_bg_id`` (the one plate every bg in the zone resolves to). ``render_order_index`` is recomputed over the fresh anchor set only; aliases are not DAG nodes (``None``). fail-closed: an unusable / uncovered / single-zone / synthetic plan is a NO-OP — every field is left as the upstream pipeline produced it and stamped ``render_action_source = geometry_route`` + ``partition_fallback_reason``. (In production this path is selected only when the plan is usable; the NO-OP is a defensive guard.) Returns ``(nodes, summary)``. """ node_bg_ids: FrozenSet[str] = frozenset( n.get("bg_id") for n in nodes if isinstance(n.get("bg_id"), str) and n.get("bg_id") ) usable, fallback_reason = _zone_plan_is_usable(zone_plan, node_bg_ids) if not usable: for n in nodes: n["render_action_source"] = RENDER_ACTION_SOURCE_GEOMETRY n["partition_fallback_reason"] = fallback_reason return nodes, _zone_summary( nodes, applied=False, fallback_reason=fallback_reason ) bg_assign: Dict[str, Any] = zone_plan["bg_zone_assignments"] # narrowed # group member nodes by their zone id (exact-string join). members_by_zone: Dict[str, List[Dict[str, Any]]] = {} for n in nodes: zid = bg_assign[n.get("bg_id")]["zone_id"] members_by_zone.setdefault(zid, []).append(n) # one anchor per zone (deterministic order over zone ids). anchor_diagnostics: List[str] = [] anchor_by_zone: Dict[str, str] = {} for zid in sorted(members_by_zone): anchor_by_zone[zid] = select_zone_anchor( members=members_by_zone[zid], zone_id=zid, clean_anchor_candidate_bg_ids=clean_anchor_candidate_bg_ids, bg_zone_assignments=bg_assign, diagnostics=anchor_diagnostics, ) for n in nodes: bg = n.get("bg_id") zid = bg_assign[bg]["zone_id"] anchor = anchor_by_zone[zid] for f in _PARTITION_CANONICAL_FIELDS: n[f"geometry_{f}_diagnostic"] = n.get(f) is_anchor = bg == anchor n["render_action"] = RENDER_ACTION_NEW if is_anchor else RENDER_ACTION_REUSE n["reuse_target_bg_id"] = "" if is_anchor else anchor n["needs_new_plate"] = is_anchor n["plate_group_id"] = zid n["plate_anchor_bg_id"] = anchor # ★ alias, NOT derive: no ref_tree_parents (the derive signal). The # non-anchor reuses the one zone plate; Phase 3 i2i adds angle/character. n["ref_tree_parents"] = [] n["ref_role_per_parent"] = {} n["max_refs"] = MAX_REFS_PER_BG n["plate_shareability"] = ( "shareable_partition" if len(members_by_zone[zid]) > 1 else "exclusive" ) n["render_action_source"] = RENDER_ACTION_SOURCE_ZONE_MAP n["partition_fallback_reason"] = "" # Phase 3 contract: every bg resolves to its zone's single plate. n["zone_id"] = zid n["zone_plate_bg_id"] = anchor # render_order_index over the fresh anchor set only; aliases get None. fresh = [n for n in nodes if n.get("needs_new_plate")] def _idx(n: Dict[str, Any]) -> int: ni = n.get("node_index") return ni if isinstance(ni, int) and not isinstance(ni, bool) else 0 for n in nodes: if not n.get("needs_new_plate"): n["render_order_index"] = None for order_i, n in enumerate(sorted(fresh, key=_idx)): n["render_order_index"] = order_i return nodes, _zone_summary( nodes, applied=True, fallback_reason="", anchor_diagnostics=anchor_diagnostics, ) # ───────────────────────────────────────────────────────────────────── # Plan builder # ───────────────────────────────────────────────────────────────────── def _empty_plan( *, fp_id: str, readback_status: str, blockers: List[str] ) -> Dict[str, Any]: return { "fp_id": fp_id, "shot_aware_bg_render_plan_status": "not_applicable" if not blockers else "failed", "graph": {"nodes": []}, "validators": { "all_validators_passed": False, "diagnostics": list(blockers), }, "real_api_call_counts": {"image": 0, "llm": 0, "vlm": 0}, "readback_status": readback_status, "production_clear": False, "diagnostics": [], } # W20F11 (2026-07-23 Codex 합의 — 슬라이스 E fp_open_sea 실측): 같은 # camera_unit 에서 camera/look_at 후보가 둘 다 ≥1 인 viable unit 이 하나도 # 없는 fp 는 어떤 LLM 출력도 camera 멤버십 검증을 통과할 수 없는 구조적 # 계획 불가 — provider 0콜로 typed not_applicable 격리. reason code 는 # 이 camera geometry 불가 사유만 좁게 소유(다른 실패 합류 금지) — # background_render 가 이 코드만 명시적으로 direct-plate 경로로 보낸다. CAMERA_LOOKAT_NOT_APPLICABLE_REASON = "no_viable_camera_look_at_pair" def assess_camera_lookat_viability( geometry: Dict[str, Any], ) -> Dict[str, Any]: """viable unit 존재 여부 — validator 와 동일한 same-unit pair SOT. validator 는 한 노드의 camera_unit 하나로 camera/look_at 두 후보 배열을 함께 조회하므로, 전역 합계가 아니라 **같은 u** 에서 양쪽이 비어 있지 않은 unit 이 하나는 있어야 계획 가능하다 (Codex: camera= {1:[..]}, look_at={2:[..]} 처럼 전역 nonzero 여도 공통 unit 0 이면 불가). malformed(비 dict/비 list/비 정수쌍 cell)=ShotAwareBgRender PlanError fail-closed — 상류 계약 위반을 not_applicable 로 세탁 금지. """ # NARROW-1 (Codex W20F11 리뷰): helper viability == 'validator+schema # 에서 실제 선택 가능' 이 SOT — validator 는 emitted camera_unit(int) # 을 str() 로 바꿔 **원본 dict** 에서 조회하므로 int key/비정규 # key("01")는 도달 불가, 응답 schema 는 cell 좌표 minimum=0 이라 음수 # 후보도 선택 불가. 이런 후보를 viable 로 오판하면 손상 CP 에서 # 불필요한 LLM 호출→영구 failed. key coercion 없이 원 key 검증. if not isinstance(geometry, dict): raise ShotAwareBgRenderPlanError( f"geometry malformed: expected dict, got " f"{type(geometry).__name__} (fail-closed)" ) counts: Dict[str, Dict[str, int]] = {} for name in ( "camera_cell_candidates_per_unit", "look_at_cell_candidates_per_unit", ): mapping = geometry.get(name) if not isinstance(mapping, dict): raise ShotAwareBgRenderPlanError( f"geometry.{name} malformed: expected dict, got " f"{type(mapping).__name__} (fail-closed — upstream " "contract failure, not not_applicable)" ) per_unit: Dict[str, int] = {} for unit, cells in mapping.items(): # total 가드 (Codex NARROW): lstrip/isdigit 사전 검사는 # "--1"(전 하이픈 제거 후 통과)·초장문 숫자(int digit limit) # 에서 raw ValueError 를 누출 — try/except 로 전 입력 커버. canonical = False if isinstance(unit, str): try: parsed_unit = int(unit) except (TypeError, ValueError): pass else: canonical = str(parsed_unit) == unit if not canonical: raise ShotAwareBgRenderPlanError( f"geometry.{name} unit key {unit!r} malformed: " "expected canonical integer string (validator " "str(camera_unit) 조회로 도달 불가 — fail-closed)" ) if not isinstance(cells, list): raise ShotAwareBgRenderPlanError( f"geometry.{name}[{unit!r}] malformed: expected " f"list, got {type(cells).__name__} (fail-closed)" ) for cell in cells: if not ( isinstance(cell, (list, tuple)) and len(cell) == 2 and all( isinstance(x, int) and not isinstance(x, bool) and x >= 0 for x in cell ) ): raise ShotAwareBgRenderPlanError( f"geometry.{name}[{unit!r}] malformed cell " f"{cell!r}: expected [int>=0, int>=0] — 응답 " "schema minimum=0 이라 선택 불가 (fail-closed)" ) per_unit[unit] = len(cells) counts[name] = per_unit cam_counts = counts["camera_cell_candidates_per_unit"] look_counts = counts["look_at_cell_candidates_per_unit"] viable_units = sorted( u for u, n in cam_counts.items() if n > 0 and look_counts.get(u, 0) > 0 ) return { "viable": bool(viable_units), "viable_units": viable_units, "camera_candidate_counts": cam_counts, "look_at_candidate_counts": look_counts, } def build_render_plan_for_fp( *, fp_id: str, planner_input: Dict[str, Any], llm_provider: Optional[ Callable[..., Dict[str, Any]] ] = None, projection_card_index: Optional[ Dict[str, Dict[str, Dict[str, Any]]] ] = None, card_content_index: Optional[ Dict[str, Dict[str, Dict[str, Any]]] ] = None, space_partition_plan: Optional[Dict[str, Any]] = None, zone_plan: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """Build the per-fp shot-aware bg render plan. ``llm_provider`` is a dependency-injected callable. When ``None`` (default in production) the planner returns a structurally-empty plan with ``status='not_applicable'`` and a diagnostic — i.e. no LLM call is made. When provided, it is invoked as:: llm_provider( fp_id=..., dossier=..., geometry=..., shot_readiness=..., candidate_catalog=..., ) -> {"graph": {"nodes": [...]}} The output is validated against ``validate_llm_output``; on failure the plan is marked ``failed`` with the validator diagnostics surfaced. ``projection_card_index`` (metadata-only ``{bg:{shot:{card_state, card_id, fallback_reason}}}``) feeds the C3 enrich pass (⑤). ``card_content_index`` (the envelope-bearing ``{bg:{shot:{card_state, card_id, card}}}``) feeds the card-aware router (④) for D5 horizontal-band withhold. Both default ``None`` (projection subsystem OFF) — the router is then a baseline no-op and enrich stamps ``not_available``. The 3a/3b mirrors (⑥/⑦) run regardless. """ dossier = planner_input["dossier"] geometry = planner_input["geometry"] readback_status = planner_input.get("readback_status", "unknown") shot_readiness = planner_input["shot_readiness"] same_fp_bg_ids: FrozenSet[str] = frozenset( (dossier.get("per_bg_render_facts_by_bg_id") or {}).keys() ) # W20E6-B: renderable subset = BGs with at least one staged shot. # Falls back to the assemble step's pre-computed list when present # (canonical path), otherwise derives from per_bg ok flags. renderable_list = shot_readiness.get("renderable_bg_ids") if isinstance(renderable_list, (list, tuple)): renderable_bg_ids: FrozenSet[str] = frozenset( bgid for bgid in renderable_list if isinstance(bgid, str) ) else: renderable_bg_ids = frozenset( bgid for bgid, entry in ( (shot_readiness.get("per_bg") or {}).items() ) if entry.get("ok") ) blockers: List[str] = [] if not shot_readiness.get("ok") or not renderable_bg_ids: # W20E6-B nano: an FP with zero renderable BGs is # not_applicable, not failed. The LLM planner has nothing to # plan for, so we skip the provider call entirely and surface # the per-bg readiness diagnostics. Step aggregate must NOT # count this as failed (no upstream contract was violated -- # the staged-shot subset just does not intersect this FP). blockers.append( "shot_readiness gate skipped (no BG has at least one " "staged consuming shot in shot_staging cp; not_applicable)" ) blockers.extend(shot_readiness.get("blockers") or []) out = _empty_plan( fp_id=fp_id, readback_status=readback_status, blockers=blockers, ) out["shot_aware_bg_render_plan_status"] = "not_applicable" return out # W20F11: same-unit camera/look_at viability preflight — 불가 fp 는 # provider 0콜 typed not_applicable (malformed 는 위 helper 가 raise # → step 이 per-fp failed 격리). viable=기존 경로 byte-identical. viability = assess_camera_lookat_viability(geometry) if not viability["viable"]: out = _empty_plan( fp_id=fp_id, readback_status=readback_status, blockers=[ "no viable camera/look_at candidate pair in any unit — " "camera geometry cannot satisfy the plan contract for " "any node (not_applicable; provider not called)" ], ) out["shot_aware_bg_render_plan_status"] = "not_applicable" out["not_applicable_reason_code"] = ( CAMERA_LOOKAT_NOT_APPLICABLE_REASON ) out["camera_lookat_viability"] = { "camera_candidate_counts": viability[ "camera_candidate_counts"], "look_at_candidate_counts": viability[ "look_at_candidate_counts"], } return out if llm_provider is None: out = _empty_plan( fp_id=fp_id, readback_status=readback_status, blockers=[ "no llm_provider wired — W20B production default; plan " "left empty by design. Wire a provider via the step " "wrapper for mock/dry runs or for a future real smoke " "behind explicit approval." ], ) out["shot_aware_bg_render_plan_status"] = "not_applicable" return out try: llm_output = llm_provider( fp_id=fp_id, dossier=dossier, geometry=geometry, shot_readiness=shot_readiness, candidate_catalog=planner_input.get("candidate_catalog") or [], ) except Exception as exc: # W20F10 (Codex 조건 5): retry 경로 실패는 실제 completion 2회 — # provider 예외의 completion_call_count 로 실비용을 감사에 보존 _fail_llm_calls = int(getattr(exc, "completion_call_count", 1)) return { "fp_id": fp_id, "shot_aware_bg_render_plan_status": "failed", "graph": {"nodes": []}, "validators": { "all_validators_passed": False, "diagnostics": [ f"llm_provider raised: {type(exc).__name__}: {exc}" ], }, "real_api_call_counts": { "image": 0, "llm": _fail_llm_calls, "vlm": 0, }, "readback_status": readback_status, "production_clear": False, "diagnostics": [], } if not isinstance(llm_output, dict): raise ShotAwareBgRenderPlanError( f"llm_provider returned non-dict ({type(llm_output).__name__})" ) nodes = (llm_output.get("graph") or {}).get("nodes") or [] # W20E7-C: derive the clean anchor candidate surface from the dossier # (always a frozenset, possibly empty -- empty fails closed inside the # validator). The dossier surface is the SOT; the LLM cannot pick an # anchor outside this set even if the prompt would otherwise allow it. clean_anchor_candidate_bg_ids: FrozenSet[str] = frozenset( bid for bid in ( (dossier.get("anchor_selection_metadata") or {}).get( "candidate_bg_ids" ) or [] ) if isinstance(bid, str) and bid ) validators = validate_llm_output( fp_id=fp_id, nodes=nodes, same_fp_bg_ids=same_fp_bg_ids, geometry=geometry, readback_status=readback_status, renderable_bg_ids=renderable_bg_ids, clean_anchor_candidate_bg_ids=clean_anchor_candidate_bg_ids, ) # ④ Commit 2a + W21B-w4 3a — card-aware router. route_render_actions_v2 # stamps canonical render_action / reuse_target_bg_id (the SOT) and, when a # projection card content index is available, WITHHOLDS reuses that carry a # D5 horizontal-band conflict (reuse→render_new_plate). With no card index # this is byte-identical to the baseline route_render_actions (brief §10). nodes, routing = route_render_actions_v2( nodes=nodes, dossier=dossier, card_content_index=card_content_index, per_bg_readiness=(shot_readiness.get("per_bg") or {}), ) validators["render_action_routing_ok"] = routing[ "render_action_routing_ok" ] validators["reuse_plate_count"] = routing["reuse_plate_count"] validators["card_withhold_count"] = routing.get("card_withhold_count", 0) # ⑤ W21B-w4 C3 — stamp per-node projection-card fields AFTER routing (so # reuse children can inherit their resolved target's projection). When # the projection subsystem is OFF (``projection_card_index=None``, # default) the fields are stamped ``not_available`` — additive only, # existing plan semantics unchanged. nodes, projection_summary = enrich_nodes_with_projection_cards( nodes=nodes, card_index=projection_card_index, per_bg_readiness=(shot_readiness.get("per_bg") or {}), ) validators["projection_card_state_counts"] = projection_summary[ "projection_card_state_counts" ] # ⑥ 3a plate partition + ⑦ 3b reference DAG — additive deterministic # mirrors of the canonical render_action SOT. (A) decision: these run # UNCONDITIONALLY (card OFF → geometry-only partition + fresh-plate DAG), # so the persisted SCHEMA-9 node shape always carries the partition/DAG # fields and downstream never branches on field presence. Mirror runs # after C3 enrich so plate_anchor_shot_id can read the stamped anchor. nodes, partition_summary = mirror_plate_partition(nodes) nodes, dag_summary = build_reference_dag(nodes) validators["plate_group_count"] = partition_summary["plate_group_count"] validators["plate_shareability_counts"] = partition_summary[ "plate_shareability_counts" ] validators["dag_node_count"] = dag_summary["dag_node_count"] # ⑧ W21B-w5 STEP4 (A wiring) — space_partition_plan canonical normalizer. # Runs LAST so it is the FINAL authority: a usable plan OVERRIDES the whole # canonical render surface from the LLM partition SOT and demotes the wave4 # geometry route/mirror/DAG to geometry_*_diagnostic. Default (None) → no-op, # legacy geometry surface untouched + partition_fallback_reason stamped. The # post-partition summary is recomputed from the FINAL nodes (Codex cond.3), # so the canonical counts never reflect the demoted geometry basis. # W21B Phase 2 selection: a usable dwelling zone plan (one-plate-per-zone) # takes precedence over the partition normalizer. When ``zone_plan`` is # absent (default None — every existing caller) or unusable, the legacy # partition path runs UNCHANGED (byte-identical). Both stamp the same # summary keys so the validators block below reads either uniformly. zone_node_bg_ids: FrozenSet[str] = frozenset( n.get("bg_id") for n in nodes if isinstance(n.get("bg_id"), str) and n.get("bg_id") ) if zone_plan is not None and _zone_plan_is_usable( zone_plan, zone_node_bg_ids )[0]: nodes, sp_summary = apply_dwelling_zone_map_plan( nodes, zone_plan, clean_anchor_candidate_bg_ids=clean_anchor_candidate_bg_ids, ) else: nodes, sp_summary = apply_space_partition_plan( nodes, space_partition_plan ) validators["space_partition_applied"] = sp_summary["space_partition_applied"] validators["space_partition_fallback_reason"] = sp_summary[ "space_partition_fallback_reason" ] if sp_summary["space_partition_applied"]: # partition is the SOT — recompute the canonical counts and accept its # internally-consistent routing (members reuse a fresh anchor by # construction, so there is no reuse-of-reuse to fail on). validators["reuse_plate_count"] = sp_summary["reuse_plate_count"] validators["plate_group_count"] = sp_summary["plate_group_count"] validators["plate_shareability_counts"] = sp_summary[ "plate_shareability_counts" ] validators["dag_node_count"] = sp_summary["dag_node_count"] validators["render_action_routing_ok"] = True routing = {**routing, "render_action_routing_ok": True} # W21B Phase 2 (B2): surface the zone-map decision + its no-clean-candidate # diagnostics to the FINAL validators (and the human-facing diagnostics # blob) so the visual gate sees them — they must not vanish after the ⑧ # summary copy. Present only when the zone path ran (the partition summary # has no zone_map_applied key); the partition path is unaffected. zone_diags: List[str] = [] if "zone_map_applied" in sp_summary: validators["zone_map_applied"] = sp_summary["zone_map_applied"] zone_diags = list(sp_summary.get("zone_anchor_diagnostics") or []) validators["zone_anchor_diagnostics"] = zone_diags validators["diagnostics"] = ( list(validators.get("diagnostics") or []) + routing["diagnostics"] + zone_diags ) validators["all_validators_passed"] = bool( validators["all_validators_passed"] and routing["render_action_routing_ok"] ) status = ( "ok" if validators["all_validators_passed"] else "failed" ) # W20F9 — provider 가 retry sentinel 을 박았으면 per_fp diagnostics 로 # surface (manifest 에 카운터 노출, silent repair 차단). retry_meta = llm_output.pop("_w20f9_retry_metadata", None) diagnostics_block: List[Dict[str, Any]] = [] llm_call_count = 1 if isinstance(retry_meta, dict): diagnostics_block.append({"w20f9_camera_validator_retry": retry_meta}) llm_call_count = 2 # first pass + retry # W20F10 — graph/anchor 교정 retry sentinel (W20F9 와 상호 배타, # 카메라 메타 키 재사용 금지 — Codex 조건 5) ga_retry_meta = llm_output.pop("_graph_anchor_retry_metadata", None) if isinstance(ga_retry_meta, dict): diagnostics_block.append( {"w20f10_graph_anchor_validator_retry": ga_retry_meta}) llm_call_count = 2 # first pass + retry return { "fp_id": fp_id, "shot_aware_bg_render_plan_status": status, "graph": {"nodes": list(nodes)}, "validators": validators, "real_api_call_counts": {"image": 0, "llm": llm_call_count, "vlm": 0}, "readback_status": readback_status, # production_clear is gated on both validators AND a non-synthetic # readback. Synthetic readback never promotes. "production_clear": bool( validators["all_validators_passed"] and validators["synthetic_readback_production_clear"] ), "diagnostics": diagnostics_block, }