"""W21B-w5 STEP5-B: FloorPlanLightSidecarStep (NB2 sparse text-to-image). opt-in (default OFF) producer of a per-fp **simplified floor-plan sidecar PNG** — a SPARSE CV-readable PARTITION DIAGRAM (thick enclosing walls, an unnumbered geometry skeleton, only 8-10 essential numbered markers) that the background renderer reads as an I2I anchor far better than the dense, colored, text-labelled detailed floor plan. Winning design (L05 canary, gallery 8815 — user visual gate; supersedes the v0 ref-edit and the first sparse pass): 1. **gpt-5.5 frequency-aware selection** — from the floor_plan_prompt ``numbered_elements`` plus a per-element ``camera_use_count`` (aggregated from ``camera_recommendations`` = how many shot cameras frame each element), the model picks 8-10 ESSENTIAL numbered markers (a high-frequency element such as a repeatedly-framed TV/curtain is protected from being dropped) + an UNNUMBERED geometry skeleton (rooms/walls/doors/windows), and writes a SHORT ``room_schematic_prompt`` with explicit marker-placement constraints (every circle inside the building outline, door/window markers on their wall, one circle per number). A long prompt renders badly; few markers render cleanly. 2. **gpt-image-2** PURE text-to-image — NO reference image (a reference leaks/duplicates marker numbers; the v0 ref-edit produced "11" twice). ``best_of_n`` draws are rendered; the primary sidecar is the FIRST successful draw, and every draw is persisted for the visual gate. (Render model = gpt-image-2 per the 2026-06-01 user visual gate: on the same prompt it honours "one circle per number" far better than Nano Banana 2, which duplicated markers on repeated office furniture; the residual STRUCTURE issues are an upstream floor_plan_prompt skeleton problem shared by both models.) The DETAILED ``floor_plan_render`` PNG is left untouched — it stays the ``shot_projection_card`` / edge-judge substrate, so the W21B-w5 partition SOT is never disturbed (STEP5-B boundary lock). This step only ADDS a sidecar; the consumer (``background_render._apply_light_fp_sidecar``) swaps the FP anchor PNG to the light sidecar when it rendered, and otherwise falls back to the detailed FP (default behaviour, byte-identical when OFF). Pipeline order **21.56** — after ``floor_plan_prompt`` (21.5x, source of ``numbered_elements`` + ``camera_recommendations``) and ``floor_plan_render``, and before ``floor_plan_overlay_payload`` (21.57). Per fp_id (status ``ok`` in the floor_plan_prompt checkpoint): 1. Aggregate ``camera_use_count`` from ``camera_recommendations``. 2. Build the freq-aware selection bundle → one gpt-5.5 call → validate the STRUCTURAL skeleton (rooms drawn, no state overlay essential, prompt present). 3. Render ``best_of_n`` NB2 text-to-image draws; persist them; primary = first ok. 4. Persist provenance (essential/skeleton numbers, diagnostics, draws). 5. A selection/render miss / missing data / unsafe id falls back to the detailed FP for that fp ONLY — never a step failure (optional sidecar). Validation is STRUCTURAL + render-status only (Codex lock): geometry fidelity (no duplicate / floating markers) and best-of-N quality are a VISUAL gate + optional later OCR/VLM diagnostic, NOT a deterministic hard gate (avoids an expensive false-negative-prone gate that would fall back to the cluttered detailed FP). Gates (all must be true, else ``not_applicable`` with byte-stable empty data): - ``settings.background_mode`` ∈ {"on", "floor_plan_anchored"}. - ``settings.floor_plan_light_sidecar_enabled`` = True. - ``settings.floor_plan_prompt_version`` ∈ {"6", "7"}. """ from __future__ import annotations import hashlib import json import logging import re from pathlib import Path from typing import Any, Callable, Dict, Optional from app.core.step_runner import StepRunner from app.modules.pipeline.floor_plan_light_prompt import ( FREQ_SELECT_PROMPT_VERSION, build_freq_aware_selection_bundle, compute_camera_use_counts, freq_select_prompt_hash, validate_freq_aware_selection, ) logger = logging.getLogger(__name__) SCHEMA_VERSION = 2 PROMPT_VERSION = "2" VALIDATION_MODE = "selection_plus_render_visual_gate" # floor_plan_prompt selector values this step supports — same v6-compatible # output schema (numbered_elements). SOT in app.core.fp_prompt_compat. from app.core.fp_prompt_compat import ( # noqa: E402 V6_COMPATIBLE_FP_PROMPT_VERSIONS as _SUPPORTED_FP_PROMPT_VERSIONS, ) # fp_id whitelist: lowercase ascii + digits + underscore. path traversal 방어 # (floor_plan_render_step._SAFE_FP_RE 와 동일 정책). _SAFE_FP_RE = re.compile(r"^[a-z0-9][a-z0-9_]*$") class FloorPlanLightSidecarStep(StepRunner): # Test-only injection slots. Default None — production resolves real clients. # _select_override(*, bundle) -> Optional[dict] (raw LLM JSON, or None) # _render_override(*, prompt, out_path, fp_id) -> dict # with keys {status, png_path, attempts, error}. _select_override: Optional[Callable[..., Optional[Dict[str, Any]]]] = None _render_override: Optional[Callable[..., Dict[str, Any]]] = None # Lazily-resolved NB2 image client (only when an fp needs render). _client: Any = None def set_select_for_testing( self, select: Optional[Callable[..., Optional[Dict[str, Any]]]] ) -> None: """Mocking helper for the gpt-5.5 selection. Production must NOT touch.""" self._select_override = select def set_render_for_testing( self, render: Optional[Callable[..., Dict[str, Any]]] ) -> None: """Mocking helper for the image render. Production must NOT touch.""" self._render_override = render # ──────────────────────────── selection (gpt-5.5) ──────────────────────────── def _run_selection( self, *, numbered_elements: Any, camera_use_counts: Dict[int, int], fp_id: str, ) -> Optional[Dict[str, Any]]: """One frequency-aware selection call → raw LLM JSON (or None on miss). The pure module owns the bundle/schema/validator; this only issues the IO. A provider fail-close returns None (per-fp detailed fallback, never a step failure).""" bundle = build_freq_aware_selection_bundle( numbered_elements, camera_use_counts ) if self._select_override is not None: return self._select_override(bundle=bundle) from app.core.config import settings from app.modules.pipeline.floor_plan_light_prompt_provider import ( FloorPlanLightProviderError, litellm_light_fp_provider, ) try: return litellm_light_fp_provider( prompt_bundle=bundle, model=getattr( settings, "floor_plan_light_sidecar_select_model", "openai/gpt-6-astra", ), ) except FloorPlanLightProviderError as exc: logger.warning( "floor_plan_light_sidecar: selection failed for %s: %s", fp_id, exc, ) return None def _resolve_openai_client(self): """OpenAI image client (mirrors floor_plan_render_step._resolve_openai_client). api_key 는 `openai_keys` 브로커가 정한다(2슬롯 failover, 2026-07-30). bare `OpenAI()` 는 os.environ 만 읽어 .env 키·슬롯 전환을 놓친다. timeout 은 settings.llm_timeout_image_gen. """ from app.core.openai_keys import openai_client from app.core.config import settings return openai_client( timeout=float(settings.llm_timeout_image_gen), ) # ──────────────────────────── render (gpt-image-2 text-to-image) ──────────────────────────── def _render_light_png( self, *, prompt: str, out_path: Path, fp_id: str ) -> Dict[str, Any]: """PURE text-to-image draw via gpt-image-2 (NO reference image). The render model is gpt-image-2 (user visual gate, 2026-06-01): on the same room_schematic_prompt it honours "exactly one circle per number" far better than NB2 (which duplicated markers on repeated office furniture), while the residual STRUCTURE issues (over-segmented / interior-convention exteriors) are upstream floor_plan_prompt skeleton problems shared by both models, not a render-model fault. ``ref_paths=[]`` routes ``render_one_floor_plan`` to ``images.generate`` (text-to-image — a reference leaks/duplicates marker numbers). The client is resolved lazily so a not_applicable run touches no SDK; tests bypass via ``_render_override``. """ if self._render_override is not None: return self._render_override( prompt=prompt, out_path=out_path, fp_id=fp_id ) from app.core.config import settings from app.modules.pipeline.floor_plan_render import render_one_floor_plan if self._client is None: self._client = self._resolve_openai_client() res = render_one_floor_plan( openai_client=self._client, image_model=getattr( settings, "floor_plan_light_sidecar_render_model", "gpt-image-2.5-sunburst"), prompt=prompt, out_path=out_path, ref_paths=[], # text-to-image (no reference image) fp_id=fp_id, ) return { "status": res.status, "png_path": res.png_path, "attempts": res.attempts, "error": res.error, } def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict[str, Any]]: """Read a prior step's ``manifest.json`` from the checkpoint dir.""" 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( "floor_plan_light_sidecar: %s parse failed: %s", step_id, exc ) return None 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, "floor_plan_light_sidecar_enabled": bool( getattr(settings, "floor_plan_light_sidecar_enabled", False) ), # NB2 sparse path levers: selection prompt/schema + models + N. "select_model": getattr( settings, "floor_plan_light_sidecar_select_model", ""), "render_model": getattr( settings, "floor_plan_light_sidecar_render_model", ""), "best_of_n": int( getattr(settings, "floor_plan_light_sidecar_best_of_n", 2)), "freq_select_prompt_version": FREQ_SELECT_PROMPT_VERSION, "freq_select_prompt_hash": freq_select_prompt_hash(), "schema_version": SCHEMA_VERSION, "prompt_version": PROMPT_VERSION, } return hashlib.sha256( json.dumps(payload, sort_keys=True).encode("utf-8") ).hexdigest()[:16] 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( getattr(settings, "floor_plan_light_sidecar_enabled", False) ): return self._not_applicable() if settings.floor_plan_prompt_version not in _SUPPORTED_FP_PROMPT_VERSIONS: return self._not_applicable() # source = floor_plan_prompt numbered_elements + camera_recommendations. prompt_cp = self._load_prev_checkpoint("floor_plan_prompt") floor_plans = ( ((prompt_cp or {}).get("data", {}) or {}).get("floor_plans", {}) or {} ) if not floor_plans: return self._not_applicable() best_of_n = max( 1, int(getattr(settings, "floor_plan_light_sidecar_best_of_n", 2))) # sidecar PNGs live under this step's checkpoint dir (not images/) — they # are an internal I2I reference, not a catalogued ImageAsset. out_dir = ( Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / "floor_plan_light_sidecar" / "light_renders" ) out_dir_resolved = out_dir.resolve() from app.core.file_paths import to_relative_image_path per_fp: Dict[str, Any] = {} # OPTIONAL sidecar: a selection / render miss is a per-fp DETAILED fallback # by design, NOT a step failure. failed_count stays 0 (else StepRunner # finalizes partial/failed). Granular outcomes live in data. rendered = 0 fallback = 0 render_failed = 0 # counts selection ATTEMPTS (one per fp that reached selection), not # confirmed network calls — a provider preflight fail returns None without # issuing a call (Codex W21B-w5 review: honest naming for cost reporting). selection_attempts = 0 image_call_total = 0 for fp_id in sorted(floor_plans): entry = floor_plans.get(fp_id) or {} if not isinstance(entry, dict) or entry.get("status") != "ok": continue base: Dict[str, Any] = { "validation_mode": VALIDATION_MODE, "prompt_version": PROMPT_VERSION, } def _fb(reason: str, **extra: Any) -> None: per_fp[fp_id] = { **base, "status": "fallback", "fallback_reason": reason, "light_png_relative_path": None, **extra, } # unsafe id → never render (path traversal guard). if not _SAFE_FP_RE.match(fp_id): logger.error( "floor_plan_light_sidecar: unsafe fp_id %r — skip", fp_id ) _fb("unsafe_fp_id") fallback += 1 continue numbered = entry.get("numbered_elements") or [] if not isinstance(numbered, list) or not numbered: _fb("no_numbered_elements") fallback += 1 continue cameras = entry.get("camera_recommendations") or [] # 1) frequency-aware selection (one gpt-5.5 call). counts = compute_camera_use_counts(numbered, cameras) raw = self._run_selection( numbered_elements=numbered, camera_use_counts=counts, fp_id=fp_id ) selection_attempts += 1 if raw is None: _fb("selection_failed") fallback += 1 continue val = validate_freq_aware_selection( raw=raw, numbered_elements=numbered, camera_use_counts=counts ) if not val["ok"]: _fb("selection_invalid", diagnostics=val["diagnostics"][:8]) fallback += 1 continue prompt = val["room_schematic_prompt"] # 2) best-of-N NB2 text-to-image draws (primary = first ok). out_dir.mkdir(parents=True, exist_ok=True) attempts_rel: list = [] primary_rel: Optional[str] = None last_error = "" path_escape = False for i in range(1, best_of_n + 1): out_path = out_dir / f"{fp_id}_{i}.png" try: out_path.resolve().relative_to(out_dir_resolved) except (ValueError, OSError): path_escape = True break render_res = self._render_light_png( prompt=prompt, out_path=out_path, fp_id=fp_id ) image_call_total += int(render_res.get("attempts", 1) or 1) png_out = render_res.get("png_path") or "" if ( render_res.get("status") == "ok" and png_out and Path(png_out).exists() ): rel = to_relative_image_path(png_out) attempts_rel.append(rel) if primary_rel is None: primary_rel = rel else: last_error = (render_res.get("error") or "")[:200] if path_escape: logger.error( "floor_plan_light_sidecar: out_path escapes out_dir for %s", fp_id, ) _fb("rejected_path") fallback += 1 continue if primary_rel is None: # every draw missed → detailed fallback for this fp only (NOT a # step failure — see the counter note above). _fb("render_failed", render_error=last_error) fallback += 1 render_failed += 1 continue per_fp[fp_id] = { **base, "status": "rendered", "light_png_relative_path": primary_rel, "best_of_n_paths": attempts_rel, "essential_numbers": val["essential_numbers"], "skeleton_numbers": val["skeleton_numbers"], "unjustified_high_freq": val["unjustified_high_freq"], "selection_diagnostics": val["diagnostics"][:8], } rendered += 1 return { "applicable_count": 1 if per_fp else 0, # whole-step completion (mirrors floor_plan_render): the optional # sidecar "completed" its work whenever it processed any fp, even if # every fp fell back. failed_count is held at 0 by design so a # selection / render miss never degrades the step to partial/failed. "completed_count": 1 if per_fp else 0, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": { "per_fp": per_fp, "rendered_count": rendered, "fallback_count": fallback, "render_failed_count": render_failed, "best_of_n": best_of_n, "selection_attempt_count": selection_attempts, "image_api_call_count": image_call_total, }, }