"""W20B: ShotAwareBgRenderPlanStep. opt-in (default OFF) producer for the dwelling-scoped reference graph + per-bg camera/reference decisions. Consumes ``base_location_dossier`` (W20A), ``floor_plan_geometry_readback`` (W20A2), ``floor_plan_overlay_payload`` (W19B-1), ``background_master_plan``, and ``shot_staging``. Gates (all must be true → run; any false → not_applicable): - ``settings.background_mode`` ∈ {"on", "floor_plan_anchored"} - ``settings.shot_aware_bg_render_plan_enabled`` = True - ``settings.base_location_dossier_enabled`` = True - ``settings.floor_plan_geometry_readback_enabled`` = True - ``settings.floor_plan_prompt_version`` = "6" The step does NOT call any external LLM by default. ``build_render_plan_for_fp(llm_provider=None)`` returns a structurally- empty plan with diagnostics, so an enable-only flag flip in production still costs 0 external API calls. A real LLM smoke is gated on a future wave behind explicit user/Codex approval. To run a mock/dry smoke, inject an llm_provider via ``set_llm_provider_for_testing``. No image / VLM / DB / ImageAsset write. Output is checkpoint-only. """ from __future__ import annotations import hashlib import json import logging from pathlib import Path from typing import Any, Callable, Dict, Optional from app.core.step_runner import StepRunner from app.modules.pipeline.shot_aware_bg_render_plan import ( ShotAwareBgRenderPlanError, assemble_planner_inputs, build_render_plan_for_fp, ) logger = logging.getLogger(__name__) SCHEMA_VERSION = 10 # W21B-w5 STEP4 A wiring (2026-05-31): ⑧ apply_space_partition_plan canonical normalizer runs LAST, consuming the bg_space_partition (21.594) space_partition_plan. When a usable plan is present it OVERRIDES the whole canonical render surface (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) from the LLM partition SOT and demotes the wave4 geometry route/mirror/DAG to geometry_*_diagnostic; default (no plan) → no-op + render_action_source=geometry_route + partition_fallback_reason. Every node now additively carries render_action_source + partition_fallback_reason (+ geometry_*_diagnostic when applied) → persisted node shape changed → 9→10 bump. (prev 9 = W21B-w4 3a/3b card-aware routing + plate-partition / reference-DAG mirrors.) W21B Phase 2 (2026-06-08): the dwelling_zone_map path adds zone_id / zone_plate_bg_id / render_action_source=dwelling_zone_map to nodes, but ONLY when dwelling_zone_map_enabled AND a usable multi-zone plan applies (flag-gated, conditional — NOT every node like the partition fields). flag OFF → byte-identical v10 shape, so SCHEMA stays 10 (a bump would assert a "v11 = zone fields present" invariant the conditional path does not honour). The dwelling_zone_map_enabled selector is in config_hash, so toggling it invalidates this checkpoint. # W21B-w5: the bg_space_partition (21.594) checkpoint SCHEMA the ⑧ normalizer # consumes. A mismatch fails closed (partition cp ignored → legacy geometry # surface). Must track bg_space_partition_step.SCHEMA_VERSION. BG_SPACE_PARTITION_SCHEMA = 1 # W20F10 (2026-07-23 Codex 리뷰 BLOCKING-1): 로컬 사본 상수가 provider # 팩 승격(v3)과 비동기돼 config_hash 가 drift 하지 않던 결함 — provider # 모듈의 PROMPT_VERSION 을 단일 SOT 로 소비한다(교차 핀 테스트 별도). from app.modules.pipeline.shot_aware_bg_render_plan_llm_provider import ( PROMPT_VERSION, ) class ShotAwareBgRenderPlanStep(StepRunner): # Test-only injection slot. Default None — production behaviour # makes zero LLM calls. _llm_provider_override: Optional[Callable[..., Dict[str, Any]]] = None def set_llm_provider_for_testing( self, provider: Optional[Callable[..., Dict[str, Any]]] ) -> None: """Mocking helper. Production callers must NOT touch this.""" self._llm_provider_override = provider def _resolve_llm_provider(self) -> Optional[Callable[..., Dict[str, Any]]]: """Resolve the LLM planner provider for this run. Resolution order (only one path active per run): 1. Explicit ``set_llm_provider_for_testing`` injection — used by mock-provider tests. Bypasses settings. 2. ``settings.shot_aware_bg_render_plan_real_provider_enabled`` True → lazy-import + return the production-adjacent ``litellm_shot_aware_bg_render_plan_provider`` (W20B helper). The helper itself is fail-closed at every preflight joint; flipping this selector without ``OPENAI_API_KEY`` set or without the prompt pack present therefore raises inside the helper rather than silently issuing real API traffic. 3. Default → ``None``; ``build_render_plan_for_fp`` emits a structurally-empty plan with diagnostic. Zero external API calls. """ from app.core.config import settings if self._llm_provider_override is not None: return self._llm_provider_override if not bool( getattr( settings, "shot_aware_bg_render_plan_real_provider_enabled", False, ) ): return None # Lazy import keeps the provider module (which carries the # lazy ``import litellm``) out of this step wrapper's import # graph until the selector is actively flipped. from app.modules.pipeline.shot_aware_bg_render_plan_llm_provider import ( litellm_shot_aware_bg_render_plan_provider, ) return litellm_shot_aware_bg_render_plan_provider @staticmethod def _build_projection_card_index( card_cp: Optional[Dict[str, Any]], ) -> Optional[Dict[str, Dict[str, Dict[str, Any]]]]: """Assemble ``{bg_id: {shot_id: {card_state, card_id, fallback}}}``. Reads the ``shot_projection_card`` checkpoint's per-card entries (keyed ``{bg_id}::{shot_id}``, C2 step output). ``card_id`` is the envelope's content-addressed id; precondition-blocked / error entries carry no envelope, so the id is left empty and the fallback reason is taken from the recorded reason / validator_state. Returns ``None`` when the checkpoint is absent (projection subsystem unavailable — the plan then stamps ``not_available``). """ if not card_cp: return None cards = (card_cp.get("data") or {}).get("cards") or {} index: Dict[str, Dict[str, Dict[str, Any]]] = {} for entry in cards.values(): if not isinstance(entry, dict): continue bg_id = entry.get("bg_id") shot_id = entry.get("shot_id") if not bg_id or not shot_id: continue state = entry.get("card_state") card_id = ((entry.get("card") or {}).get("card_id")) or "" fallback = ( entry.get("fallback_reason") or entry.get("validator_state") or "" ) index.setdefault(str(bg_id), {})[str(shot_id)] = { "card_state": state, "card_id": str(card_id), "fallback_reason": str(fallback), } return index @staticmethod def _build_projection_card_content_index( card_cp: Optional[Dict[str, Any]], ) -> Optional[Dict[str, Dict[str, Dict[str, Any]]]]: """Assemble ``{bg_id: {shot_id: {card_state, card_id, card}}}``. Like ``_build_projection_card_index`` but carries the full card ``envelope`` (``vlm_output.visible_items`` etc.) so the card-aware router (``route_render_actions_v2`` → ``_self_card_base_bands`` → ``extract_base_bands``) can read each anchor card's base-marker bands for D5 horizontal-band withhold. Kept separate from the metadata-only index (which carries ``fallback_reason`` for C3 enrich) so the enrich path never loads envelopes it does not need. Returns ``None`` when the checkpoint is absent (projection subsystem unavailable → router no-op). """ if not card_cp: return None cards = (card_cp.get("data") or {}).get("cards") or {} index: Dict[str, Dict[str, Dict[str, Any]]] = {} for entry in cards.values(): if not isinstance(entry, dict): continue bg_id = entry.get("bg_id") shot_id = entry.get("shot_id") if not bg_id or not shot_id: continue envelope = entry.get("card") or {} card_id = envelope.get("card_id") or "" index.setdefault(str(bg_id), {})[str(shot_id)] = { "card_state": entry.get("card_state"), "card_id": str(card_id), "card": envelope, } return index @staticmethod def _build_space_partition_lookup( sp_cp: Optional[Dict[str, Any]], ) -> Dict[str, Dict[str, Any]]: """Assemble ``{fp_id: space_partition_plan}`` from the bg_space_partition (21.594) checkpoint — the LLM partition SOT the ⑧ normalizer consumes. fail-closed (Codex condition 4): a fp is consumed ONLY when its per-fp entry has ``status == 'ok'`` and a dict ``space_partition_plan``. A missing checkpoint, a schema_version mismatch (BG_SPACE_PARTITION_SCHEMA), or a blocked / errored fp yields no entry → ``build_render_plan_for_fp`` gets ``space_partition_plan=None`` for that fp and the normalizer no-ops to the legacy geometry surface. NEVER returns a half-built plan. """ if not sp_cp: return {} if sp_cp.get("schema_version") != BG_SPACE_PARTITION_SCHEMA: logger.warning( "shot_aware_bg_render_plan: bg_space_partition schema " "%s != expected %s — ignoring partition cp (fail-closed)", sp_cp.get("schema_version"), BG_SPACE_PARTITION_SCHEMA, ) return {} per_fp = (sp_cp.get("data") or {}).get("per_fp") or {} out: Dict[str, Dict[str, Any]] = {} for fp_id, entry in per_fp.items(): if not isinstance(entry, dict) or entry.get("status") != "ok": continue plan = entry.get("space_partition_plan") if isinstance(plan, dict) and plan.get("render_actions"): out[str(fp_id)] = plan return out @staticmethod def _build_zone_map_lookup( zm_cp: Optional[Dict[str, Any]], ) -> Dict[str, Dict[str, Any]]: """Assemble ``{fp_id: zone_map}`` from the dwelling_zone_map (21.5945) checkpoint — the one-plate-per-zone SOT the ⑧ selection consumes (W21B Phase 2). A dwelling is consumed ONLY when its per-fp zone_map is a NON-synthetic dict (the synthetic fixture asserts nothing and must never drive the render surface). A missing checkpoint / synthetic / malformed entry yields no entry → ``build_render_plan_for_fp`` gets ``zone_plan=None`` for that fp and the legacy partition path runs (byte-identical). Final usability (multi-zone, full node coverage) is judged inside ``build_render_plan_for_fp`` over the actual graph nodes. """ if not zm_cp: return {} per_dwelling = (zm_cp.get("data") or {}).get("per_dwelling") or {} out: Dict[str, Dict[str, Any]] = {} for fp_id, zm in per_dwelling.items(): if isinstance(zm, dict) and not zm.get("synthetic"): out[str(fp_id)] = zm return out def _config_hash(self) -> str: from app.core.config import settings payload = { "background_mode": settings.background_mode, "floor_plan_prompt_version": settings.floor_plan_prompt_version, "base_location_dossier_enabled": bool( settings.base_location_dossier_enabled ), "floor_plan_geometry_readback_enabled": bool( settings.floor_plan_geometry_readback_enabled ), "shot_aware_bg_render_plan_enabled": bool( settings.shot_aware_bg_render_plan_enabled ), "shot_aware_bg_render_plan_real_provider_enabled": bool( getattr( settings, "shot_aware_bg_render_plan_real_provider_enabled", False, ) ), "bg_space_partition_enabled": bool( getattr(settings, "bg_space_partition_enabled", False) ), # W21B Phase 2 — flipping the zone-map selector changes the ⑧ # canonical surface (one-plate-per-zone vs partition), so it must # invalidate this step's checkpoint. "dwelling_zone_map_enabled": bool( getattr(settings, "dwelling_zone_map_enabled", False) ), "schema_version": SCHEMA_VERSION, "prompt_version": PROMPT_VERSION, } return hashlib.sha256( json.dumps(payload, sort_keys=True).encode("utf-8") ).hexdigest()[:16] def _load_prev_checkpoint( self, step_id: str ) -> Optional[Dict[str, Any]]: from app.core.config import settings cp = ( Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / step_id / "manifest.json" ) if cp.exists(): try: return json.loads(cp.read_text(encoding="utf-8")) except Exception as exc: logger.warning( "shot_aware_bg_render_plan: %s parse failed: %s", step_id, exc, ) return None def _not_applicable(self) -> Dict[str, Any]: return { "applicable_count": 0, "completed_count": 0, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {}, } def _execute(self, mode: str = "resume") -> Dict[str, Any]: from app.core.config import settings if settings.background_mode not in {"on", "floor_plan_anchored"}: return self._not_applicable() if not bool(settings.shot_aware_bg_render_plan_enabled): return self._not_applicable() if not bool(settings.base_location_dossier_enabled): return self._not_applicable() if not bool(settings.floor_plan_geometry_readback_enabled): return self._not_applicable() # v7/v8/v9 are schema-compatible with v6 (numbered_elements identical); # accept all. v5 still routes to not_applicable. SOT: fp_prompt_compat. from app.core.fp_prompt_compat import V6_COMPATIBLE_FP_PROMPT_VERSIONS if settings.floor_plan_prompt_version not in V6_COMPATIBLE_FP_PROMPT_VERSIONS: return self._not_applicable() # W19B-3/W20B mutual exclusion gate. background_render w18j_overlap # 와 shot_aware_bg_render_plan_enabled 가 동시에 켜져 있으면 # reference-graph SOT 가 두 path 에서 다르게 산출돼 silent # divergence — fail closed before reading deeper checkpoints or # invoking llm provider. from app.core.steps._selector_guards import detect_w19b3_w20b_conflict conflict_msg = detect_w19b3_w20b_conflict(settings) if conflict_msg is not None: return { "applicable_count": 1, "completed_count": 0, "failed_count": 1, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": { "selector_conflict": conflict_msg, "per_fp": {}, }, "error": conflict_msg, } dossier_cp = self._load_prev_checkpoint("base_location_dossier") geometry_cp = self._load_prev_checkpoint( "floor_plan_geometry_readback" ) overlay_cp = self._load_prev_checkpoint("floor_plan_overlay_payload") master_plan_cp = self._load_prev_checkpoint("background_master_plan") shot_staging_cp = self._load_prev_checkpoint("shot_staging") # W21B-w4 C3 — projection-card index (deterministic consume). When # the shot_projection_card checkpoint is absent (card step OFF, # default) the index is None and the plan stamps every node # ``projection_card_state=not_available`` (additive, plan otherwise # unchanged). NO LLM / VLM / image — the C2 gate already ran. projection_card_cp = self._load_prev_checkpoint("shot_projection_card") projection_card_index = self._build_projection_card_index( projection_card_cp ) # W21B-w4 3a — envelope-bearing index for the card-aware router (④). # Separate from the metadata-only index above (C3 enrich, ⑤): the # router needs vlm_output.visible_items base bands; enrich needs # fallback_reason. None when the card cp is absent → router no-op. projection_card_content_index = ( self._build_projection_card_content_index(projection_card_cp) ) # W21B-w5 STEP4 — the bg_space_partition (21.594) LLM SOT. {fp_id: plan} # for ok fps only; missing cp / schema mismatch / blocked fp → no entry → # the ⑧ normalizer no-ops to the legacy geometry surface (fail-closed). space_partition_cp = self._load_prev_checkpoint("bg_space_partition") space_partition_lookup = self._build_space_partition_lookup( space_partition_cp ) # W21B Phase 2 STEP — dwelling zone map (one-plate-per-zone). Gated # separately (dwelling_zone_map_enabled, default OFF). When OFF the # lookup is empty → zone_plan=None per fp → legacy partition path runs # (byte-identical). When ON, a usable per-fp zone_map supersedes the # partition normalizer inside build_render_plan_for_fp's ⑧ selection. zone_map_lookup: Dict[str, Dict[str, Any]] = {} if bool(getattr(settings, "dwelling_zone_map_enabled", False)): zone_map_cp = self._load_prev_checkpoint("dwelling_zone_map") zone_map_lookup = self._build_zone_map_lookup(zone_map_cp) dossier_data = (dossier_cp or {}).get("data", {}) or {} geometry_data = (geometry_cp or {}).get("data", {}) or {} overlay_data = (overlay_cp or {}).get("data", {}) or {} master_plan_data = (master_plan_cp or {}).get("data", {}) or {} shot_staging_data = (shot_staging_cp or {}).get("data", {}) or {} shot_staging_diag: Dict[str, Any] = {} try: planner_inputs = assemble_planner_inputs( dossier_cp_data=dossier_data, geometry_cp_data=geometry_data, overlay_cp_data=overlay_data, master_plan_cp_data=master_plan_data, shot_staging_cp_data=shot_staging_data, shot_staging_diagnostics=shot_staging_diag, ) except ShotAwareBgRenderPlanError as exc: return { "applicable_count": 1, "completed_count": 0, "failed_count": 1, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": { "error": str(exc)[:300], "per_fp": {}, "shot_staging_index_diagnostics": shot_staging_diag, }, } if not planner_inputs: na = self._not_applicable() na_data = dict(na.get("data") or {}) na_data["shot_staging_index_diagnostics"] = shot_staging_diag na["data"] = na_data return na # Resolve the LLM provider once per run. The selector / override # / default-None ordering is the single SOT — every per-fp call # passes through a counted closure so a future regression that # double-calls the provider for one fp surfaces in # ``real_api_call_counts.llm`` immediately. ``base_provider`` # may be None (default OFF) in which case build_render_plan_for_fp # returns the structurally-empty plan and never calls anything. base_provider = self._resolve_llm_provider() per_fp_out: Dict[str, Dict[str, Any]] = {} per_fp_provider_attempts: Dict[str, int] = {} completed = 0 failed = 0 llm_total = 0 for fp_id, bundle in planner_inputs.items(): if base_provider is None: counted_provider: Optional[ Callable[..., Dict[str, Any]] ] = None else: def _make_counted(_fp_id: str, _provider: Callable[..., Dict[str, Any]]): def _counted(**kwargs: Any) -> Dict[str, Any]: per_fp_provider_attempts[_fp_id] = ( per_fp_provider_attempts.get(_fp_id, 0) + 1 ) return _provider(**kwargs) return _counted counted_provider = _make_counted(fp_id, base_provider) try: plan = build_render_plan_for_fp( fp_id=fp_id, planner_input=bundle, llm_provider=counted_provider, projection_card_index=projection_card_index, card_content_index=projection_card_content_index, space_partition_plan=space_partition_lookup.get(fp_id), zone_plan=zone_map_lookup.get(fp_id), ) except ShotAwareBgRenderPlanError as exc: logger.error( "shot_aware_bg_render_plan fp_id=%s: %s", fp_id, exc, ) # Preserve provider attempt count: if the closure ran # (i.e. the provider was actually called) before the # downstream ShotAwareBgRenderPlanError, the real API # was attempted and the counter must reflect that. A # pre-call fail (e.g. shot_readiness gate) leaves # attempts at 0 verbatim. attempts = per_fp_provider_attempts.get(fp_id, 0) per_fp_out[fp_id] = { "fp_id": fp_id, "shot_aware_bg_render_plan_status": "failed", "error": str(exc)[:300], "real_api_call_counts": { "image": 0, "llm": attempts, "vlm": 0, }, } failed += 1 llm_total += attempts continue except Exception as exc: # pragma: no cover — defensive logger.exception( "shot_aware_bg_render_plan fp_id=%s unexpected: %s", fp_id, exc, ) attempts = per_fp_provider_attempts.get(fp_id, 0) per_fp_out[fp_id] = { "fp_id": fp_id, "shot_aware_bg_render_plan_status": "failed", "error": f"unexpected: {type(exc).__name__}: {exc}"[ :300 ], "real_api_call_counts": { "image": 0, "llm": attempts, "vlm": 0, }, } failed += 1 llm_total += attempts continue per_fp_out[fp_id] = plan status = plan.get("shot_aware_bg_render_plan_status") if status == "ok": completed += 1 elif status == "failed": failed += 1 llm_total += int( plan.get("real_api_call_counts", {}).get("llm", 0) ) return { "applicable_count": 1, "completed_count": completed, "failed_count": failed, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": { "per_fp": per_fp_out, "real_api_call_counts": { "image": 0, "llm": llm_total, "vlm": 0, }, # Mirror the per-fp provider attempt counts observed by # the step's counted closure. Always equal in shape to # the per-fp plan's ``real_api_call_counts.llm`` for a # healthy build_render_plan_for_fp run, but surfaced # separately so a future regression that double-calls # the provider for one fp shows up here distinctly from # the plan's hardcoded ``llm: 1`` bookkeeping. "llm_provider_attempts_per_fp": dict( per_fp_provider_attempts ), "shot_staging_index_diagnostics": shot_staging_diag, }, }