"""W20A2: FloorPlanGeometryReadbackStep. opt-in (default OFF) producer for the geometry-readback bridge between W20A (base_location_dossier) and W20B (shot-aware LLM planner). Per fp_id present in the dossier checkpoint, the step: - Builds a synthetic_fixture readback (W20A2 default — no real VLM). - Computes geometry candidates (camera cells, look-at cells, direction vectors, view cone records, visible units/openings candidates, wall/door diagnostics). - Renders a review HTML overlay and writes it under the checkpoint directory. Gates (all must be true): - ``settings.background_mode`` ∈ {"on", "floor_plan_anchored"}. - ``settings.floor_plan_geometry_readback_enabled`` = True. - ``settings.base_location_dossier_enabled`` = True. - ``settings.floor_plan_prompt_version`` = "6". Anything else → ``not_applicable`` with byte-stable empty payload. LLM / image / VLM API call 0, DB / ImageAsset write 0. The HTML file is the only sidecar artifact; manifest.json carries summaries + the file's relative path under the checkpoint dir. 재개 재사용 (2026-08-19 사용자 지시) — 실제 VLM 제공자를 켜면 이 단계는 도면 하나마다 모델을 부르고 **답이 매번 조금씩 다르다**. 그런데 도면 몇 개가 늘 실패해서 스텝이 `partial` 로 남고, `partial` 은 재개마다 다시 불린다. 다시 불릴 때 성공한 도면까지 전부 다시 읽으니 기하 값이 바뀌고, 그 도면을 쓰는 장면이 「재료가 낡았다」로 판정돼 그림이 다시 만들어졌다 (08-19 저녁 한 바퀴에 66장). 그래서 **직전 체크포인트에서 성공한 도면은 그대로 쓰고 실패한 것만 다시 읽는다.** 재사용 조건은 셋 다 만족일 때만 이다 — ①이 스텝의 config_hash 가 같다 ②그 도면의 입력(dossier)이 1비트도 안 바뀌었다 ③산출 HTML 파일이 실재한다. `mode="force"` 는 재사용하지 않는다(force 의 뜻이 "처음부터 다시"이므로). """ from __future__ import annotations import hashlib import json import logging from pathlib import Path from typing import Any, Callable, Dict, List, Optional from app.core.step_runner import StepRunner from app.modules.pipeline.floor_plan_geometry_readback import ( GeometryReadbackError, compute_geometry_candidates, compute_readback, ) from app.modules.pipeline.floor_plan_review_html import ( render_review_html, ) logger = logging.getLogger(__name__) SCHEMA_VERSION = 2 # W20F7-A (Codex 2026-05-28): VLM provider max_completion_tokens 2000→16000 stale 자동 인식용 bump. PROMPT_VERSION = "1" # Directory name under the checkpoint root where review HTML files land. # Generic — no scenario tokens. _HTML_SUBDIR_NAME = "review_html" class FloorPlanGeometryReadbackStep(StepRunner): # Test-only injection slot for a mock VLM provider. Default None — # production behaviour never injects a provider directly; the # ``_resolve_vlm_provider`` method consults the settings selector. _vlm_provider_override: Optional[Callable[..., Dict[str, Any]]] = None def set_vlm_provider_for_testing( self, provider: Optional[Callable[..., Dict[str, Any]]] ) -> None: """Mocking helper. Production callers must NOT touch this.""" self._vlm_provider_override = provider def _resolve_vlm_provider(self) -> Optional[Callable[..., Dict[str, Any]]]: """Resolve the VLM provider for this run. Resolution order (only one path active per run): 1. Explicit ``set_vlm_provider_for_testing`` injection — used by mock-provider tests. Bypasses settings. 2. ``settings.floor_plan_vlm_readback_real_provider_enabled`` True → resolve to the production-adjacent ``litellm_vlm_provider`` (W20A2.5 helper). The helper itself currently raises NotImplementedError because the actual litellm request body is intentionally deferred to a future approved wave. Flipping the selector without that follow-up wave therefore fails closed at the call site, not silently in production. 3. Default → ``None``; the core ``compute_readback`` emits the synthetic_fixture path. Zero external API calls. """ from app.core.config import settings if self._vlm_provider_override is not None: return self._vlm_provider_override if not bool( getattr( settings, "floor_plan_vlm_readback_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.floor_plan_vlm_provider import ( dual_vlm_provider, gemini_vlm_provider, ) # ★#92 6단계 — 두 모델로 읽는 판(2026-08-27). 기본 OFF 다: # 켜면 도면마다 두 번 사고(실측 도면당 약 $0.09) 하류가 쓰는 # 값은 그대로이며 `readback["agreement"]` 등급 한 칸이 는다. if bool(getattr( settings, "floor_plan_vlm_readback_dual_enabled", False)): return dual_vlm_provider # ★#92 (2026-08-27): 단일 경로도 **gemini** 다. Sol 로 읽던 # `litellm_vlm_provider` 는 OpenAI 전용 preflight 를 갖고 있어 # 기본값만 바꾸면 막힌다 — Router 를 타는 어댑터로 간다. return gemini_vlm_provider 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_vlm_readback_real_provider_enabled": bool( getattr( settings, "floor_plan_vlm_readback_real_provider_enabled", False, ) ), # ★#92 6단계 — 켜면 **다른 provider** 가 돌고 산출에 # `agreement` 칸이 는다. 안 접으면 완주 CP 가 옛 산출을 # 그대로 재사용해 두 번째 모델을 부른 적이 없는 것이 된다. # ★모델 이름까지 접는다 — 별칭 뒤 물리 모델만 바뀌어도 # 다른 판정이다(era R1 BLOCK-2 와 같은 계약). "floor_plan_vlm_readback_dual_enabled": bool( getattr( settings, "floor_plan_vlm_readback_dual_enabled", False ) ), "floor_plan_vlm_readback_dual_models": ( _dual_models_stamp() if getattr( settings, "floor_plan_vlm_readback_dual_enabled", False ) else "" ), "floor_plan_geometry_readback_enabled": bool( settings.floor_plan_geometry_readback_enabled ), # TASK3-A: retry count alters readback behaviour for a real # provider → invalidate the checkpoint when it changes. "floor_plan_geometry_readback_max_retries": int( getattr( settings, "floor_plan_geometry_readback_max_retries", 0, ) or 0 ), "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( "floor_plan_geometry_readback: %s parse failed: %s", step_id, exc, ) return None def _checkpoint_dir(self) -> Path: from app.core.config import settings return ( Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / "floor_plan_geometry_readback" ) def _write_html(self, *, fp_id: str, html_str: str) -> str: """Write the review HTML file and return its checkpoint-relative path.""" target_dir = self._checkpoint_dir() / _HTML_SUBDIR_NAME target_dir.mkdir(parents=True, exist_ok=True) target = target_dir / f"{fp_id}.html" target.write_text(html_str, encoding="utf-8") # Return path relative to the project's checkpoint episode root — # so downstream consumers can reconstruct the URL via the # external HTTP server pattern. return f"floor_plan_geometry_readback/{_HTML_SUBDIR_NAME}/{fp_id}.html" @staticmethod def _dossier_input_hash(dossier: Dict[str, Any]) -> str: """이 도면 하나의 입력 지문 — 재사용이 낡은 값을 붙잡지 않게 한다. config_hash 는 설정만 본다. 도면 자체(마커 목록·이미지 경로·격자) 가 바뀌었는데 설정이 같으면 config_hash 로는 못 가른다 — 그래서 입력을 따로 접는다. ★도면 **그림 내용**도 함께 접는다. dossier 는 그림의 경로만 갖고 내용 지문이 없어서(base_location_dossier.py:647), 도면이 **같은 경로에서 다시 그려지면** dossier 만 봐서는 알 수 없다. 이 단계가 실제로 보는 것은 그 그림이므로, 그림이 바뀌면 다시 읽어야 한다. 파일 한 번 읽기는 모델 호출에 비하면 없는 값이다. 그림이 없거나 못 읽으면 **그 사실을** 접는다 — 읽히던 때와 안 읽히는 때가 갈려야 읽히게 된 순간에 다시 읽는다. """ payload: Dict[str, Any] = {"dossier": dossier} raw_path = dossier.get("fp_image_path") if raw_path: try: payload["fp_image_sha256"] = hashlib.sha256( Path(str(raw_path)).read_bytes()).hexdigest() except OSError as exc: payload["fp_image_unreadable"] = f"{type(exc).__name__}" else: payload["fp_image_missing"] = True return hashlib.sha256( json.dumps(payload, sort_keys=True, ensure_ascii=False) .encode("utf-8") ).hexdigest()[:16] def _reusable_entries(self, mode: str) -> Dict[str, Dict[str, Any]]: """직전 체크포인트에서 **성공한** 도면 항목만 골라 돌려준다. force 는 처음부터 다시라는 뜻이라 재사용하지 않는다. config_hash 가 다르면 계약이 바뀐 것이므로 전부 다시 읽는다. """ if mode == "force": return {} prior = self._load_prev_checkpoint("floor_plan_geometry_readback") if not isinstance(prior, dict): return {} if prior.get("config_hash") != self._config_hash(): return {} per_fp = (prior.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): continue if entry.get("error") or not entry.get("readback"): continue # 실패분은 다시 읽는다 if not entry.get("input_hash"): continue # 지문 없는 옛 기록은 재사용 대상 아님 out[fp_id] = entry return out 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.floor_plan_geometry_readback_enabled): return self._not_applicable() if not bool(settings.base_location_dossier_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() dossier_cp = self._load_prev_checkpoint("base_location_dossier") dossiers = ( (dossier_cp or {}).get("data", {}).get("dossiers") or {} ) if not dossiers: return self._not_applicable() per_fp: Dict[str, Dict[str, Any]] = {} completed = 0 failed = 0 vlm_call_total = 0 # Resolve once per run. Either every fp uses the synthetic # fixture path (provider=None) or every fp uses the resolved # provider; mixing per-fp would make the call-count audit # ambiguous. base_provider = self._resolve_vlm_provider() # TASK3-A: per-fp bounded retry. Only meaningful with a real # provider — the synthetic-fixture path is deterministic, so a # retry would fail identically; pinning it to 0 there keeps the # default-OFF payload byte-for-byte identical to legacy. max_retries = int( getattr(settings, "floor_plan_geometry_readback_max_retries", 0) or 0 ) effective_retries = max_retries if base_provider is not None else 0 reusable = self._reusable_entries(mode) reused_fp_ids: List[str] = [] for fp_id, dossier in dossiers.items(): # ── 재개 재사용 (2026-08-19) ────────────────────────────── # 성공한 도면을 다시 읽으면 답이 달라지고, 그 도면을 쓰는 # 장면이 통째로 다시 그려진다. 입력이 그대로면 그대로 쓴다. input_hash = self._dossier_input_hash(dossier) prior_entry = reusable.get(fp_id) if prior_entry is not None \ and prior_entry.get("input_hash") == input_hash \ and (self._checkpoint_dir() / _HTML_SUBDIR_NAME / f"{fp_id}.html").is_file(): # 항목을 1비트도 바꾸지 않고 그대로 옮긴다 — 재사용 표식을 # 항목 안에 넣으면 하류가 읽는 값이 바뀐다. per_fp[fp_id] = prior_entry # 2026-08-20: 위 조건은 HTML 파일의 **존재**만 본다. 중간에 # 죽은 회차가 다른 값으로 그려 둔 화면이 남아 있으면 사람이 # 체크포인트와 어긋난 화면을 본다(기계 소비는 항목만 쓴다). # 항목에서 다시 그려 덮는다 — 순수 함수라 모델 호출 0. try: self._write_html( fp_id=fp_id, html_str=render_review_html( dossier=dossier, geometry=prior_entry.get("geometry") or {})) except Exception as exc: # noqa: BLE001 # 화면 실패가 재사용을 죽이면 도면을 전부 다시 읽는다 — # 그쪽이 실제 지출이다. logger.warning( "readback 재사용 fp_id=%s: 검토 화면 재작성 실패 %s", fp_id, exc) reused_fp_ids.append(fp_id) completed += 1 continue counted_provider: Optional[Callable[..., Dict[str, Any]]] = None # Per-iteration counter — mutated by ``counted`` closure # only when a provider is resolved. Cleanly re-initialised # per fp so the aggregate ``vlm_call_total`` audit stays # exact. _counter = [0] if base_provider is not None: def counted(**kw): # noqa: E306 — defined per-iter # ★**wrapper 진입이 곧 호출 수는 아니다** (2026-08-27 # Codex BLOCK-2). 두 모델 읽기는 이 한 번 안에서 # provider 를 두 번 부른다 — 진입만 세면 기록이 # 절반이 되고 「도면당 두 번 산다」와 반대가 된다. # # ★그래서 산출이 `agreement.attempts` 로 **실제 시도 # 수를 스스로 밝히면** 그 길이를 센다. 없으면 1 — # 단일 경로는 그대로다(byte-identical). # ★**진입 즉시 1 을 센다** — 예외가 나도 그 왕복은 # 이미 산 것이다. 이 저장소가 여러 번 세운 계약이고 # (`test_retry_recovers_transient_readback_failure`), # 성공 뒤에 세도록 옮겼다가 그 시험 셋을 깼다. # ★BLOCK 을 고치다 **원래 계약을 되살렸다** — # 지적의 범위를 넘기지 않는다. _counter[0] += 1 out = base_provider(**kw) # 두 모델 읽기는 이 한 번 안에서 두 번 부른다. # **두 번째부터**를 더한다(첫 번은 위에서 셌다). if isinstance(out, dict): rows = (out.get("agreement") or {}).get("attempts") if isinstance(rows, list) and len(rows) > 1: _counter[0] += len(rows) - 1 return out counted_provider = counted readback_attempts = 0 try: # TASK3-A: bounded retry around the (sole) VLM call. A # transient provider failure (e.g. a stochastic duplicate # marker NUMBER tripping the validator) is re-attempted up # to ``effective_retries`` extra times. The broad catch # mirrors the real failure surface — ``compute_readback`` # re-raises GeometryReadbackError for shape errors but lets # a provider-raised ``VlmProviderError`` (and any transient) # propagate verbatim. Each attempt is a fresh counted call, # so ``_counter`` / ``real_vlm_call_count`` stay honest. readback = None last_exc: Optional[Exception] = None for _attempt in range(effective_retries + 1): readback_attempts += 1 try: readback = compute_readback( dossier=dossier, fp_image_path=dossier.get("fp_image_path"), vlm_provider=counted_provider, ) last_exc = None break except Exception as exc: last_exc = exc readback = None if readback is None: # Retries exhausted — re-raise so the except branches # below build the failed entry with the original error # semantics (GeometryReadbackError vs unexpected). raise ( last_exc if last_exc is not None else GeometryReadbackError( "readback failed with no captured exception" ) ) geometry = compute_geometry_candidates( dossier=dossier, readback=readback ) html_str = render_review_html( dossier=dossier, geometry=geometry ) html_rel = self._write_html(fp_id=fp_id, html_str=html_str) per_fp_vlm = ( _counter[0] if counted_provider is not None else 0 ) vlm_call_total += per_fp_vlm entry: Dict[str, Any] = { "fp_id": fp_id, "readback_status": readback["status"], "readback": readback, "geometry": geometry, "review_html_relative_path": html_rel, "real_vlm_call": bool(per_fp_vlm), "real_vlm_call_count": per_fp_vlm, # 재개 재사용의 전제 — 이 값을 만든 입력이 무엇이었나. "input_hash": input_hash, } # Additive only when a retry actually happened — keeps the # single-attempt / synthetic payload byte-identical. if readback_attempts > 1: entry["readback_attempts"] = readback_attempts per_fp[fp_id] = entry completed += 1 except GeometryReadbackError as exc: logger.error( "floor_plan_geometry_readback fp_id=%s: %s", fp_id, exc, ) per_fp_vlm = ( _counter[0] if counted_provider is not None else 0 ) vlm_call_total += per_fp_vlm entry = { "fp_id": fp_id, "error": str(exc)[:300], "real_vlm_call": bool(per_fp_vlm), "real_vlm_call_count": per_fp_vlm, } if readback_attempts > 1: entry["readback_attempts"] = readback_attempts per_fp[fp_id] = entry failed += 1 except Exception as exc: # pragma: no cover — defensive logger.exception( "floor_plan_geometry_readback fp_id=%s unexpected: %s", fp_id, exc, ) per_fp_vlm = ( _counter[0] if counted_provider is not None else 0 ) vlm_call_total += per_fp_vlm entry = { "fp_id": fp_id, "error": f"unexpected: {type(exc).__name__}: {exc}"[:300], "real_vlm_call": bool(per_fp_vlm), "real_vlm_call_count": per_fp_vlm, } if readback_attempts > 1: entry["readback_attempts"] = readback_attempts per_fp[fp_id] = entry failed += 1 # Either at least one fp_id was processed OK or every one # failed. Empty per_fp means dossier was empty (already returned # not_applicable above). if reused_fp_ids: logger.info( "floor_plan_geometry_readback: 직전 결과 재사용 %d개 / " "다시 읽음 %d개 — 성공한 도면을 다시 읽으면 값이 달라져 " "그 도면을 쓰는 장면이 통째로 다시 그려진다", len(reused_fp_ids), len(dossiers) - len(reused_fp_ids)) return { "applicable_count": 1 if per_fp else 0, "completed_count": completed, "failed_count": failed, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": { "per_fp": per_fp, # 이번 실행에서 다시 읽지 않고 그대로 쓴 도면. 아래 # real_vlm_call_count 는 **이번에 실제로 부른 횟수**라 # 재사용분은 안 들어간다(항목 안의 값은 그것을 만든 # 실행의 것 — 둘을 더해 읽으면 안 된다). **({"reused_fp_ids": sorted(reused_fp_ids)} if reused_fp_ids else {}), # Aggregate counter. With the default-OFF selector + # no test injection this is always 0. When a mock # provider is injected (or the real-provider selector # is flipped) it reflects the actual call count. "real_vlm_call_count": vlm_call_total, "image_api_call_count": 0, "llm_call_count": 0, }, } def _dual_models_stamp() -> str: """두 모델 읽기의 **물리 모델 쌍** — 지문에 접을 한 줄. ★별칭(`gemini-pro`·`grok`)만 접으면 설정의 실제 모델이 바뀌어도 지문이 안 움직여 옛 판정이 재사용된다. era R1 BLOCK-2 가 세운 계약과 같다. """ from app.modules.pipeline.floor_plan_vlm_provider import ( DUAL_READBACK_ALIASES, _physical_of, ) return ",".join(f"{a}={_physical_of(a)}" for a in DUAL_READBACK_ALIASES)