"""FloorPlanRenderStep — Phase 7 Step 4. floor_plan_prompt 결과 + master plan의 depends_on_fp를 따라 PNG 생성. level 병렬 (compute_dag_levels). Phase 7 ImageAsset UPSERT (Phase 5 패턴 미러): - asset_type='floor_plan' - entity_id = primary_loc_id의 EntityCanon.id - variant_index = primary_loc_id 단위 0-base counter (sorted fp_id) - variant_label = 'v00'/'v01'... (counter 기반) - variant_type = fp_id (UPSERT 매치 키, 재실행 idempotent 보장) - is_primary = 1 (counter 0일 때만), 0 (그 외) - file_path = projects_root 기준 relative - prompt_used = t2i_prompt - generation_model = 'gpt-image-2' 도면간 depends_on_fp DAG는 거의 single-level이지만 generic helper로 안전 처리. """ from __future__ import annotations import json import logging import re import uuid from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from app.core.image_call_budget import bind_current_budget from app.core.step_runner import StepRunner from app.services.image_capture.annotate import annotate_generated_asset logger = logging.getLogger(__name__) SCHEMA_VERSION = 1 PROMPT_VERSION = "1" # fp_id whitelist: lowercase ascii + digits + underscore. path traversal 방어. _SAFE_FP_RE = re.compile(r"^[a-z0-9][a-z0-9_]*$") def _bind_step_identity(fn): """호출 시점의 **스텝 컨텍스트**를 캡처해 worker thread 에서 재설치한다. ★왜 필요한가 (감사 A-1 ③, 2026-08-28): `call_gpt_image_bytes` 가 쓰는 `ambient_call_meta()` 는 StepRunner 가 심어 둔 **thread-local** 스텝 컨텍스트를 읽는데, 그것은 worker thread 로 자동 전파되지 않는다. 그 한계는 `image_tracer.ambient_call_meta` 독스트링이 이미 명시하고 있고(「나중에 ThreadPool 로 나누면 신원이 다시 빈다」), 이 스텝이 정확히 그 자리였다 — 유료 gpt-image 호출 **133건이 `project_id`·`episode_id` 둘 다 NULL** 로 남아 프로젝트 단위로 「이 도면이 살아 있나」를 물을 수 없었다(실측 2026-08-25 까지). `bind_current_budget` · `bind_current_generation_context` 와 **같은 모양**이다. 전역 helper 를 새로 만들지 않고 이 스텝 안에 둔다 — 다른 ThreadPool 스텝이 같은 증상을 보이면 그때 옮긴다. 캡처된 컨텍스트가 없으면 no-op 이라 기존 동작과 byte-identical. """ from app.modules.llm.llm_client import (_get_thread_opik_meta, set_opik_context) captured = _get_thread_opik_meta() def _wrapped(*args, **kwargs): if not captured: return fn(*args, **kwargs) prev = _get_thread_opik_meta() set_opik_context(captured) try: return fn(*args, **kwargs) finally: set_opik_context(prev) return _wrapped def _resolve_openai_client(): """OpenAI 이미지 클라이언트 resolve. api_key 는 넘기지 않는다 — `openai_keys` 브로커가 활성 슬롯의 키를 정하고 키 수준 실패(billing/quota/인증) 시 보조 슬롯으로 전환한다(2026-07-30). bare `OpenAI()` 를 쓰면 os.environ 만 읽어 .env 키도, 슬롯 전환도 놓친다. timeout 은 settings.llm_timeout_image_gen (env LLM_TIMEOUT_IMAGE_GEN override 가능) — single source. default 600s 의 long hang (단일 image API 호출이 14분까지 wait 한 사고) 회귀 가드. retry 는 OpenAI SDK default (max 2). """ from app.core.openai_keys import openai_client from app.core.config import settings return openai_client( timeout=float(settings.llm_timeout_image_gen), ) def _compute_fp_primary_loc_ids(plans_map: Dict[str, Any]) -> Dict[str, str]: """fp_id → primary_loc_id 역참조 매핑. primary_loc_id = 그 floor plan을 ref하는 master_plan background의 loc_id (background.depends_on_fp 역참조). background는 그 floor plan과 같은 group의 것만 본다. 여러 loc에 걸치면 정렬 첫 값 (deterministic), ref하는 background가 0이면 "". 동일 fp_id가 여러 group에 등장하면 first-wins (먼저 등장한 group). 이 helper는 primary_loc 역참조만 담당한다 — fp_id의 safe-id 필터링 (`_SAFE_FP_RE`)은 caller(`_execute`) 책임이다. `verify_completion`과 `_execute`가 동일 매핑을 쓰도록 단일 소스로 추출. """ result: Dict[str, str] = {} for entry in plans_map.values(): if entry.get("status") != "ok": continue plan = entry.get("plan") or {} backgrounds = plan.get("backgrounds") or [] for fp in plan.get("floor_plans") or []: fid = fp.get("fp_id") if not fid or fid in result: continue ref_loc_ids = sorted({ bg.get("loc_id", "") or "" for bg in backgrounds if fid in (bg.get("depends_on_fp") or []) and bg.get("loc_id") }) result[fid] = ref_loc_ids[0] if ref_loc_ids else "" return result class FloorPlanRenderStep(StepRunner): def _config_hash(self) -> str: import hashlib import json as _json from app.core.config import settings payload = { "background_mode": settings.background_mode, "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_render: %s parse failed: %s", step_id, exc ) return None def verify_completion(self): """Phase 7 Step 4 산출물 무결성 검증 — Task 24. gate: settings.background_mode ∈ {"on","floor_plan_anchored"} 일 때만 실제 검증. off/chain_only는 step 자체가 skip되므로 expected=0 clean. expected source: ``background_master_plan.data.plans[gid].plan.floor_plans[].fp_id`` (production return shape — plan §4.4가 가정한 ``data.groups[].needs_floor_plan`` 는 production code에 존재하지 않음). renderable filter: ``floor_plan_prompt.data.floor_plans[fid].status=='ok'`` 인 fp만 후보. prompt 단계 실패한 fp는 production이 애초에 렌더하지 않으므로 verify에서도 제외해야 false-missing 방지. registerable narrowing: renderable 중 ``_register_image_assets``가 실제로 ImageAsset을 등록하는 집합 — {primary_loc_id 비어있지 않음 AND 그 location이 location EntityCanon으로 resolve 가능} — 만 expected에 포함. 미사용 location의 floor plan(ref background 0)처럼 등록 불가능한 fp를 영구 missing으로 오인하지 않기 위함. 제외분은 사유별 (no_primary_loc_id / missing_location_canon)로 metadata ``skipped_unregistered_floor_plan``에 기록. match key: ImageAsset(asset_type='floor_plan', variant_type=fp_id) + file_path 실재 — production ``_register_image_assets`` UPSERT와 동일. cleanup_artifacts override 안 함 → default noop (부분 재생성 보호). """ from app.core.config import settings from app.core.file_paths import resolve_image_path from app.core.integrity_report import CompletionReport from app.models.project import EntityCanon, ImageAsset # background_mode gate — production line 93 mirror if settings.background_mode not in {"on", "floor_plan_anchored"}: return CompletionReport( is_complete=True, missing=[], severity="clean", metadata={ "floor_plan_expected": 0, "floor_plan_found": 0, "skipped_reason": "background_mode_off", }, ) # expected: master_plan.data.plans[gid].plan.floor_plans[].fp_id (status=ok 만) plans_cp = self._load_prev_checkpoint("background_master_plan") plans_map = ((plans_cp or {}).get("data", {}) or {}).get("plans", {}) or {} expected_fp_ids: set[str] = set() for entry in plans_map.values(): if entry.get("status") != "ok": continue for fp in (entry.get("plan") or {}).get("floor_plans") or []: fid = fp.get("fp_id") if fid: expected_fp_ids.add(fid) # renderable filter: floor_plan_prompt.data.floor_plans[fid].status=='ok' prompt_cp = self._load_prev_checkpoint("floor_plan_prompt") prompt_fps = ( ((prompt_cp or {}).get("data", {}) or {}).get("floor_plans", {}) or {} ) renderable_fp_ids = { fid for fid in expected_fp_ids if (prompt_fps.get(fid) or {}).get("status") == "ok" } # registerable narrowing — _register_image_assets와 parity. # production은 renderable fp 중 {primary_loc_id 비어있지 않음 AND 그 # location이 EntityCanon으로 resolve 가능}인 것만 ImageAsset으로 등록한다. # verify의 expected도 동일 집합으로 좁혀야 — 미사용 location의 floor plan # (ref background 0 → primary_loc_id 빈값)처럼 애초에 등록 불가능한 fp를 # 영구 missing으로 오인하지 않는다. 제외분은 사유별로 metadata에 보존한다. # (unsafe fp_id 등 그 외 edge까지 닫지는 않는다 — _execute의 _SAFE_FP_RE는 # 별개 관심사.) primary_loc_by_fp = _compute_fp_primary_loc_ids(plans_map) location_canon_short_ids = { c.short_id for c in self.db.query(EntityCanon).filter( EntityCanon.project_id == self.project_id, EntityCanon.entity_type == "location", ).all() if c.short_id } registerable_fp_ids: set[str] = set() skipped_no_primary: List[str] = [] skipped_no_canon: List[str] = [] for fid in renderable_fp_ids: primary_loc = primary_loc_by_fp.get(fid, "") or "" if not primary_loc: skipped_no_primary.append(fid) elif primary_loc not in location_canon_short_ids: skipped_no_canon.append(fid) else: registerable_fp_ids.add(fid) skipped_meta: Dict[str, List[str]] = {} if skipped_no_primary: skipped_meta["no_primary_loc_id"] = sorted(skipped_no_primary) if skipped_no_canon: skipped_meta["missing_location_canon"] = sorted(skipped_no_canon) for reason, fids in skipped_meta.items(): logger.info( "floor_plan_render verify: %d fp skipped (%s) — %s", len(fids), reason, fids, ) def _meta(**base: Any) -> Dict[str, Any]: # skipped metadata는 expected==0 clean 경로 포함 모든 반환에 보존 # (all-skipped 케이스 디버깅 가능하도록). m: Dict[str, Any] = dict(base) if skipped_meta: m["skipped_unregistered_floor_plan"] = skipped_meta return m expected = len(registerable_fp_ids) if expected == 0: return CompletionReport( is_complete=True, missing=[], severity="clean", metadata=_meta(floor_plan_expected=0, floor_plan_found=0), ) rows = self.db.query(ImageAsset).filter( ImageAsset.project_id == self.project_id, ImageAsset.episode_id == self.episode_id, ImageAsset.asset_type == "floor_plan", ImageAsset.variant_type.in_(registerable_fp_ids), ).all() fp_ids_with_file = set() for r in rows: p = resolve_image_path(r.file_path) if p and p.exists(): fp_ids_with_file.add(r.variant_type) found = len(fp_ids_with_file) missing = sorted(registerable_fp_ids - fp_ids_with_file) if found < expected: return CompletionReport( is_complete=False, missing=missing, severity="missing" if found == 0 else "partial", metadata=_meta( floor_plan_expected=expected, floor_plan_found=found, rows=len(rows), ), ) return CompletionReport( is_complete=True, missing=[], severity="clean", metadata=_meta( floor_plan_expected=expected, floor_plan_found=found, rows=len(rows), ), ) 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 { "applicable_count": 0, "completed_count": 0, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {}, } prompts_cp = self._load_prev_checkpoint("floor_plan_prompt") plans_cp = self._load_prev_checkpoint("background_master_plan") prompts_map = ( ((prompts_cp or {}).get("data", {}) or {}).get("floor_plans", {}) or {} ) plans_map = ((plans_cp or {}).get("data", {}) or {}).get("plans", {}) or {} # fp dependency 그래프 구성 (master_plan 기준; depends_on_fp[0]를 parent로 사용) # 또한 fp_id → primary_loc_id 매핑을 도출 (ImageAsset entity_id 용). # floor_plan spec 자체엔 loc_id 없음 → bg.depends_on_fp 역참조하여 결정. # verify_completion과 동일 매핑을 쓰도록 helper로 단일화. primary_loc_by_fp = _compute_fp_primary_loc_ids(plans_map) items: Dict[str, Dict[str, Any]] = {} order: List[str] = [] for gid, entry in plans_map.items(): if entry.get("status") != "ok": continue plan = entry.get("plan") or {} for fp in plan.get("floor_plans") or []: fid = fp.get("fp_id") if not fid or not _SAFE_FP_RE.match(fid): logger.error( "floor_plan_render: unsafe fp_id %r — skip", fid ) continue if fid in items: # 중복 fp_id (다른 group에서 동일 ID 등장 시 first-wins) continue deps = fp.get("depends_on_fp") or [] items[fid] = { "parent_id": deps[0] if deps else "", "spec": fp, "group_id": gid, "primary_loc_id": primary_loc_by_fp.get(fid, ""), } order.append(fid) # 실제 렌더 가능한 fp = prompt가 ok 인 것만 renderable = { fid for fid in order if prompts_map.get(fid, {}).get("status") == "ok" } if not renderable: return { "applicable_count": 1, "completed_count": 1, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"floor_plans": {}}, } from app.modules.pipeline._dag_levels import compute_dag_levels from app.modules.pipeline.floor_plan_render import render_one_floor_plan client = _resolve_openai_client() levels = compute_dag_levels( order, items, renderable, parent_field="parent_id" ) image_dir = ( Path(settings.projects_dir) / self.project_id / "episodes" / self.episode_id / "images" / "floor_plan" ) image_dir.mkdir(parents=True, exist_ok=True) image_dir_resolved = image_dir.resolve() rendered_paths: Dict[str, Path] = {} results: Dict[str, Any] = {} failed = 0 from app.modules.pipeline._workers import resolve_workers max_workers = resolve_workers(default=3, cap=8) def _process(fid: str, snapshot: Dict[str, Path]) -> Tuple[str, Dict[str, Any]]: prompt_entry = prompts_map[fid] spec = items[fid]["spec"] ref_paths: List[Path] = [] for dep in spec.get("depends_on_fp") or []: p = snapshot.get(dep) if p is not None and p.exists(): ref_paths.append(p) out_path = image_dir / f"{fid}.png" # path traversal 2중 가드 (fp_id 정규식 + resolve relative_to) try: out_path.resolve().relative_to(image_dir_resolved) except (ValueError, OSError): logger.error( "floor_plan_render: out_path escapes image_dir for %s", fid ) return fid, { "status": "rejected_path", "png_path": "", "attempts": 0, "ref_used": "text_only", "error": "out_path escapes image_dir", "t2i_prompt": prompt_entry.get("t2i_prompt", ""), } res = render_one_floor_plan( openai_client=client, image_model="gpt-image-2.5-sunburst", prompt=prompt_entry["t2i_prompt"], out_path=out_path, ref_paths=ref_paths, fp_id=fid, ) return fid, { "status": res.status, "png_path": res.png_path, "attempts": res.attempts, "ref_used": res.ref_used, "error": res.error, "t2i_prompt": prompt_entry.get("t2i_prompt", ""), } for level in levels: if not level: continue # snapshot at level start so threads see deterministic parent set snapshot = dict(rendered_paths) level_workers = max(1, min(max_workers, len(level))) with ThreadPoolExecutor(max_workers=level_workers) as pool: # W20E5 Codex B1 — propagate parent-thread image-call # budget into pool workers. # # ★신원도 같이 넘긴다 (감사 A-1 ③, 2026-08-28). # `call_gpt_image_bytes → ambient_call_meta()` 는 # StepRunner 의 **thread-local** 스텝 컨텍스트를 읽는데, # 그것은 worker thread 로 자동 전파되지 않는다 # (`image_tracer.ambient_call_meta` 독스트링이 이 한계를 # 명시하고 `bind_current_generation_context` 를 본보기로 # 가리킨다). 그래서 이 경로의 유료 gpt-image 호출 **133건이 # `project_id`·`episode_id` 둘 다 NULL** 로 남았다 — # 프로젝트 단위로 「이 도면이 살아 있나」를 물을 수 없었다. # # ★전역 helper 를 새로 만들지 않는다 — 이 자리에서 부모의 # 메타를 캡처해 worker 시작 때 다시 설치하고 끝나면 # 되돌린다(`bind_current_budget` 과 같은 모양). _submit_process = _bind_step_identity( bind_current_budget(_process)) futures = { pool.submit(_submit_process, fid, snapshot): fid for fid in level } for fut in as_completed(futures): fid = futures[fut] try: _fid, res = fut.result() except Exception as exc: logger.error( "floor_plan_render: fp %s thread raised: %s", fid, exc, ) res = { "status": "failed", "png_path": "", "attempts": 0, "ref_used": "text_only", "error": str(exc)[:200], "t2i_prompt": prompts_map.get(fid, {}).get( "t2i_prompt", "" ), } results[fid] = res if res.get("status") == "ok" and res.get("png_path"): rendered_paths[fid] = Path(res["png_path"]) else: failed += 1 # input order 보존 ordered = {fid: results[fid] for fid in order if fid in results} # ── ImageAsset UPSERT (best-effort) — main thread, after all renders ── try: self._register_image_assets(ordered, items) except Exception as exc: logger.error( "floor_plan_render: ImageAsset DB sync failed — files on disk: %s", exc, ) try: self.db.rollback() except Exception as rb_exc: logger.error( "floor_plan_render: rollback also failed: %s", rb_exc ) return { "applicable_count": 1, "completed_count": 1 if failed == 0 else 0, "failed_count": failed, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"floor_plans": ordered}, } def _register_image_assets( self, results: Dict[str, Dict[str, Any]], items: Dict[str, Dict[str, Any]], ) -> None: """Phase 7 — fp PNG를 ImageAsset(asset_type='floor_plan') UPSERT. Phase 5 ``location_floor_plan_step._register_image_assets`` 패턴을 미러. Phase 5는 location당 1 fp 였으므로 variant_index=0 고정. Phase 7은 primary_location당 multiple fp 가능 → primary_loc 단위 0-base counter + variant_type=fp_id를 UPSERT 매치 키로 사용 (idempotent). match key (application-level): (project_id, episode_id, asset_type, entity_id, variant_type=fp_id). """ from app.core.config import settings from app.models.project import EntityCanon, ImageAsset # location EntityCanon 매핑 (short_id → canon_id) canons = ( self.db.query(EntityCanon) .filter( EntityCanon.project_id == self.project_id, EntityCanon.entity_type == "location", ) .all() ) canon_id_by_short: Dict[str, str] = { c.short_id: c.id for c in canons if c.short_id } if not canon_id_by_short: logger.warning( "floor_plan_render: no location EntityCanon — skip ImageAsset DB sync" ) return # primary_location 단위 0-base counter — sorted fp_id 로 deterministic. # Phase 5 chain_bg와 동일 전략 (location_variant_counter). ok_fp_ids = sorted( fid for fid, res in results.items() if res.get("status") == "ok" ) loc_to_fps: Dict[str, List[str]] = {} for fid in ok_fp_ids: entry = items.get(fid) or {} primary_loc = entry.get("primary_loc_id", "") or "" if not primary_loc: continue loc_to_fps.setdefault(primary_loc, []).append(fid) # fp_id → variant_index (location 내 정렬 순서) variant_idx_by_fp: Dict[str, int] = {} for loc_short, fids in loc_to_fps.items(): for idx, fid in enumerate(fids): variant_idx_by_fp[fid] = idx now = datetime.now(timezone.utc).isoformat() registered = 0 for fp_id, result in results.items(): if result.get("status") != "ok": continue png_path_str = result.get("png_path", "") or "" if not png_path_str: continue entry = items.get(fp_id) or {} primary_loc = entry.get("primary_loc_id", "") or "" if not primary_loc: # W20E6-C: an FP whose master_plan group has no # background referencing it (depends_on_fp) has no # primary_location to bind its ImageAsset to. This is # by-design -- verify_completion already narrows the # expected set via skipped_unregistered_floor_plan # metadata. Surface a parity log so the skip is # auditable instead of silent (mirrors the no_canon # warning two lines below). logger.info( "floor_plan_render: %s skipped DB register " "(no_primary_loc_id) -- no master_plan background " "depends_on_fp this fp", fp_id, ) continue canon_id = canon_id_by_short.get(primary_loc) if not canon_id: logger.warning( "floor_plan_render: %s primary_location %s has no canon — skip DB", fp_id, primary_loc, ) continue # Phase 3 컨벤션: file_path 는 projects_root 기준 relative. # to_relative_image_path 는 root 외부면 절대 그대로 반환 (e2e tmp_path 호환). from app.core.file_paths import to_relative_image_path rel_png_path = to_relative_image_path(png_path_str) variant_index = variant_idx_by_fp.get(fp_id, 0) variant_label = f"v{variant_index:02d}" is_primary = 1 if variant_index == 0 else 0 existing = ( self.db.query(ImageAsset) .filter_by( project_id=self.project_id, episode_id=self.episode_id, asset_type="floor_plan", entity_id=canon_id, variant_type=fp_id, ) .first() ) if existing: existing.file_path = rel_png_path existing.prompt_used = result.get("t2i_prompt", "") or "" existing.variant_index = variant_index existing.variant_label = variant_label existing.t2i_guide = None existing.is_primary = is_primary existing.status = "generated" existing.generation_model = "gpt-image-2.5-sunburst" _fp_row = existing else: _fp_row = ImageAsset( id=str(uuid.uuid4()), project_id=self.project_id, episode_id=self.episode_id, asset_type="floor_plan", entity_id=canon_id, variant_index=variant_index, variant_label=variant_label, variant_type=fp_id, # traceability + UPSERT 매치 키 t2i_guide=None, file_path=rel_png_path, prompt_used=result.get("t2i_prompt", "") or "", generation_model="gpt-image-2.5-sunburst", status="generated", is_primary=is_primary, created_at=now, ) self.db.add(_fp_row) # persist-all Wave 1 — 도면=T2I 루트(입력이미지 없음). UPSERT 양 분기 backfill. annotate_generated_asset(_fp_row, pipeline_role="floor_plan", input_image_ids=[]) registered += 1 if registered: self.db.commit() logger.info( "floor_plan_render: registered %d floor_plan ImageAssets", registered, )