"""floor_plan_prompt — Phase 7 Step 3 / Phase 8 v2 / D6 v4 schema. 도면별 detailed t2i prompt 생성. plot-critical 시각 요소(커튼/시체 가림 등)를 LLM이 해당 도면 spec + scene segments에서 자연히 picking하도록. Phase 8 v2: schema에 ``numbered_elements`` + ``camera_recommendations`` 추가. - numbered_elements: 도면 안 가구/문/창/소품/영역 라벨에 부여한 번호 마커 매핑. - camera_recommendations: 각 background bg_id에 대한 도면 좌표계 기준 카메라 추천. D6 v4: BG_ID_RE (`^L\\d{2,3}B\\d{2,3}$`) 정합 + 3-layer 방어. - L1 static schema: ``bg_id`` pattern → BG_ID_RE. 옛 lowercase 패턴 (`display`, `bg_a`, hash garbage) 구조적 reject. - L2 runtime enum injection: ``run_floor_plan_prompt`` 가 ``expected_bg_ids`` 비 어있지 않으면 schema deep-copy + ``bg_id.enum=sorted(expected_bg_ids)`` 주입. LLM 출력을 정확한 set 으로 강제. - L3 validator exact-set: ``_validate_fp_prompt_extras`` 가 cam_bg == expected_bg_ids (extras + missing + duplicate) 검사. silent fallback 회로 차단. 추가: ``build_fp_user_prompt`` 가 D6 raw intent 의 ``state_label_raw`` 필드 사용 (legacy ``state_label`` fallback 유지). ``Valid bg_ids:`` block 으로 LLM 에게 정확한 ID list 명시. """ from __future__ import annotations import copy import logging import re import time from typing import Any, Callable, Dict, Iterable, List, Optional, Set from app.modules.prompt_loader import load_prompt, load_schema logger = logging.getLogger(__name__) _NON_ASCII_TEXT_RE = re.compile( r"[ㄱ-ㆎ가-힣" r"一-鿿㐀-䶿豈-﫿" r"぀-ヿ]" ) _MODULE = "floor_plan_prompt" # W19A: selector ↔ on-disk pack directory mapping. Default selector "5" # resolves to the unchanged v5 pack; opt-in selector "6" resolves to the # new v6 pack added under prompts/_base/floor_plan_prompt/6.202605262300/. # Adding a new selector requires landing the matching pack directory plus # updating consumers (validator branch, tests). PROMPT_VERSION_MAP: Dict[str, str] = { "5": "5.202605201406", "6": "6.202605262300", # W21B-wave-4: v7 is a v6-compatible pack — schema.json and # user_template.md are byte-identical to v6; only system.md adds the # Rule 11 self-fidelity wording. All v6-family validation/gates apply # to v7 (see ``_V6_COMPATIBLE_SELECTORS``). "7": "7.202605291624", # W21B-wave-5: v8 is a v6-compatible pack — schema.json is byte-identical # to v7; system.md adds space-type adaptation and user_template.md adds a # {space_type_block}. Output contract unchanged, so v6 validation applies. "8": "8.202606011200", # W21B-wave-5: v9 is a v6-compatible pack — schema.json and user_template.md # are byte-identical to v8; system.md adds the Spatial scale fidelity section # and folds scale-fidelity into Rule 11 (over-scale / over-rooming fix). The # output contract is unchanged, so v6 validation applies. v9 keeps v8's # space-type relay; the two are orthogonal (open-vs-enclosed form vs how many # rooms/zones to draw). "9": "9.202606011400", # W21B scale-guideline: v10 is a v6-based pack whose schema.json is an # additive superset of v6 — it adds the REQUIRED string field # ``scale_guideline`` (minimal vertical/size facts: storey count, # approximate height/footprint, opening counts, full stair extents). # A flat top-down plan carries no vertical scale, so downstream # volumetric conversions (perspective sketch / elevation / render) # collapsed every building to a single storey; v10 makes the FP LLM # author that guidance as evidence-bound metadata (scene segments / # creator corrections / visual_world_rules only; otherwise # era-region-typical range marked as typical). All existing v6-family # fields and gates are unchanged, so v6 validation still applies on # top of the v10-only scale_guideline safety check. "10": "10.202607081930", # v11 (2026-07-17 E2E7 실측): 단일 연속 위치 계약 — 다중 벽/분리 위치 # 요소를 마커 1개로 저작하면 렌더가 물리 위치마다 마커를 복제해 # exactly-once 계약이 깨진다 (fp_l14 마커 6·7/12 중복 4연속 실측). "11": "11.202607170145", } # Selectors that share the v6 output contract (numbered_elements partition # + camera_recommendations use/ignore semantics). v7 added the Rule 11 # self-fidelity wording, v8 space-type adaptation, v9 scale-fidelity to # system.md, but none changed the schema. Single SOT in app.core.fp_prompt_compat # so downstream consumer gates stay in lockstep (see that module's docstring). from app.core.fp_prompt_compat import ( # noqa: E402 SCALE_GUIDELINE_FP_PROMPT_VERSIONS as _SCALE_GUIDELINE_SELECTORS, V6_COMPATIBLE_FP_PROMPT_VERSIONS as _V6_COMPATIBLE_SELECTORS, ) # W21B-wave-5: code does NOT decide the fp space type. It only relays the # LLM-assigned per-background ``surface_role`` signals into the v8 prompt; # the FP LLM decides the diagram form from those roles + scope + scenes. No # priority / tie-break in code (that would make code the meaning-decider). def derive_fp_space_roles(applied_backgrounds: List[Dict[str, Any]]) -> List[str]: """이 fp 를 참조하는 background 들의 surface_role 을 그대로 모아 반환. dedup 없이 bg 순서대로 (count 신호 보존). 빈 리스트면 분류 정보 없음 — prompt 가 보수적으로 처리한다. 코드 의미판단 0. """ return [ b.get("surface_role", "") for b in (applied_backgrounds or []) if b.get("surface_role") ] def _build_space_type_block(space_roles: Optional[List[str]]) -> str: """user prompt 의 {space_type_block} 채움 (selector "8" 전용) — role 신호 relay만.""" roles = [r for r in (space_roles or []) if r] if not roles: return ( "No surface-role classification is available for this floor plan's " "backgrounds. Treat the space type as unknown: draw conservatively from the " "scope and scenes, and do NOT assume an enclosed multi-room dwelling." ) from collections import Counter listed = ", ".join(f"{role} x{n}" for role, n in sorted(Counter(roles).items())) return ( "An upstream model classified this floor plan's backgrounds with these surface " f"roles: {listed}. Decide the diagram form from these roles together with the " "scope and scenes (see the Space-type adaptation rule)." ) # v6 numbered_elements.items.base_layer_decision enum (W19 brief §2.2). # Used by the version-aware validator below; the on-disk v6 schema.json # enforces the same set on the LLM call surface. _V6_BASE_LAYER_DECISION_ENUM = frozenset({ "base_structural_unit", "base_opening", "base_persistent_fixture", "base_persistent_furniture", "state_overlay_plot_cue", "state_overlay_transient_object", }) def resolve_prompt_version(selector: str) -> str: """selector ("5"|"6") → on-disk pack directory name. Raises ValueError on unknown selector so a typo cannot silently route to the latest pack (the entire point of the W19A versioned-load surface). """ try: return PROMPT_VERSION_MAP[selector] except KeyError as exc: raise ValueError( f"unknown floor_plan_prompt selector {selector!r}; " f"allowed: {sorted(PROMPT_VERSION_MAP)}" ) from exc class FloorPlanPromptError(Exception): """retry 한도까지 실패.""" def build_fp_user_prompt( fp_spec: Dict[str, Any], applied_backgrounds: List[Dict[str, Any]], applied_shots: List[str], scene_segments: List[Dict[str, Any]], visual_world_rules: str, prompt_version: str = "5", space_roles: Optional[List[str]] = None, ) -> str: """도면 user prompt — scene text 무절단 inject. .replace() 사용 (.format() brace 충돌 회피, T4/T7과 동일 패턴). D6 v4: - ``Valid bg_ids:`` block 에 정확한 L##B## list inject (LLM 이 schema pattern 으로 추측하지 않게 명시적으로 보여줌). - bg detail line 은 ``state_label_raw`` (D6 raw intent 필드) 우선, 없으면 legacy ``state_label`` fallback. W19A: ``prompt_version`` selector ("5"|"6") 를 받아 versioned-load surface (prompt_loader) 로 pack 디렉토리를 pin. v5/v6 의 user_template 이 placeholder 동일하므로 동작 차이는 없지만, latest-pack auto-resolve silent shift 를 방지하기 위해 항상 pin 한다. """ template = load_prompt( _MODULE, "user_template", version=resolve_prompt_version(prompt_version) ) # D6 v4: Valid bg_ids block — LLM 이 verbatim copy 하도록 단일-라인 list. valid_bg_ids = [b.get("bg_id", "") for b in applied_backgrounds if b.get("bg_id")] valid_bg_ids_block = ( "\n".join(f"- {bg_id}" for bg_id in valid_bg_ids) if valid_bg_ids else "(none)" ) # D6 v4: detail block — bg_id, loc_id, state_label_raw, applies_to_shots. # state_label_raw (D6) 우선, 없으면 legacy state_label fallback. bg_lines: List[str] = [] for b in applied_backgrounds: bg_id = b.get("bg_id", "") or "(unassigned)" loc_id = b.get("loc_id", "") state = b.get("state_label_raw") or b.get("state_label") or "" shots = b.get("applies_to_shots") or [] shots_str = ", ".join(shots) if shots else "(none)" bg_lines.append( f"- bg_id: {bg_id}\n" f" loc_id: {loc_id}\n" f" state: {state}\n" f" applies_to_shots: {shots_str}" ) bg_block = "\n".join(bg_lines) or "(none)" shots_block = "\n".join(f"- {s}" for s in applied_shots) or "(none)" seg_lines: List[str] = [] for seg in scene_segments: si = seg.get("scene_index") heading = (seg.get("heading") or "").strip() text = (seg.get("text") or "").strip() seg_lines.append( f"### Scene {si} — {heading}" if heading else f"### Scene {si}" ) seg_lines.append(text) seg_lines.append("") seg_block = "\n".join(seg_lines).rstrip() or "(none)" # selector "8" only — older packs have no {space_type_block} token, so this # replace is a no-op for them (output stays byte-identical). space_type_block = ( _build_space_type_block(space_roles) if space_roles is not None else "" ) return ( template .replace("{space_type_block}", space_type_block) .replace("{fp_id}", fp_spec.get("fp_id", "")) .replace("{sub_location}", fp_spec.get("sub_location", "")) .replace("{scope}", fp_spec.get("scope", "")) .replace("{valid_bg_ids_block}", valid_bg_ids_block) .replace("{backgrounds_block}", bg_block) .replace("{shots_block}", shots_block) .replace("{scene_segments_block}", seg_block) .replace("{visual_world_rules}", visual_world_rules or "(none)") ) def _validate_fp_prompt_extras( *, fp_result: Dict[str, Any], expected_bg_ids: Set[str], prompt_version: str = "5", ) -> None: """v2/v4 신규 필드 검증 — D6 exact-set 계약. - ``numbered_elements[].number`` 가 unique integer ≥1 인지 확인. - ``camera_recommendations[].bg_id`` 가 ``expected_bg_ids`` 와 **정확히 일치** (extras + missing + duplicate 모두 fail-fast). D6 v4 변경 (이전 subset → exact-set): v3 까지는 ``cam_bg - expected_bg_ids`` (extras) 만 검사 → LLM 이 cam=[] 또는 partial set 반환해도 통과 (silent fallback). v4 는 system prompt rule 10 의 "for every bg_id ... emit one entry" 를 코드로 강제. empty cam_bg + non-empty expected → missing fail-fast. schema 기반 type/required validation은 ``call_structured`` 가 처리하므로 여기서는 cross-field invariant(번호 중복, bg_id exact set, duplicate)만 본다. W19A version-aware (selector "5"|"6"): ``prompt_version="5"`` (default) — v5 동작 verbatim 유지. 추가 검증 없음. ``prompt_version="6"`` — 모든 ``numbered_elements`` entry 가 ``base_layer_decision`` (W19 brief §2.2 의 6-value enum 중 하나) 을 carry 해야 한다. exact-set membership (regex / substring 없음). on-disk v6 schema.json 도 동일 enum 을 LLM 호출 단계에서 강제하지만, 본 validator 는 mocked / strict-off provider 안전망으로 다시 점검. W19A2 (selector "6") — per-bg marker applicability: v6 의 모든 ``camera_recommendations`` entry 는 ``use_numbered_elements`` / ``ignore_numbered_elements`` (integer 배열) 을 carry 해야 한다. - 두 배열 모두 list 이고 각 원소는 ``numbered_elements[].number`` 의 set 안에 있어야 한다 (exact integer join — natural language / substring 추론 금지). - ``use_numbered_elements ∩ ignore_numbered_elements`` 는 공집합. - 모든 ``state_overlay_*`` numbered_element 는 적어도 하나의 camera 의 ``use_numbered_elements`` 에 등장해야 한다. 어떤 bg 에서도 ``use`` 에 등장하지 않으면 fail (overlay leak 후보). - v5 path 는 이 검증을 적용하지 않음 (use/ignore 필드 자체가 v5 schema 에 없음 — verbatim 보존). """ nums_raw = fp_result.get("numbered_elements") or [] nums = [int(e["number"]) for e in nums_raw if "number" in e] if len(nums) != len(set(nums)): raise FloorPlanPromptError( f"duplicate numbered_elements.number: {nums}" ) if any(n < 1 for n in nums): raise FloorPlanPromptError( f"numbered_elements.number must be >=1: {nums}" ) if prompt_version in _V6_COMPATIBLE_SELECTORS: for idx, entry in enumerate(nums_raw): if not isinstance(entry, dict): raise FloorPlanPromptError( f"numbered_elements[{idx}] not a dict under v6" ) decision = entry.get("base_layer_decision") if decision is None: raise FloorPlanPromptError( f"numbered_elements[{idx}] missing base_layer_decision under v6" ) if decision not in _V6_BASE_LAYER_DECISION_ENUM: raise FloorPlanPromptError( f"numbered_elements[{idx}] invalid base_layer_decision " f"{decision!r} under v6 (allowed: " f"{sorted(_V6_BASE_LAYER_DECISION_ENUM)})" ) if prompt_version in _SCALE_GUIDELINE_SELECTORS: # v10 safety net — on-disk schema.json (required, minLength 30) 이 # 1차 강제하지만 mocked / strict-off provider 경로를 재점검한다. guideline = fp_result.get("scale_guideline") if not isinstance(guideline, str) or len(guideline.strip()) < 30: raise FloorPlanPromptError( "scale_guideline missing or too short under v10 " f"(need string >=30 chars, got {guideline!r:.80})" ) cams_raw = fp_result.get("camera_recommendations") or [] # D6 v4: empty bg_id 도 fail-fast (silent drop 차단). schema 가 1차 막지만 # mocked / strict-off provider 경로 안전망. for idx, c in enumerate(cams_raw): if not c.get("bg_id"): raise FloorPlanPromptError( f"camera_recommendations[{idx}] missing bg_id" ) cam_bg_list = [c["bg_id"] for c in cams_raw] # duplicate 검사 — 같은 bg_id 두 entries 면 fail. dups = sorted({bg for bg in cam_bg_list if cam_bg_list.count(bg) > 1}) if dups: raise FloorPlanPromptError( f"camera_recommendations has duplicate bg_id: {dups}" ) cam_bg = set(cam_bg_list) expected = set(expected_bg_ids) extra = cam_bg - expected if extra: raise FloorPlanPromptError( f"camera_recommendations references unknown bg_ids: {sorted(extra)}" ) missing = expected - cam_bg if missing: raise FloorPlanPromptError( f"camera_recommendations missing required bg_ids: {sorted(missing)}" ) if prompt_version in _V6_COMPATIBLE_SELECTORS: valid_numbers: Set[int] = { int(e["number"]) for e in nums_raw if "number" in e } state_overlay_numbers: Set[int] = { int(e["number"]) for e in nums_raw if isinstance(e, dict) and "number" in e and isinstance(e.get("base_layer_decision"), str) and e["base_layer_decision"].startswith("state_overlay_") } covered_state_overlay: Set[int] = set() for cam_idx, cam in enumerate(cams_raw): if not isinstance(cam, dict): raise FloorPlanPromptError( f"camera_recommendations[{cam_idx}] not a dict under v6" ) for field in ("use_numbered_elements", "ignore_numbered_elements"): arr = cam.get(field) if not isinstance(arr, list): raise FloorPlanPromptError( f"camera_recommendations[{cam_idx}].{field} missing or " f"not a list under v6" ) for item_idx, item in enumerate(arr): # bool 도 isinstance(int) 라 별도 차단 — JSON-side 정수만 허용. if isinstance(item, bool) or not isinstance(item, int): raise FloorPlanPromptError( f"camera_recommendations[{cam_idx}].{field}[{item_idx}] " f"is not an integer under v6 (got {item!r})" ) use_set = set(cam["use_numbered_elements"]) ignore_set = set(cam["ignore_numbered_elements"]) unknown_use = use_set - valid_numbers if unknown_use: raise FloorPlanPromptError( f"camera_recommendations[{cam_idx}].use_numbered_elements " f"references unknown numbers under v6: {sorted(unknown_use)}" ) unknown_ignore = ignore_set - valid_numbers if unknown_ignore: raise FloorPlanPromptError( f"camera_recommendations[{cam_idx}].ignore_numbered_elements " f"references unknown numbers under v6: {sorted(unknown_ignore)}" ) overlap = use_set & ignore_set if overlap: raise FloorPlanPromptError( f"camera_recommendations[{cam_idx}] use_numbered_elements ∩ " f"ignore_numbered_elements non-empty under v6: " f"{sorted(overlap)}" ) covered_state_overlay |= use_set & state_overlay_numbers missing_state_overlay = state_overlay_numbers - covered_state_overlay if missing_state_overlay: raise FloorPlanPromptError( f"state_overlay numbered_elements not covered by any " f"camera_recommendations.use_numbered_elements under v6 " f"(overlay leak candidate): {sorted(missing_state_overlay)}" ) def _inject_bg_id_enum( base_schema: Dict[str, Any], bg_ids: List[str] ) -> Dict[str, Any]: """schema deep-copy + ``camera_recommendations[].bg_id.enum`` 주입. D6 v4 L2 방어: LLM 출력을 정확한 set 으로 강제 (schema-level constraint). base_schema 는 매 호출마다 ``load_schema`` 가 새로 반환하지만, LRU 캐시 등 공유 가능성 + 같은 process 안 반복 호출 방어를 위해 deep-copy. Returns: 새 schema dict (mutation-safe). """ schema = copy.deepcopy(base_schema) cam_props = ( schema.get("properties", {}) .get("camera_recommendations", {}) .get("items", {}) .get("properties", {}) ) if "bg_id" in cam_props: # pattern 은 보존하되 enum 으로 좁힘. OpenAI structured output 은 둘 다 지원. cam_props["bg_id"] = { "type": "string", "pattern": cam_props["bg_id"].get("pattern", r"^L\d{2,3}B\d{2,3}$"), "enum": list(bg_ids), } return schema def validate_fp_prompt_output(output: Dict[str, Any], expected_fp_id: str) -> None: """3 invariants: 1. fp_id matches expected 2. t2i_prompt no non-ASCII (Korean/Hanja/kana) 3. t2i_prompt >= 30 chars """ if output.get("fp_id") != expected_fp_id: raise ValueError( f"fp_id mismatch: got {output.get('fp_id')!r}, expected {expected_fp_id!r}" ) t2i = output.get("t2i_prompt") or "" if _NON_ASCII_TEXT_RE.search(t2i): raise ValueError( f"floor_plan_prompt {expected_fp_id} t2i contains non-ASCII text" ) if len(t2i) < 30: raise ValueError( f"floor_plan_prompt {expected_fp_id} t2i too short ({len(t2i)})" ) def run_floor_plan_prompt( *, user_prompt: str, expected_fp_id: str, call_structured_fn: Callable[..., Dict[str, Any]], project_config: Optional[Dict[str, Any]] = None, opik_metadata: Optional[Dict[str, Any]] = None, max_retries: int = 3, backoff_base_sec: float = 2.0, sleep_fn: Callable[[float], None] = time.sleep, expected_bg_ids: Optional[Iterable[str]] = None, prompt_version: str = "5", creator_corrections_block: str = "", ) -> Dict[str, Any]: """LLM 호출 후 schema-validated dict 반환. Phase 8 v2: ``expected_bg_ids`` 가 주어지면 ``camera_recommendations[].bg_id`` 가 그 집합의 부분집합인지 추가 검증한다(``_validate_fp_prompt_extras``). None이면 cross-field 검증만 skip하고 결과는 dict 그대로 반환. D6 v4: ``expected_bg_ids`` 비어있지 않으면 schema deep-copy 후 ``bg_id.enum = sorted(expected_bg_ids)`` 주입 (L2 방어). validator 는 exact-set 검사 (L3 방어). schema 는 매 호출 deep-copy 되므로 호출 간 mutation 누수 없음. W19A: ``prompt_version`` ("5" default | "6" opt-in) 가 prompt_loader 의 versioned-load surface 로 plumbing 된다. v5 selector 는 5.202605201406 pack 을 정확히 pin 하고 validator 도 v5 동작 verbatim 유지. v6 selector 는 6.202605262300 pack 을 pin 하고 validator 가 모든 ``numbered_elements`` entry 의 ``base_layer_decision`` enum membership 까지 추가 검증한다. """ resolved = resolve_prompt_version(prompt_version) system = load_prompt(_MODULE, "system", version=resolved) # 제작자 정정 채널 (wave3 패턴, W21B scale-guideline 에서 FP 로 확장) — # 빈 문자열이면 기존과 byte-identical. v10 Rule 11 이 정정을 scale # evidence source 로 인용하므로 이 주입이 있어야 운영 계약이 정확하다. system += creator_corrections_block or "" base_schema = load_schema(_MODULE, "schema", version=resolved) last_err: Optional[Exception] = None bg_set: Set[str] = set(expected_bg_ids) if expected_bg_ids is not None else set() # L2: 비어있지 않을 때만 enum 주입. empty enum 은 invalid JSON Schema. # empty set 도 deep-copy 는 진행 — 호출 간 mutation 누수 차단 (test_run_does_not # _mutate_loaded_schema_across_calls). load_schema 는 file fallback 시 매번 # parse 하지만 DB cache 등 다른 source 가 같은 dict 를 공유할 가능성 방어. if bg_set: schema = _inject_bg_id_enum(base_schema, sorted(bg_set)) else: schema = copy.deepcopy(base_schema) for attempt in range(max_retries): try: result = call_structured_fn( step="floor_plan_prompt", system_prompt=system, user_prompt=user_prompt, response_schema=schema, project_config=project_config, schema_name="floor_plan_prompt", opik_metadata=opik_metadata, ) validate_fp_prompt_output(result, expected_fp_id) if expected_bg_ids is not None: _validate_fp_prompt_extras( fp_result=result, expected_bg_ids=bg_set, prompt_version=prompt_version, ) return result except Exception as exc: # T20 I3: LLM client errors (litellm/openai/httpx) 포함 last_err = exc logger.warning( "floor_plan_prompt %s attempt %d/%d failed: %s", expected_fp_id, attempt + 1, max_retries, exc, ) if attempt + 1 < max_retries: sleep_fn(backoff_base_sec * (attempt + 1)) raise FloorPlanPromptError( f"floor_plan_prompt {expected_fp_id} exhausted {max_retries} retries: {last_err}" )