"""background_chain_render — chain plan에 따라 location별 배경 PNG를 체이닝 생성. PR #4 (2026-04-27). background_chain_planning 결과를 입력으로 받아: Phase 1 (per node): gpt-5.5로 노드별 t2i prompt 생성 Phase 2 (per node): gpt-image-2 images.edit로 PNG 생성 - root anchor: location의 entity reference image를 ref로 사용 (없으면 text-only generate) - 자식 node: 부모 PNG ref + 자식 prompt → images.edit PromptSanitizer로 moderation 차단 시 film_previs/movie_poster/aftermath retry. 체크포인트 출력: data.locations[loc_id].nodes[].{...planning fields..., t2i_prompt, image_path, render_status, render_attempts, sanitize_strategies, shot_guides} # Phase 2: list of {shot_id, guide} data.locations[loc_id].shot_backgrounds[].{scene_index, shot_index, shot_label, node_id, image_path} """ from __future__ import annotations import json import logging import os import re import time from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from app.core.image_call_budget import ( ImageCallBudgetExceeded, bind_current_budget, reserve_current_call, ) from app.modules.llm.gpt_image_primitive import call_gpt_image_bytes from app.modules.llm.llm_client import call_structured from app.modules.pipeline._dag_levels import compute_dag_levels from app.modules.prompt_loader import load_prompt, load_schema from app.modules.prompt_sanitizer import PromptSanitizer logger = logging.getLogger(__name__) _NON_ASCII_TEXT_RE = re.compile( r"[ㄱ-ㆎ가-힣" r"一-鿿㐀-䶿豈-﫿" r"぀-ヿ]" ) # OpenAI moderation 차단 keyword 휴리스틱 (block_categories 추정) _MODERATION_KEYWORDS = ( "moderation", "safety", "content_policy", "prohibited", "policy", "blocked", "violates", "violation", ) # node_id allowlist — LLM/체크포인트에서 오는 임의 문자열을 file path로 사용하기 # 전 path traversal 차단 (Codex PR #4 HIGH 1). _SAFE_NODE_ID_RE = re.compile(r"^[a-z0-9][a-z0-9_]*$") # Background-only sanitize prefix — PromptSanitizer가 일반 T2I용으로 인물/자세를 # 요청하는 generic strategy를 사용하므로, background-only 컨트랙트 강제용 prefix를 # 추가로 prepend (Codex PR #4 HIGH 3). _BACKGROUND_ONLY_REINFORCEMENT = ( "BACKGROUND-ONLY architectural still — empty space, NO people, NO faces, " "NO body posture, NO action, NO weapons, NO blood. " "Photoreal scene without any human figures.\n\n" ) # ────────────────────────────────────────────── # Phase 1: per-node t2i prompt generation # ────────────────────────────────────────────── def _build_user_prompt_for_node( location_id: str, location_description: str, node: Dict[str, Any], parent: Optional[Dict[str, Any]], shots_in_node: List[Dict[str, Any]], ) -> str: template = load_prompt("background_chain_render", "user_template") if parent: parent_block = ( f"id: {parent.get('id', '')}\n" f"label: {parent.get('label', '')}\n" f"description: {parent.get('description', '')}\n" f"[The parent's already-rendered photo will be the reference image for this node. " f"Reuse parent's atmosphere/material/lighting language. Explicitly mention each " f"shared_visual_anchors_with_parent entry.]" ) else: parent_block = ( "(this is a ROOT anchor — no parent. The location reference image will be the input. " "Thoroughly describe wall/floor/ceiling/lighting/palette since later children inherit " "from this rendered photo.)" ) shots_lines: List[str] = [] for s in shots_in_node: shots_lines.append( f"- {s.get('shot_id', '?')}: {s.get('description', '')}" ) shots_block = "\n".join(shots_lines) if shots_lines else "(no shots)" def _esc(s: str) -> str: return (s or "").replace("{", "{{").replace("}", "}}") return template.format( location_id=_esc(location_id), location_description=_esc(location_description) or "(no description)", node_id=_esc(node.get("id", "")), node_kind=_esc(node.get("kind", "")), node_label=_esc(node.get("label", "")), node_description=_esc(node.get("description", "")), shared_anchors_json=_esc(json.dumps( node.get("shared_visual_anchors_with_parent", []), ensure_ascii=False, )), parent_block=_esc(parent_block), shot_count=len(shots_in_node), shots_block=_esc(shots_block), ) def generate_node_prompt( location_id: str, location_description: str, node: Dict[str, Any], parent: Optional[Dict[str, Any]], shots_in_node: List[Dict[str, Any]], opik_metadata: Optional[Dict[str, Any]] = None, ) -> Tuple[str, List[Dict[str, Any]]]: """노드 1개에 대한 t2i prompt + shot_guides 생성. 영어 ASCII만 검증. Returns: (t2i_prompt, shot_guides) - t2i_prompt: PNG 생성용 prompt (영어 한 단락) - shot_guides: list of {"shot_id": str, "guide": str}. v2 schema 미지원 prompt 버전 사용 시 빈 list 반환 (legacy fallback). """ system = load_prompt("background_chain_render", "system") schema = load_schema("background_chain_render", "schema") user = _build_user_prompt_for_node( location_id, location_description, node, parent, shots_in_node, ) result = call_structured( step="background_chain_render", system_prompt=system, user_prompt=user, response_schema=schema, opik_metadata=opik_metadata, ) t2i = (result.get("t2i_prompt") or "").strip() if _NON_ASCII_TEXT_RE.search(t2i): raise ValueError( f"node {node.get('id')!r} t2i_prompt contains non-ASCII text " f"(Korean/Hanja/kana detected — universal-noun rule violated)" ) # v2 schema: shot_guides 추출. 누락 시 빈 list (v1 호환 fallback) shot_guides_raw = result.get("shot_guides") or [] shot_guides: List[Dict[str, Any]] = [] if isinstance(shot_guides_raw, list): for sg in shot_guides_raw: if not isinstance(sg, dict): continue sid = sg.get("shot_id") or "" guide = sg.get("guide") or "" if sid and guide: shot_guides.append({"shot_id": sid, "guide": guide}) return t2i, shot_guides # ────────────────────────────────────────────── # Phase 2: image generation (gpt-image-2) # ────────────────────────────────────────────── def _looks_like_moderation_block(exc: Exception) -> bool: msg = str(exc).lower() return any(k in msg for k in _MODERATION_KEYWORDS) def render_node_image( openai_client: Any, image_model: str, prompt: str, out_path: Path, ref_path: Optional[Path] = None, # backward-compat sanitizer: Optional[PromptSanitizer] = None, size: str = "1024x1024", quality: str = "high", max_attempts: int = 4, ref_paths: Optional[List[Path]] = None, # 신규 — 우선 ) -> Dict[str, Any]: """단일 노드 이미지 생성. ref_paths 1+ 지원 (multi-image edit, gpt-image-2). ref_paths 우선. ref_path는 backward-compat (단일 ref). 둘 다 None이면 [] (text_only). Returns: {"status": "ok"|"failed", "attempts": int, "strategies": [str], "final_block_reason": str | None, "ref_used": str} ref_used 라벨: - "text_only" — 0 refs - "ref" — 1 ref (기존 호환) - "refs_N" — N >= 2 refs (multi-image edit) 의미 라벨(parent/location)은 caller(render_one_location)가 결정. """ # backward-compat: ref_path 단일 → ref_paths = [ref_path] if ref_paths is None: ref_paths = [ref_path] if ref_path is not None else [] # 존재하는 파일만 필터 (None / 미존재 제외) valid_refs = [p for p in ref_paths if p is not None and p.exists()] info: Dict[str, Any] = { "status": "failed", "attempts": 0, "strategies": [], "final_block_reason": None, "ref_used": ( "text_only" if not valid_refs else "ref" if len(valid_refs) == 1 else f"refs_{len(valid_refs)}" ), } current_prompt = prompt for attempt in range(1, max_attempts + 1): info["attempts"] = attempt try: # gpt-image 호출 + b64 decode 는 primitive wrapper 로 위임(생성물 capture # 동시 수행, scope 미배선이면 no-op). reserve/검증/write/retry 는 여기 유지. # wrapper 가 ref 개수로 edit_multi(ExitStack list)/edit_single/generate 재현. call_kwargs = { "model": image_model, "size": size, "quality": quality, "n": 1, } if len(valid_refs) >= 2: reserve_current_call(source="background_chain_render.edit_multi") png = call_gpt_image_bytes( openai_client, mode="edit", prompt=current_prompt, ref_paths=valid_refs, call_kwargs=call_kwargs, capture_role="background_chain_node", capture_metadata={ "budget_source": "background_chain_render.edit_multi", "ref_used": info["ref_used"], "ref_count": len(valid_refs), }, ) elif len(valid_refs) == 1: reserve_current_call(source="background_chain_render.edit_single") png = call_gpt_image_bytes( openai_client, mode="edit", prompt=current_prompt, ref_paths=valid_refs, call_kwargs=call_kwargs, capture_role="background_chain_node", capture_metadata={ "budget_source": "background_chain_render.edit_single", "ref_used": info["ref_used"], "ref_count": 1, }, ) else: reserve_current_call(source="background_chain_render.generate") png = call_gpt_image_bytes( openai_client, mode="generate", prompt=current_prompt, ref_paths=None, call_kwargs=call_kwargs, capture_role="background_chain_node", capture_metadata={ "budget_source": "background_chain_render.generate", "ref_used": info["ref_used"], "ref_count": 0, }, ) if not png: raise RuntimeError("empty b64 response") out_path.write_bytes(png) info["status"] = "ok" return info except ImageCallBudgetExceeded: raise except Exception as exc: if _looks_like_moderation_block(exc): logger.warning( "render_node_image: moderation block (attempt %d): %s", attempt, str(exc)[:160], ) if sanitizer is None or attempt >= max_attempts: info["final_block_reason"] = str(exc)[:200] return info try: s_attempt = min(attempt, 3) sanitize_result = sanitizer.sanitize( original_prompt=current_prompt, block_reason="SAFETY", block_categories=[], attempt=s_attempt, ) sanitized = sanitize_result["sanitized_prompt"] # background-only contract reinforcement (Codex HIGH 3): # generic sanitizer는 인물/자세 prefix를 추가하므로 background # 전용 reinforcer를 앞에 prepend하여 사람/액션 명시 차단. if not sanitized.lstrip().startswith("BACKGROUND-ONLY"): sanitized = _BACKGROUND_ONLY_REINFORCEMENT + sanitized current_prompt = sanitized info["strategies"].append(sanitize_result.get("strategy")) except Exception as se: logger.error("render_node_image: sanitize failed: %s", se) info["final_block_reason"] = f"sanitize_failed: {se}" return info continue # non-moderation transient → 짧게 재시도, 단 마지막 attempt면 종료 logger.error("render_node_image: non-moderation error attempt %d: %s", attempt, str(exc)[:200]) if attempt >= max_attempts: info["final_block_reason"] = str(exc)[:200] return info time.sleep(min(2 * attempt, 5)) return info # ────────────────────────────────────────────── # 진입점 — location별 chain 렌더 # ────────────────────────────────────────────── def _resolve_location_ref_path( location_id: str, location_ref_paths: Dict[str, Path], ) -> Optional[Path]: p = location_ref_paths.get(location_id) if p and p.exists(): return p return None def render_one_location( location_id: str, location_data: Dict[str, Any], image_dir: Path, location_ref_paths: Dict[str, Path], openai_client: Any, image_model: str, sanitizer: Optional[PromptSanitizer], size: str = "1024x1024", quality: str = "high", max_attempts: int = 4, opik_metadata: Optional[Dict[str, Any]] = None, shot_meta_by_id: Optional[Dict[str, Dict[str, Any]]] = None, floor_plan_path: Optional[Path] = None, ) -> Dict[str, Any]: """단일 location chain plan을 받아 노드별 이미지를 순차 생성. location 안에서는 부모-자식 종속이 있어 sequential. location 간은 caller가 병렬화. """ loc_image_dir = image_dir / location_id loc_image_dir.mkdir(parents=True, exist_ok=True) nodes_in: List[Dict[str, Any]] = list(location_data.get("nodes", []) or []) nodes_by_id: Dict[str, Dict[str, Any]] = {n["id"]: n for n in nodes_in} execution_order: List[str] = list(location_data.get("execution_order", []) or []) location_description = ( location_data.get("location_description") or location_data.get("rationale_summary") or "" ) rendered_paths: Dict[str, Path] = {} enriched_nodes: Dict[str, Dict[str, Any]] = {} failed_nodes = 0 for node_id in execution_order: node = nodes_by_id.get(node_id) if not node: logger.error( "render_one_location: %s execution_order entry %r not in nodes — skip", location_id, node_id, ) continue # Path traversal 차단 (Codex HIGH 1) — node_id가 LLM/체크포인트에서 # 온 임의 문자열이므로 snake_case allowlist + resolve guard 둘 다 적용. if not _SAFE_NODE_ID_RE.match(node_id or ""): logger.error( "render_one_location: %s node_id %r not snake_case ASCII — skip", location_id, node_id, ) enriched_nodes[node_id] = { **node, "render_status": "rejected_node_id", "render_error": "node_id must match ^[a-z0-9][a-z0-9_]*$", } failed_nodes += 1 continue parent_id = (node.get("parent_id") or "").strip() parent = nodes_by_id.get(parent_id) if parent_id else None shots_in_node: List[Dict[str, Any]] = [] for sid in node.get("shot_ids", []) or []: if shot_meta_by_id and sid in shot_meta_by_id: meta = dict(shot_meta_by_id[sid]) meta.setdefault("shot_id", sid) shots_in_node.append(meta) else: shots_in_node.append({"shot_id": sid, "description": ""}) # Phase 1: t2i prompt + shot_guides try: t2i_prompt, shot_guides = generate_node_prompt( location_id=location_id, location_description=location_description, node=node, parent=parent, shots_in_node=shots_in_node, opik_metadata=opik_metadata, ) except Exception as exc: logger.error( "render_one_location: %s/%s prompt gen failed: %s", location_id, node_id, exc, ) enriched_nodes[node_id] = { **node, "t2i_prompt": "", "image_path": "", "render_status": "prompt_failed", "render_attempts": 0, "sanitize_strategies": [], "render_error": str(exc)[:200], "shot_guides": [], } failed_nodes += 1 continue # Phase 2: image — ref 우선순위: parent PNG → location ref → text_only. # 각 단계에서 file existence 재검증 (parent PNG는 동시 thread에서 # truncate되거나 disk 이슈로 사라질 수 있음). render_node_image의 # `info["ref_used"]`는 ref/text_only만 알므로 호출자가 의미 분류 결정. out_path = loc_image_dir / f"{node_id}.png" # 2차 가드 — resolve 후 loc_image_dir 안에 머무르는지 확인 try: out_path.resolve().relative_to(loc_image_dir.resolve()) except (ValueError, OSError) as exc: logger.error( "render_one_location: %s/%s out_path escapes loc_image_dir: %s — skip", location_id, node_id, exc, ) enriched_nodes[node_id] = { **node, "render_status": "rejected_path", "render_error": "out_path resolved outside location dir", } failed_nodes += 1 continue # Phase 4: ref 우선순위 # 1순위: floor_plan PNG (있으면 무조건) # 2순위: parent (자식 노드) 또는 location ref (root 노드) — 둘 중 하나 # 3순위: text_only (refs 비었음) ref_paths: List[Path] = [] parent_used = False location_used = False if floor_plan_path is not None and floor_plan_path.exists(): ref_paths.append(floor_plan_path) if parent_id and parent_id in rendered_paths: candidate = rendered_paths[parent_id] if candidate.exists(): ref_paths.append(candidate) parent_used = True if not parent_used: candidate = _resolve_location_ref_path(location_id, location_ref_paths) if candidate is not None: ref_paths.append(candidate) location_used = True floor_plan_present = ( floor_plan_path is not None and floor_plan_path.exists() ) if not ref_paths: ref_used = "text_only" elif floor_plan_present and parent_used: ref_used = "floor_plan+parent" elif floor_plan_present and location_used: ref_used = "floor_plan+location" elif floor_plan_present: ref_used = "floor_plan_only" elif parent_used: ref_used = "parent" elif location_used: ref_used = "location" else: ref_used = "text_only" result = render_node_image( openai_client=openai_client, image_model=image_model, prompt=t2i_prompt, out_path=out_path, ref_paths=ref_paths, sanitizer=sanitizer, size=size, quality=quality, max_attempts=max_attempts, ) # render_node_image의 ref_used("ref"/"refs_N"/"text_only")는 의미 분류에 # 무지하므로 caller가 결정한 라벨로 무조건 override. result["ref_used"] = ref_used if result["status"] == "ok": rendered_paths[node_id] = out_path enriched_nodes[node_id] = { **node, "t2i_prompt": t2i_prompt, "image_path": str(out_path), "render_status": "ok", "render_attempts": result["attempts"], "sanitize_strategies": result["strategies"], "ref_used": ref_used, "shot_guides": shot_guides, } else: enriched_nodes[node_id] = { **node, "t2i_prompt": t2i_prompt, "image_path": "", "render_status": "failed", "render_attempts": result["attempts"], "sanitize_strategies": result["strategies"], "render_error": result.get("final_block_reason") or "", "ref_used": ref_used, "shot_guides": shot_guides, } failed_nodes += 1 # shot_backgrounds: shot_id → 첫 매칭 노드의 image_path shot_backgrounds: List[Dict[str, Any]] = [] seen_shots: set = set() for nid in execution_order: n = enriched_nodes.get(nid) if not n or not n.get("image_path"): continue for sid in n.get("shot_ids", []) or []: if sid in seen_shots: continue seen_shots.add(sid) m = re.match(r"S(\d+)_Shot(\d+)", sid) if not m: continue scene_index = int(m.group(1)) shot_index = int(m.group(2)) shot_backgrounds.append({ "scene_index": scene_index, "shot_index": shot_index, "shot_label": sid, "node_id": nid, "image_path": n["image_path"], }) return { "location_id": location_id, "location_name": location_data.get("location_name", location_id), "rationale_summary": location_data.get("rationale_summary", ""), "nodes": [enriched_nodes[nid] for nid in execution_order if nid in enriched_nodes], "shot_backgrounds": shot_backgrounds, "failed_node_count": failed_nodes, "total_node_count": len(execution_order), } def run_background_chain_render( planning_data: Dict[str, Any], image_dir: Path, location_ref_paths: Dict[str, Path], openai_client: Any, image_model: str = "gpt-image-2", sanitizer: Optional[PromptSanitizer] = None, size: str = "1024x1024", quality: str = "high", max_attempts: int = 4, opik_metadata: Optional[Dict[str, Any]] = None, shot_meta_by_id: Optional[Dict[str, Dict[str, Any]]] = None, max_workers: int = 2, floor_plan_paths: Optional[Dict[str, Path]] = None, planner_chain_order: Optional[List[str]] = None, planner_floor_plan_specs: Optional[Dict[str, Dict[str, Any]]] = None, ) -> Dict[str, Any]: """전체 chain 렌더 진입점. 두 경로: - planner_chain_order=None (LEGACY, Phase 4): location 단위 ThreadPool 병렬 (max_workers=2). 각 location 내부는 부모-자식 sequential. mode=off / chain_only 회귀 보장. - planner_chain_order=List[str] (PLANNER-DRIVEN, Phase 5): chain_bg_order에 따라 group_id별 SEQUENTIAL 처리. variant_index/variant_label은 location_id 단위로 v01부터 단조 증가. floor_plan PNG 1순위 ref, parent_group PNG 2순위, location ref 3순위. 결과는 data.groups[group_id]에 저장. floor_plan_paths: location_id → floor_plan PNG path. legacy/planner 양쪽 모두 1순위 ref로 사용. 빈 dict이면 기존 flow 유지. planner_floor_plan_specs: fp_id → {primary_location_id, building_group, location_ids, ...}. building_group reverse lookup용. """ if sanitizer is None: sanitizer = PromptSanitizer() floor_plan_paths = floor_plan_paths or {} # ── Phase 5 분기 ── if planner_chain_order is not None: return _run_planner_driven_render( planner_chain_order=planner_chain_order, planning_groups=planning_data.get("groups") or {}, floor_plan_specs=planner_floor_plan_specs or {}, floor_plan_paths=floor_plan_paths, image_dir=image_dir, location_ref_paths=location_ref_paths, openai_client=openai_client, image_model=image_model, sanitizer=sanitizer, size=size, quality=quality, max_attempts=max_attempts, opik_metadata=opik_metadata, shot_meta_by_id=shot_meta_by_id, ) # ── Phase 4 LEGACY 경로 (변경 없음) ── locations_in: Dict[str, Dict[str, Any]] = ( planning_data.get("locations") or {} ) if not locations_in: logger.info("background_chain_render: no locations in planning data") return {"locations": {}, "_failed_count": 0} locations_out: Dict[str, Dict[str, Any]] = {} total_failed_nodes = 0 def _process(loc_id: str, data: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]: if data.get("status") != "ok" and not data.get("nodes"): return loc_id, { "location_id": loc_id, "location_name": data.get("location_name", loc_id), "skipped_reason": "no nodes in planning data", "nodes": [], "shot_backgrounds": [], "failed_node_count": 0, "total_node_count": 0, } try: return loc_id, render_one_location( location_id=loc_id, location_data=data, image_dir=image_dir, location_ref_paths=location_ref_paths, openai_client=openai_client, image_model=image_model, sanitizer=sanitizer, size=size, quality=quality, max_attempts=max_attempts, opik_metadata=opik_metadata, shot_meta_by_id=shot_meta_by_id, floor_plan_path=floor_plan_paths.get(loc_id), ) except Exception as exc: logger.error("background_chain_render: %s exception: %s", loc_id, exc, exc_info=True) return loc_id, { "location_id": loc_id, "location_name": data.get("location_name", loc_id), "error": str(exc), "nodes": [], "shot_backgrounds": [], "failed_node_count": 1, "total_node_count": len(data.get("nodes", [])), } workers = min(max_workers, len(locations_in)) or 1 with ThreadPoolExecutor(max_workers=workers) as pool: # W20E5 Codex B1 — propagate parent-thread image-call budget. _submit_process = bind_current_budget(_process) futures = { pool.submit(_submit_process, lid, data): lid for lid, data in locations_in.items() if isinstance(data, dict) } for fut in as_completed(futures): lid, result = fut.result() locations_out[lid] = result total_failed_nodes += result.get("failed_node_count", 0) logger.info( "background_chain_render: done — %d locations, %d failed nodes total", len(locations_out), total_failed_nodes, ) return {"locations": locations_out, "_failed_count": total_failed_nodes} # ────────────────────────────────────────────── # Phase 5 — planner-driven render path # ────────────────────────────────────────────── def _find_primary_loc_for_chain_bg( loc_id: str, fp_specs: Dict[str, Dict[str, Any]], ) -> Optional[str]: """building_group reverse lookup. loc_id가 어떤 floor_plan의 location_ids에 포함되면 그 fp의 primary_location_id를 반환. 단독 location도 자체 floor_plan(primary=loc_id)에 매핑되므로 동일 로직으로 처리된다. fp_specs 미제공/매칭 실패 시 None. """ if not fp_specs: return None for fp in fp_specs.values(): if not isinstance(fp, dict): continue ids = fp.get("location_ids") or [] if loc_id in ids: primary = fp.get("primary_location_id") or "" return primary or None return None def _classify_planner_ref_used( floor_plan_used: bool, parent_used: bool, location_used: bool, ) -> str: """Phase 5 ref_used 라벨 분류 — 디버그 traceability.""" if not (floor_plan_used or parent_used or location_used): return "text_only" if floor_plan_used and parent_used: return "floor_plan+parent_group" if floor_plan_used and location_used: return "floor_plan+location" if floor_plan_used: return "floor_plan_only" if parent_used: return "parent_group" return "location" def _build_synthetic_node_for_group( group_id: str, group_result: Dict[str, Any], ) -> Dict[str, Any]: """T6 group result에서 LLM 호출용 synthetic node spec 빌드. T6의 _run_planner_driven은 그룹당 LLM 1회 호출로 nodes[]를 산출한다 (보통 1개 root anchor). T7은 그룹당 PNG 1개를 생성하므로, 그룹의 root node를 선택하여 generate_node_prompt에 그대로 전달한다. nodes가 비었거나 root가 없으면 group_id 자체를 임시 anchor_root로 합성한다. """ nodes: List[Dict[str, Any]] = list(group_result.get("nodes") or []) # 우선순위: kind=anchor_root 노드 → execution_order 첫 entry → nodes[0] chosen: Optional[Dict[str, Any]] = None for n in nodes: if n.get("kind") == "anchor_root": chosen = n break if chosen is None: order = group_result.get("execution_order") or [] if order: for n in nodes: if n.get("id") == order[0]: chosen = n break if chosen is None and nodes: chosen = nodes[0] if chosen is not None: # group 차원의 shot_ids로 보강 — synthetic node는 group 전체를 대표 synth = dict(chosen) # parent_id는 GROUP 차원 부모(다른 group)와 별개로 — 자체는 group root처럼 다룸 synth["parent_id"] = "" # shot_ids = 그룹 안 모든 노드의 shot 합집합 all_shots: List[str] = [] seen: set = set() for n in nodes: for sid in n.get("shot_ids", []) or []: if sid in seen: continue seen.add(sid) all_shots.append(sid) if all_shots: synth["shot_ids"] = all_shots return synth # fallback — minimal anchor_root 합성 return { "id": group_id, "kind": "anchor_root", "label": group_result.get("location_name") or group_id, "description": group_result.get("rationale_summary", "") or "", "shot_ids": [], "parent_id": "", "depth": 0, "rationale": group_result.get("rationale_summary", "") or "", "shared_visual_anchors_with_parent": [], } def _compute_chain_bg_levels( planner_chain_order: List[str], planning_groups: Dict[str, Dict[str, Any]], renderable_set: set, ) -> List[List[str]]: """planner_chain_order를 level-based parallelizable batches로 분할. Phase 5.3 wrapper — Phase 7에서 추출한 generic compute_dag_levels로 위임. 동일한 시그니처/동작 유지 (회귀 0). """ return compute_dag_levels( order=planner_chain_order, items=planning_groups, renderable_set=renderable_set, parent_field="parent_id", ) def _render_one_group_planner( *, group_id: str, group_result: Dict[str, Any], variant_index: int, variant_label: str, rendered_paths_snapshot: Dict[str, Path], floor_plan_specs: Dict[str, Dict[str, Any]], floor_plan_paths: Dict[str, Path], location_ref_paths: Dict[str, Path], image_dir: Path, openai_client: Any, image_model: str, sanitizer: Optional[PromptSanitizer], size: str, quality: str, max_attempts: int, opik_metadata: Optional[Dict[str, Any]], shot_meta_by_id: Optional[Dict[str, Dict[str, Any]]], ) -> Dict[str, Any]: """단일 renderable group 1개에 대한 prompt 생성 + 이미지 render. Pure function. 호출자가 미리 status=='ok' + loc_id 비어있지 않음을 확인한 group만 들어옴. rendered_paths_snapshot은 이번 level 시작 시점의 부모 PNG dict (read-only). 공유 상태(rendered_paths/results/total_failed)는 호출자가 결과 수신 후 갱신. Returns: 결과 dict (status: ok | prompt_failed | rejected_path | failed) """ loc_id = group_result.get("location_id", "") parent_group_id = (group_result.get("parent_id") or "").strip() location_name = group_result.get("location_name", "") # ref 우선순위 결정 ref_paths: List[Path] = [] floor_plan_used = False parent_used = False location_used = False # 1순위: building_group reverse lookup으로 floor_plan PNG primary_for_group = _find_primary_loc_for_chain_bg(loc_id, floor_plan_specs) if primary_for_group and primary_for_group in floor_plan_paths: fp_path = floor_plan_paths[primary_for_group] if fp_path is not None and fp_path.exists(): ref_paths.append(fp_path) floor_plan_used = True # fallback: floor_plan_paths의 직접 매핑 (loc_id == primary) if not floor_plan_used and loc_id in floor_plan_paths: fp_path = floor_plan_paths[loc_id] if fp_path is not None and fp_path.exists(): ref_paths.append(fp_path) floor_plan_used = True # 2순위: parent group의 chain_bg PNG (이번 level 시작 시점 snapshot에서만 조회) if parent_group_id and parent_group_id in rendered_paths_snapshot: parent_png = rendered_paths_snapshot[parent_group_id] if parent_png.exists(): ref_paths.append(parent_png) parent_used = True # 3순위: location ref (1+2 모두 없을 때만) if not floor_plan_used and not parent_used: loc_ref = location_ref_paths.get(loc_id) if loc_ref is not None and loc_ref.exists(): ref_paths.append(loc_ref) location_used = True ref_used = _classify_planner_ref_used(floor_plan_used, parent_used, location_used) # synthetic node + shots context synth_node = _build_synthetic_node_for_group(group_id, group_result) shots_in_node: List[Dict[str, Any]] = [] for sid in synth_node.get("shot_ids", []) or []: if shot_meta_by_id and sid in shot_meta_by_id: meta = dict(shot_meta_by_id[sid]) meta.setdefault("shot_id", sid) shots_in_node.append(meta) else: shots_in_node.append({"shot_id": sid, "description": ""}) # parent block (LLM 컨텍스트용) — parent PNG는 ref로만 사용. _build_synthetic_node_for_group이 # parent_id=""로 설정해 root anchor 형태로 제출. parent_for_llm = None location_description = ( group_result.get("rationale_summary") or location_name or "" ) # Phase 1: t2i prompt + shot_guides try: t2i_prompt, shot_guides = generate_node_prompt( location_id=loc_id, location_description=location_description, node=synth_node, parent=parent_for_llm, shots_in_node=shots_in_node, opik_metadata=opik_metadata, ) except Exception as exc: logger.error( "background_chain_render (planner): group %s prompt gen failed: %s", group_id, exc, ) return { "group_id": group_id, "location_id": loc_id, "location_name": location_name, "status": "prompt_failed", "render_error": str(exc)[:200], "variant_label": variant_label, "variant_index": variant_index, "ref_used": ref_used, "png_path": "", "t2i_prompt": "", "shot_guides": [], "parent_id": parent_group_id, } # PNG 파일명 — {loc_id}_{variant_label}.png png_path = image_dir / f"{loc_id}_{variant_label}.png" # path traversal guard try: png_path.resolve().relative_to(image_dir.resolve()) except (ValueError, OSError) as exc: logger.error( "background_chain_render (planner): group %s out_path escapes image_dir: %s — skip", group_id, exc, ) return { "group_id": group_id, "location_id": loc_id, "location_name": location_name, "status": "rejected_path", "render_error": "out_path escapes image_dir", "variant_label": variant_label, "variant_index": variant_index, "ref_used": ref_used, "png_path": "", "t2i_prompt": t2i_prompt, "shot_guides": shot_guides, "parent_id": parent_group_id, } # Phase 2: image render info = render_node_image( openai_client=openai_client, image_model=image_model, prompt=t2i_prompt, out_path=png_path, ref_paths=ref_paths, sanitizer=sanitizer, size=size, quality=quality, max_attempts=max_attempts, ) info["ref_used"] = ref_used if info.get("status") == "ok" and png_path.exists(): return { "group_id": group_id, "location_id": loc_id, "location_name": location_name, "status": "ok", "variant_label": variant_label, "variant_index": variant_index, "ref_used": ref_used, "png_path": str(png_path), "t2i_prompt": t2i_prompt, "shot_guides": shot_guides, "parent_id": parent_group_id, "render_attempts": info.get("attempts", 0), "sanitize_strategies": info.get("strategies", []), "scenes": list(group_result.get("scenes") or []), "shot_ids": list(synth_node.get("shot_ids") or []), "time": group_result.get("time", "") or "", "floor_plan_used": floor_plan_used, } return { "group_id": group_id, "location_id": loc_id, "location_name": location_name, "status": "failed", "render_error": info.get("final_block_reason") or "", "variant_label": variant_label, "variant_index": variant_index, "ref_used": ref_used, "png_path": "", "t2i_prompt": t2i_prompt, "shot_guides": shot_guides, "parent_id": parent_group_id, "render_attempts": info.get("attempts", 0), "sanitize_strategies": info.get("strategies", []), } def _resolve_workers(default: int = 4, cap: int = 8) -> int: """BACKGROUND_CHAIN_RENDER_WORKERS env → clamp to [1, cap]. 잘못된 값은 default.""" raw = os.getenv("BACKGROUND_CHAIN_RENDER_WORKERS", "").strip() if not raw: return default try: v = int(raw) except ValueError: logger.warning( "BACKGROUND_CHAIN_RENDER_WORKERS=%r not int — using default %d", raw, default, ) return default if v < 1: return 1 if v > cap: return cap return v def _run_planner_driven_render( *, planner_chain_order: List[str], planning_groups: Dict[str, Dict[str, Any]], floor_plan_specs: Dict[str, Dict[str, Any]], floor_plan_paths: Dict[str, Path], image_dir: Path, location_ref_paths: Dict[str, Path], openai_client: Any, image_model: str, sanitizer: Optional[PromptSanitizer], size: str, quality: str, max_attempts: int, opik_metadata: Optional[Dict[str, Any]], shot_meta_by_id: Optional[Dict[str, Dict[str, Any]]], ) -> Dict[str, Any]: """Phase 5 planner-driven render path (Phase 5.3 level-based parallel). Phase 5: chain_bg_order대로 SEQUENTIAL 처리 (1 group/turn). Phase 5.3: parent_id로 DAG level batch 구성 → 같은 level 내 ThreadPool 병렬. 각 group은 여전히 별도 LLM call (HARD constraint per CLAUDE.md). variant_index/ variant_label은 chain_bg_order대로 deterministic 사전 할당 — 병렬화에도 같은 결과. ref 우선순위 (변경 없음): 1. building_group reverse lookup으로 찾은 floor_plan PNG 2. group의 parent_id에 해당하는 chain_bg PNG (rendered_paths) 3. location ref workers: BACKGROUND_CHAIN_RENDER_WORKERS env (default 4, capped at 8). Returns: { "data": {"groups": {group_id: result}}, "groups": {group_id: result}, "_failed_count": int, } """ if not planner_chain_order: logger.info("background_chain_render (planner-driven): empty chain_bg_order") return { "data": {"groups": {}}, "groups": {}, "_failed_count": 0, } image_dir.mkdir(parents=True, exist_ok=True) # ── Pre-pass: skip + variant_index 사전 할당 (deterministic, sequential) ── pre_results: Dict[str, Dict[str, Any]] = {} variants: Dict[str, Tuple[int, str]] = {} # group_id → (variant_index, variant_label) location_variant_counter: Dict[str, int] = {} for group_id in planner_chain_order: group_result = planning_groups.get(group_id) if not group_result: logger.warning( "background_chain_render (planner): group_id %s not in planning groups — skip", group_id, ) continue status = group_result.get("status", "") if status != "ok": logger.info( "background_chain_render (planner): group %s status=%s — render skipped", group_id, status, ) pre_results[group_id] = { "group_id": group_id, "location_id": group_result.get("location_id", ""), "location_name": group_result.get("location_name", ""), "status": "skipped_planning", "skipped_reason": f"planning status={status}", "variant_label": "", "ref_used": "none", "png_path": "", "t2i_prompt": "", "shot_guides": [], "parent_id": group_result.get("parent_id", "") or "", } continue loc_id = group_result.get("location_id", "") or "" if not loc_id: logger.warning( "background_chain_render (planner): group %s has empty location_id — skip", group_id, ) # Claude review I2: 비어있는 loc_id도 results에 명시적 기록 — 호출자가 # planner_chain_order의 모든 group_id를 results에서 찾을 수 있게. pre_results[group_id] = { "group_id": group_id, "location_id": "", "location_name": group_result.get("location_name", ""), "status": "skipped_no_loc", "skipped_reason": "empty location_id", "variant_label": "", "ref_used": "none", "png_path": "", "t2i_prompt": "", "shot_guides": [], "parent_id": group_result.get("parent_id", "") or "", } continue next_idx = location_variant_counter.get(loc_id, 1) variants[group_id] = (next_idx, f"v{next_idx:02d}") location_variant_counter[loc_id] = next_idx + 1 renderable_set = set(variants.keys()) # ── Level batch 계산 ── levels = _compute_chain_bg_levels(planner_chain_order, planning_groups, renderable_set) # ── Per-level ThreadPool 실행 ── results: Dict[str, Dict[str, Any]] = dict(pre_results) rendered_paths: Dict[str, Path] = {} total_failed = 0 workers = _resolve_workers() failure_statuses = {"failed", "prompt_failed", "rejected_path"} if levels: logger.info( "background_chain_render (planner-driven): %d renderable groups across %d levels (workers=%d)", len(renderable_set), len(levels), workers, ) for level_idx, level in enumerate(levels): if not level: continue # parent ref는 이번 level 시작 시점의 snapshot만 사용 (thread safe) rendered_snapshot = dict(rendered_paths) level_workers = max(1, min(workers, len(level))) with ThreadPoolExecutor(max_workers=level_workers) as pool: # W20E5 Codex B1 — propagate parent-thread image-call budget # into pool workers (each group render triggers gpt-image-2). _submit_planner_render = bind_current_budget(_render_one_group_planner) futures = { pool.submit( _submit_planner_render, group_id=gid, group_result=planning_groups[gid], variant_index=variants[gid][0], variant_label=variants[gid][1], rendered_paths_snapshot=rendered_snapshot, floor_plan_specs=floor_plan_specs, floor_plan_paths=floor_plan_paths, location_ref_paths=location_ref_paths, image_dir=image_dir, openai_client=openai_client, image_model=image_model, sanitizer=sanitizer, size=size, quality=quality, max_attempts=max_attempts, opik_metadata=opik_metadata, shot_meta_by_id=shot_meta_by_id, ): gid for gid in level } for future in as_completed(futures): gid = futures[future] # Claude review I1: future.result() 무방어 시 worker exception이 그대로 # 전파되어 같은 level의 나머지 future가 누락된다. 보호 wrap으로 # group을 'failed' 상태로 기록하고 처리 지속. try: res = future.result() except Exception as exc: logger.error( "background_chain_render (planner): group %s render raised: %s", gid, exc, ) gr_meta = planning_groups.get(gid) or {} var = variants.get(gid, (0, "")) res = { "group_id": gid, "location_id": gr_meta.get("location_id", ""), "location_name": gr_meta.get("location_name", ""), "status": "failed", "render_error": str(exc)[:200], "variant_label": var[1], "variant_index": var[0], "ref_used": "none", "png_path": "", "t2i_prompt": "", "shot_guides": [], "parent_id": (gr_meta.get("parent_id") or "").strip(), } results[gid] = res res_status = res.get("status", "") if res_status == "ok": rendered_paths[gid] = Path(res["png_path"]) elif res_status in failure_statuses: total_failed += 1 logger.info( "background_chain_render (planner): level %d/%d done (%d groups, workers=%d)", level_idx + 1, len(levels), len(level), level_workers, ) logger.info( "background_chain_render (planner-driven): done — %d groups (%d failed)", len(results), total_failed, ) # Codex review L1: as_completed는 completion order를 반환하므로 같은 level 내 # group 삽입 순서가 run마다 다르다. data.groups는 dict insertion order를 유지하니 # planner_chain_order대로 재구성하여 downstream consumer (scene_context_loader의 # first-wins 매칭 등)가 deterministic 결과를 받도록 보장한다. ordered_results: Dict[str, Dict[str, Any]] = { gid: results[gid] for gid in planner_chain_order if gid in results } return { "data": {"groups": ordered_results}, "groups": ordered_results, "_failed_count": total_failed, }