"""BackgroundRenderStep — Phase 7 Step 6. 각 master plan background를 gpt-image-2로 PNG 생성. ``depends_on_bg`` DAG → ``compute_dag_levels`` 로 level 분할 후 level 내부는 ThreadPool 병렬. 출력 shape는 Phase 5 ``chain_bg_render`` 와 호환: ``data.groups[bg_id].{status, location_id, png_path, t2i_prompt, shot_guides, shot_ids, parent_id, ref_used, ...}``. 이렇게 해야 Phase 6 ``scene_context_loader._load_chain_bg_guide_by_shot`` 가 변경 없이 동작. Phase 7 ImageAsset UPSERT (Phase 5 chain_bg 패턴 미러): - asset_type='chain_bg' - entity_id = loc_id의 EntityCanon.id - variant_index = loc_id 단위 1+ counter (sorted bg_id, gen_order). v00은 floor_plan(primary)에 예약. - variant_label = state_label - variant_type = bg_id (UPSERT 매치 키, 재실행 idempotent 보장) - is_primary = 0 (chain_bg는 항상 0; v00 floor_plan만 1) - t2i_guide = shot_guides[]를 newline join (Phase 6 scene_detail consumer 가 이를 prepend하여 chain_bg PNG 가구 위치 가이드를 scene t2i에 주입) - file_path = projects_root 기준 relative """ from __future__ import annotations import json import logging import re import traceback 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.bg_state_vocab import BG_ID_RE 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__) # D6 T5c + T5-fix B2: schema_version 2 → 3 — `data.consumed_bg_catalog_hash` + # `data.consumed_shot_binding_hash` field 추가. 기존 v2 cp 는 sibling field 부재 → # step_runner._check_cp_mismatch 가 BLOCK (background_render 는 `_LEGACY_SCHEMA_BUMP_ # ALLOWLIST` 외, `step_runner.py:763-769` 참조). 운영자가 명시 force 의무. SCHEMA_VERSION = 3 # 3 (2026-07-23 슬라이스 E 육안): shot_aware_bg_render_adapter # _STYLE_CONTRACT_PREAMBLE 개정 — 실내 플레이트 bare-wall sterile # 실측 대응. Codex 재리뷰 BLOCKING 반영: 선두 유형 중립화+거주 서술 # 공급 입력 조건부+확정 사실이 계약 전체 우선(타임스탬프 갱신). # 프롬프트 텍스트 변경이라 기존 CP 무효화 의도. PROMPT_VERSION = "3.202607231950" # Legacy bg_id whitelist (pre-D6): lowercase ascii + digits + underscore. # transitional compat — D6 master_plan cp 가 sibling field (background_catalog / # bg_catalog_hash) 를 안 가지면 본 정규식 사용. _LEGACY_SAFE_BG_RE = re.compile(r"^[a-z0-9][a-z0-9_]*$") def _select_bg_id_filter(plans_cp: Optional[Dict[str, Any]]): """D6 marker-gated bg_id filter. plans_cp 의 data 에 D6 sibling marker (`background_catalog` 또는 `bg_catalog_hash`) 가 있으면 D6 strict (`BG_ID_RE` = `^L\\d{2,3}B\\d{2,3}$`) 적용. 없으면 legacy `_LEGACY_SAFE_BG_RE` (transitional compat) — 옛 cp 를 silent reject 안 함. review iter4 note: T-pre-1 단독 commit 시 background_master_plan 가 여전히 legacy `bg_*` 출력 → BG_ID_RE 강제 시 모든 bg silent skip → empty completed silent no-op. marker-gating 으로 D6 atomic transition 안전. """ data = (plans_cp or {}).get("data", {}) or {} if "background_catalog" in data or "bg_catalog_hash" in data: return BG_ID_RE return _LEGACY_SAFE_BG_RE def _unmanned_audit_fields(info: Dict[str, Any]) -> Dict[str, Any]: """무인 계약 감사 3필드 CP 영속 subset (E2E10 실측 fix — lazy import).""" from app.modules.pipeline.background_unmanned import audit_fields return audit_fields(info) def _build_attached_reference_lineage( *, fp_id: str, fp_path: Optional[Path], prior_bg_paths: List[Path], ref_used: str, building_fp_id: str = "", building_fp_path: Optional[Path] = None, building_anchor_bg_id: str = "", building_anchor_path: Optional[Path] = None, same_loc_anchor_bg_id: str = "", same_loc_anchor_path: Optional[Path] = None, aerial_loc_id: str = "", aerial_path: Optional[Path] = None, ) -> Dict[str, Any]: """W21B-wave-1 — checkpoint-only attached reference lineage diagnostic. Codex 결정 2(c): ImageAsset.reference_image_ids / llm_call_log columns 는 이번 wave 에서 건드리지 않는다. attached_reference_lineage 는 step 의 groups_out entry / checkpoint manifest 에만 기록되어 chain_bg / scene visual review 시점에 "이 BG 가 실제로 어떤 reference 를 첨부해서 생성됐는지" 를 역추적할 수 있게 한다. attached_ref_labels 는 실제로 첨부된 ref 만 포함한다 — fp_path 가 존재하면 ``fp:``, 첨부된 prior_bg PNG 마다 그 stem 을 ``bg:`` 로 기록. PNG path 의 stem 이 bg_id 와 일치하는 것은 step 자체가 그렇게 저장하기 때문 (``image_dir / f"{bid}.png"``). prior_bg_paths 가 비어 있고 fp 가 없으면 attached_ref_labels 는 빈 list — ref_used 가 ``text_only`` 인 것과 consistent. """ attached_ref_labels: List[str] = [] prior_bg_ids: List[str] = [] # W-L: 야외 loc 의 aerial establishing 은 ref **최우선(첫 번째)** 계약 # (render_one_background) — 라벨도 첫 번째. prior_bg_ids 에는 넣지 않는다 # (chain lineage 채널과 분리 — DB resolve 는 aerial_ref 구조 필드 담당). if aerial_path is not None and aerial_path.exists(): attached_ref_labels.append(f"aerial:{aerial_loc_id}") if fp_path is not None and fp_path.exists(): attached_ref_labels.append(f"fp:{fp_id}") for p in prior_bg_paths: if p is None: continue if p.exists(): bg_id = p.stem prior_bg_ids.append(bg_id) attached_ref_labels.append(f"bg:{bg_id}") # W-K: 같은 location anchor 렌더는 prior 다음·building anchor 앞 계약 # (render_one_background) — 라벨 순서 동일. prior_bg_ids 에는 넣지 않는다 # (chain lineage 채널과 분리 — DB resolve 는 same_loc_anchor_ref 구조 필드 담당). if same_loc_anchor_path is not None and same_loc_anchor_path.exists(): attached_ref_labels.append(f"same_loc_anchor:{same_loc_anchor_bg_id}") # W-I: 같은 building 그룹 anchor 렌더는 그 다음·building fp 앞 계약 # (render_one_background) — 라벨 순서 동일. prior_bg_ids 에는 넣지 않는다 # (chain lineage 채널과 분리 — DB resolve 는 building_anchor_ref 구조 필드 담당). if building_anchor_path is not None and building_anchor_path.exists(): attached_ref_labels.append(f"building_anchor:{building_anchor_bg_id}") # W-G: same-building indoor fp 는 ref 리스트 마지막에 첨부되는 계약 # (render_one_background) — 라벨도 마지막. prior_bg_ids 에는 넣지 않는다 # (bg lineage 채널 오염 방지 — DB resolve 는 building_fp_ref 구조 필드가 담당). if building_fp_path is not None and building_fp_path.exists(): attached_ref_labels.append(f"building_fp:{building_fp_id}") return { "fp_id": fp_id, "prior_bg_ids": prior_bg_ids, "attached_ref_labels": attached_ref_labels, "ref_used": ref_used, "db_write_policy": "checkpoint_only", } def _inject_star_anchors( bg_specs: Dict[str, Dict[str, Any]], items: Dict[str, Dict[str, Any]], order: List[str], renderable: set, ) -> Dict[str, str]: """3a fix (2026-07-01, Codex 합의) — depth-1 star anchor 주입 (flag-gated caller). 같은 ``loc_id`` 그룹의 anchor(bg_id 자연정렬 첫 = B01 우선)를, 그룹 내 나머지 bg 가 ``depends_on_bg`` 로 참조하도록 결정론 주입한다. render loop 이 이 필드로 prior_bg_paths(i2i) + DAG parent 를 구성하므로 생성/lineage 가 함께 흐른다. ★가드(Codex): (1) 그룹 renderable bg 2개 미만이면 skip. (2) anchor 자신은 depends_on_bg 불변. (3) 기존 depends_on_bg 가 있으면 preserve — **empty 일 때만** inject(순환/기존 체인 파괴 방지). (4) loc_id 없는 bg 는 그룹 대상 아님. 반환: ``{bid: anchor_bid}`` (실제 주입된 것만). 호출측이 flag ON 일 때만 부른다. """ by_loc: Dict[str, List[str]] = {} for bid in order: if bid not in renderable: continue loc = str((bg_specs.get(bid) or {}).get("loc_id") or "").strip() if not loc: continue by_loc.setdefault(loc, []).append(bid) injected: Dict[str, str] = {} for loc, bids in by_loc.items(): if len(bids) < 2: continue anchor = sorted(bids)[0] # 자연정렬 첫(zero-padded B01 우선), 결정론. for bid in bids: if bid == anchor: continue spec = bg_specs.get(bid) or {} if spec.get("depends_on_bg"): # 기존 체인 preserve — empty 일 때만. continue spec["depends_on_bg"] = [anchor] spec["_star_anchor_meta"] = { "anchor_injected": True, "anchor_policy": "location_star_depth1", "anchor_bg_id": anchor, "anchor_reason": "sub_location_missing_or_no_chain", } if bid in items: items[bid]["parent_id"] = anchor injected[bid] = anchor return injected def _compose_bg_input_image_ids( prior_bg_ids: List[str], fp_uuid: Optional[str], uuid_by_bgid: Dict[str, Optional[str]], *, reuse: bool, ) -> Tuple[List[str], Dict[str, Any]]: """3a fix (2026-07-01, 워크플로 스펙) — 실제 첨부 prior_bg_ids 를 background_render ImageAsset.input_image_ids([fp_uuid, *prior_uuids]) + lineage 메타로 조립 (순수). prior_bg_ids = groups[bg_id].attached_reference_lineage.prior_bg_ids (실제 i2i/alias 첨부된 BG 의 **prefix 없는 bg_id 리스트**, FP 제외 — 라벨/substring 파싱 0). uuid_by_ bgid = bg_id→ImageAsset UUID resolver(배치맵 + DB variant_type fallback 합성). 미해결 bg_id 는 unresolved_inputs 구조키("bg:")로 기록(입력없음 [] 과 구분). reuse= 구조 필드 ref_used=="reused_plate"(copy-less alias — i2i 아님)면 lineage_kind 태깅. 반환: (input_image_ids, pipeline_metadata). fp_uuid None 이면 prior 만. """ prior_uuids: List[str] = [] unresolved: List[str] = [] _seen_bg: set = set() for pb in prior_bg_ids: if pb in _seen_bg: # dedup(stable order 보존) — 방어적(phase1 도 dedup). continue _seen_bg.add(pb) u = uuid_by_bgid.get(pb) if u: prior_uuids.append(u) else: unresolved.append("bg:" + pb) # fp 를 맨 앞, 그다음 prior. fp_uuid 가 prior 에도 있으면(비정상) 중복 제거. iids: List[str] = [] for x in ([fp_uuid, *prior_uuids]): if x and x not in iids: iids.append(x) meta: Dict[str, Any] = { "bg_lineage_policy": "attached_prior_bg", "prior_bg_ids": list(prior_bg_ids), "lineage_kind": "reuse_alias" if reuse else "i2i_reference", } if unresolved: meta["unresolved_inputs"] = unresolved return iids, meta def _outdoor_loc_ids_from_classify( classify_cp: Optional[Dict[str, Any]], ) -> set: """W22 W4b — background_classify cp 에서 실외 loc_id 집합 추출. is_indoor 판정의 유일 원천 = classify cp (설계 §1). cp 부재/빈 데이터 = 빈 집합 → skip 미발동 (보수적). """ out: set = set() groups = ( (classify_cp or {}).get("data", {}) or {} ).get("building_groups", []) or [] for g in groups: for m in (g.get("members") or []): lid = m.get("loc_id") or "" if lid and m.get("is_indoor") is False: out.add(lid) return out def _split_outdoor_direct_skip( renderable: set, bg_specs: Dict[str, Dict[str, Any]], outdoor_locs: set, ) -> Tuple[set, Dict[str, Dict[str, Any]]]: """W22 W4b — renderable 에서 야외 loc bg 를 분리 (순수 함수, 테스트 잠금). Returns (남은 renderable, {bg_id: skipped 마커 entry}). """ skipped: Dict[str, Dict[str, Any]] = {} for bid in sorted(renderable): spec = bg_specs.get(bid) or {} if (spec.get("loc_id") or "") in outdoor_locs: skipped[bid] = { "status": "skipped_outdoor_direct", "location_id": spec.get("loc_id", ""), "location_name": spec.get("sub_location", ""), "shot_ids": spec.get("applies_to_shots", []) or [], } return renderable - set(skipped), skipped def _resolve_openai_client(): """OpenAI 이미지 클라이언트 resolve. T11 FloorPlanRenderStep와 동일 패턴. api_key 는 `openai_keys` 브로커가 정한다(2슬롯 failover, 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 회귀 가드. """ from app.core.openai_keys import openai_client from app.core.config import settings return openai_client( timeout=float(settings.llm_timeout_image_gen), ) class BackgroundRenderStep(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, } # E2E11 ②: STRUCTURE FACTS+VIEW AUTHORITY prefix — ON 이면 shot_aware # 플레이트 프롬프트 실질 변경(재렌더 유도). OFF=기존 hash 불변. if bool(getattr(settings, "plate_structure_facts_enabled", False)): payload["plate_structure_facts"] = 1 # s40 레시피 plate multiroll (2026-07-13, Codex 1차 리뷰 HIGH-5) — # ON 이면 렌더 엔진·롤·게이트가 출력 실질 변경. OFF byte-identical. if bool(getattr(settings, "plate_multiroll_enabled", False)): payload["plate_multiroll_enabled"] = True payload["plate_multiroll_roll_count"] = ( settings.still_recipe_roll_count ) payload["plate_multiroll_critique"] = ( settings.still_recipe_critique_enabled ) payload["plate_multiroll_nb2_model"] = settings.gemini_image_model # E2E6 ⑥: 하드코딩 스탬프 → 실사용 selector 해석값으로 동기 from app.modules.pipeline.multiroll_gemini import ( resolve_judge_pack_version, ) payload["plate_multiroll_judge_pack"] = ( resolve_judge_pack_version() ) # ★배경판 **초기 선정**의 판정 쌍·순서 정책이 이 outer hash 에 # 통째로 없었다 (2026-08-29 Codex BLOCK-3). `plate_multiroll.py` # 안쪽 지문에는 있는데, outer 가 clean 이면 스텝이 **내려가지도 # 않아** 안쪽 지문은 읽히지 않는다. 배경판도 `make_gemini_judge_fn` # 을 쓰므로 선정 심판이 바뀌면 어느 판이 뽑히는지가 바뀐다. from app.modules.pipeline.multiroll_gemini import ( CROSS_MODEL_ORDER_POLICY_VERSION as _plate_order_policy, resolve_select_judge_model_physical as _plate_sel_phys, ) payload["plate_select_judge_models_physical"] = _plate_sel_phys() payload["plate_select_order_policy"] = _plate_order_policy # E2E10 fix② Codex BLOCKING-2: i2i 수정본 재판정은 플레이트 # 최종 _sel 결정 정책 전환 — 완료 CP skip 을 뚫도록 상위 # hash 에 스탬프 (critique OFF 면 재판정 자체가 없어 무스탬프, # flag OFF byte-identical). if settings.still_recipe_critique_enabled and bool( getattr(settings, "multiroll_fix_rejudge_enabled", False) ): from app.modules.pipeline.multiroll_select import ( FIX_REJUDGE_POLICY_VERSION as _frj_policy, ) payload["plate_multiroll_fix_rejudge"] = _frj_policy # ★재판정도 같은 팩토리 — 두 심판이 다 돈다. from app.modules.pipeline.multiroll_gemini import ( resolve_select_judge_model_physical as _sel_phys2, ) payload["plate_fix_rejudge_judge_model_physical"] = ( _sel_phys2()) # Codex HIGH-4: 중립 헤더 팩(판정 계약)도 실질 입력 from app.modules.pipeline.multiroll_gemini import ( FIX_REJUDGE_HEADER_PACK_VERSION as _frj_hdr_sel, ) payload["plate_fix_rejudge_header_pack"] = ( resolve_judge_pack_version(_frj_hdr_sel) ) # E2E11 fix③: GPT 구도 critique — 플레이트 최종본 실질 입력 if settings.still_recipe_critique_enabled and bool( getattr(settings, "multiroll_gpt_composition_enabled", False) ): from app.modules.pipeline.multiroll_gemini import ( GPT_COMPOSITION_PACK_VERSION as _gc_sel, ) from app.modules.pipeline.multiroll_select import ( GPT_COMPOSITION_POLICY_VERSION as _gc_policy, ) payload["plate_gpt_composition"] = _gc_policy payload["plate_gpt_composition_pack"] = ( resolve_judge_pack_version(_gc_sel) ) # Codex HIGH-2: 물리 모델 교체가 완료 CP skip 을 뚫도록 from app.modules.pipeline.multiroll_gemini import ( GPT_COMPOSITION_MODEL as _gc_model, ) payload["plate_gpt_composition_model"] = _gc_model payload["plate_gpt_composition_model_physical"] = str( getattr(settings, "openai_model", "") ) # W22 W4b (2026-07-10) — 직행 모드는 야외 bg plate 렌더를 skip 하므로 # 출력이 실질 변경 → flag=True 일 때만 스탬프 (OFF byte-identical). if bool(getattr(settings, "outdoor_direct_compose_enabled", False)): payload["outdoor_direct_compose_enabled"] = True # 이식 ① (2026-07-20) — 무인 계약 절은 전 배경 렌더 프롬프트를 # 실질 변경 → ON 시만 팩 스탬프 (OFF byte-identical). 게이트는 # 판정·재렌더로 산출을 바꿀 수 있어 별도 스탬프. if bool(getattr(settings, "background_no_people_enabled", False)): from app.modules.pipeline.background_unmanned import ( resolve_prompt_version as _unmanned_pack, ) payload["background_no_people_pack"] = _unmanned_pack("1") if bool(getattr( settings, "background_no_people_gate_enabled", False, )): from app.modules.pipeline.multiroll_gemini import ( JUDGE_MODEL as _unmanned_judge_model, ) payload["background_no_people_gate"] = True payload["background_no_people_gate_judge_model"] = ( _unmanned_judge_model ) # Codex 리뷰 5: alias(gemini-pro)는 물리 모델 교체를 감지 # 못함(llm_client 가 settings.gemini_text_model 로 해석) — # 물리 모델 병행 스탬프 payload[ "background_no_people_gate_judge_model_physical" ] = str(settings.gemini_text_model) # W19B-3 BLOCKING 2 lock: legacy selector path keeps the payload # byte-identical to today (the new selector key is NOT added). Only # the opt-in paths stamp ``background_render_reference_mode`` so # legacy v6/v7 checkpoints stay valid; only opt-in projects # invalidate their existing hash. W20C adds "shot_aware_plan" as a # second opt-in value with the same stamping policy. if settings.background_render_reference_mode in ( "w18j_overlap", "shot_aware_plan", ): payload[ "background_render_reference_mode" ] = settings.background_render_reference_mode # TASK3-B: the missing-plan direct-plate fallback materially changes # background_render output for a dropped fp (degrade-render vs # fail-closed), so its policy must invalidate the checkpoint. Stamped # ONLY in the shot_aware_plan path (the sole mode where the gate # runs) — legacy/default stays byte-identical. if settings.background_render_reference_mode == "shot_aware_plan": payload[ "background_render_missing_shot_aware_plan_fallback_enabled" ] = bool( getattr( settings, "background_render_missing_shot_aware_plan_fallback_enabled", False, ) ) # W21B-w4 #4(C) — the substrate consumer materially changes a fresh # plate's render input (FP + ref_tree_parents vs the adapter's either-or # refs), so flipping it must invalidate existing background_render # checkpoints. Stamped ONLY when the shot_aware_plan opt-in path is active # AND the flag is True — so legacy/default and shot_aware+flag-False keep # their existing hash byte-identical; only flag=True (or True→False) # invalidates. if ( settings.background_render_reference_mode == "shot_aware_plan" and bool(settings.bg_render_substrate_enabled) ): payload["bg_render_substrate_enabled"] = True # W-G (2026-07-03) — same-building indoor fp ref 는 outdoor plate 의 렌더 # 입력(ref 리스트+프롬프트)을 실질 변경하므로 flag flip 이 checkpoint 를 # invalidate 해야 한다. substrate 정책 미러 — shot_aware_plan 경로에서 # flag=True 일 때만 스탬프(legacy/default·flag-False 는 hash byte-identical). if ( settings.background_render_reference_mode == "shot_aware_plan" and bool(getattr(settings, "outdoor_building_fp_ref_enabled", False)) ): payload["outdoor_building_fp_ref_enabled"] = True # W-I (2026-07-03) — building anchor 렌더 ref 는 같은 그룹 outdoor plate 의 # 렌더 입력(ref 리스트+프롬프트)을 실질 변경하므로 W-G 와 동일 스탬프 정책: # shot_aware_plan 경로 + flag=True 일 때만 접는다(OFF byte-identical). if ( settings.background_render_reference_mode == "shot_aware_plan" and bool(getattr( settings, "outdoor_building_anchor_ref_enabled", False)) ): payload["outdoor_building_anchor_ref_enabled"] = True # W-K (2026-07-03) — same-place 렌더 체이닝은 lane 순서·anchor ref· # 프롬프트 guidance 를 실질 변경하므로 W-G/W-I 와 동일 스탬프 정책: # shot_aware_plan 경로 + flag=True 일 때만 접는다(OFF byte-identical). if ( settings.background_render_reference_mode == "shot_aware_plan" and bool(getattr( settings, "same_place_render_chain_enabled", False)) ): payload["same_place_render_chain_enabled"] = True # W-L (2026-07-03) — 야외 aerial 1순위 ref 교체는 야외 bg 의 렌더 입력 # (ref 리스트+프롬프트)을 실질 변경하므로 W-G/W-I/W-K 와 동일 스탬프 # 정책: shot_aware_plan 경로 + flag=True 일 때만 접는다(OFF # byte-identical). if ( settings.background_render_reference_mode == "shot_aware_plan" and bool(getattr( settings, "outdoor_aerial_reference_enabled", False)) ): payload["outdoor_aerial_reference_enabled"] = True # W-M Codex NARROW: AERIAL_PROMPT_VERSION bump 는 큐 내부의 # aerial 캐시 재사용 판정만 바꾸고 step 재실행 자체는 유발하지 # 못한다 — 완료된 checkpoint 가 있는 resume 에서 config_hash 가 # 그대로면 aerial pre-pass 가 아예 돌지 않아 이전 의미의 # 배치도/plate 가 살아남는다. 버전을 hash 에 접어 프롬프트 의미 # 변경이 dispatch 레벨 stale 로 이어지게 한다(W-L flag ON 일 # 때만 — OFF 는 aerial 미사용이라 hash byte-identical). from app.modules.pipeline.location_aerial import ( AERIAL_PROMPT_VERSION, ) payload["location_aerial_prompt_version"] = AERIAL_PROMPT_VERSION # 2026-07-19 재설계 A: seed 플레이트 참조도 렌더 실질 입력 — ON 시만 # 스탬프(OFF byte-identical). guidance 절 의미도 hash 로 접는다. if bool(getattr( settings, "outdoor_seed_plate_reference_enabled", False)): import hashlib as _hashlib from app.modules.pipeline.outdoor_structure_seed import ( SEED_SITE_GUIDANCE, ) payload["outdoor_seed_plate_reference_enabled"] = True payload["seed_site_guidance_sha"] = _hashlib.sha256( SEED_SITE_GUIDANCE.encode("utf-8") ).hexdigest()[:12] # W-M (2026-07-03) — 야외 plate 참조 단계화(그룹 체인 lane + 실내 # anchor 제거 + aerial∧building fp 동시 첨부 금지)는 렌더 입력을 # 실질 변경하므로 W-G/W-I/W-K/W-L 과 동일 스탬프 정책: shot_aware_ # plan 경로 + flag=True 일 때만 접는다(OFF byte-identical). if ( settings.background_render_reference_mode == "shot_aware_plan" and bool(getattr( settings, "outdoor_plate_stage_chain_enabled", False)) ): payload["outdoor_plate_stage_chain_enabled"] = True # W21B-w5 STEP5-B — the light-FP sidecar swaps the FP anchor PNG fed to # BOTH render paths (legacy ``_process`` + shot_aware substrate), so # flipping it materially changes this step's render input and MUST # invalidate existing background_render checkpoints. Stamped ONLY when the # flag is True (mirrors the substrate policy above) — legacy/default and # flag-False keep their existing hash byte-identical; True (or True→False) # invalidates. if bool(getattr(settings, "floor_plan_light_sidecar_enabled", False)): payload["floor_plan_light_sidecar_enabled"] = True # Flipping the flag invalidates, but a sidecar REGEN (new NB2 model / # selection prompt / best_of_n / re-draw) with the flag still True must # ALSO invalidate this step — else background_render keeps a stale BG # checkpoint that references the old sidecar PNGs (Codex W21B-w5 # review). Fold the consumed sidecar checkpoint's config_hash (which # stamps those NB2 levers) so any sidecar change propagates here. try: sidecar_cp = self._load_prev_checkpoint( "floor_plan_light_sidecar") sidecar_hash = (sidecar_cp or {}).get("config_hash") or "" except Exception: # config_hash must never crash on a cp read sidecar_hash = "" if sidecar_hash: payload["floor_plan_light_sidecar_config_hash"] = sidecar_hash 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( "background_render: %s parse failed: %s", step_id, exc ) return None def _apply_light_fp_sidecar( self, fp_paths_str: Dict[str, str] ) -> Dict[str, Any]: """W21B-w5 STEP5-B: swap the FP anchor PNG to the simplified light sidecar. Mutates ``fp_paths_str`` IN PLACE: for each fp whose floor_plan_light_sidecar entry is ``rendered`` with a resolvable PNG on disk, the detailed FP path is replaced by the light PNG's absolute path (the same key the legacy and substrate render paths both read). Returns a per-fp provenance map for the checkpoint (audit). A missing/invalid sidecar falls back to the detailed FP for that fp only. Flag OFF (default): returns ``{}`` and does not touch ``fp_paths_str`` — the render path is byte-identical to the pre-STEP5-B behaviour. """ from app.core.config import settings if not bool( getattr(settings, "floor_plan_light_sidecar_enabled", False) ): return {} cp = self._load_prev_checkpoint("floor_plan_light_sidecar") per_fp = ((cp or {}).get("data", {}) or {}).get("per_fp", {}) or {} if not per_fp: return {} from app.core.file_paths import resolve_image_path provenance: Dict[str, Any] = {} swapped = 0 for fid in list(fp_paths_str.keys()): entry = per_fp.get(fid) if not isinstance(entry, dict) or entry.get("status") != "rendered": provenance[fid] = { "used": "detailed", "reason": ( entry.get("status") if isinstance(entry, dict) else "absent" ), } continue rel = entry.get("light_png_relative_path") abs_path = resolve_image_path(rel) if rel else None if abs_path is None or not abs_path.exists(): provenance[fid] = { "used": "detailed", "reason": "light_png_missing", } continue fp_paths_str[fid] = str(abs_path) provenance[fid] = {"used": "light", "light_png": rel} swapped += 1 if swapped: logger.info( "background_render: swapped %d FP anchor(s) to light sidecar", swapped, ) return provenance def _plate_multiroll_ctx(self) -> Optional[Dict[str, Any]]: """plate_multiroll(레시피) 컨텍스트 — flag OFF 면 None(기존 무영향). Codex 1차 리뷰 MEDIUM-6: 프로젝트 귀속(llm_call_log)·judge 모델 override·force 전파를 render_one_background 에 스레드. """ from app.core.config import settings if not getattr(settings, "plate_multiroll_enabled", False): return None return { "project_id": self.project_id, "episode_id": self.episode_id, "project_config": self.project_config, "force": getattr(self, "_render_mode", "resume") == "force", } def _execute(self, mode: str = "resume") -> Dict[str, Any]: from app.core.config import settings # plate_multiroll force 전파용 (nested 렌더 클로저에서 참조) self._render_mode = mode 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": { "consumed_bg_catalog_hash": "", "consumed_shot_binding_hash": "", }, } # W19B-3/W20B mutual exclusion gate. background_render w18j_overlap # 와 shot_aware_bg_render_plan_enabled 가 둘 다 켜져 있으면 reference # graph SOT 가 서로 다른 path 에서 산출돼 silent divergence — fail # closed. 한쪽만 켜져 있거나 둘 다 default off 일 때는 통과 (DB / # ImageAsset side effect 없이 즉시 return). from app.core.steps._selector_guards import detect_w19b3_w20b_conflict conflict_msg = detect_w19b3_w20b_conflict(settings) if conflict_msg is not None: return { "applicable_count": 0, "completed_count": 0, "failed_count": 1, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "error": conflict_msg, "data": { "consumed_bg_catalog_hash": "", "consumed_shot_binding_hash": "", "selector_conflict": conflict_msg, }, } plans_cp = self._load_prev_checkpoint("background_master_plan") prompts_cp = self._load_prev_checkpoint("background_prompt") fp_render_cp = self._load_prev_checkpoint("floor_plan_render") # Phase 9.1: floor_plan_prompt cp의 camera_recommendations[]를 읽어 # 각 group 결과에 bg_id별로 매칭되는 cr 1개를 보존한다. scene_detail # consumer + scene_image_pipeline framing-aware 분기에서 활용. fp_prompt_cp = self._load_prev_checkpoint("floor_plan_prompt") plans_map = ( ((plans_cp or {}).get("data", {}) or {}).get("plans", {}) or {} ) prompts_map = ( ((prompts_cp or {}).get("data", {}) or {}).get("backgrounds", {}) or {} ) fp_paths_str = { fid: entry.get("png_path", "") for fid, entry in ( ((fp_render_cp or {}).get("data", {}) or {}).get("floor_plans", {}) or {} ).items() if entry.get("status") == "ok" } # ── W21B-w5 STEP5-B: simplified floor-plan sidecar swap ── # When the sidecar step ran and produced a structurally-valid LIGHT FP # PNG, swap the I2I anchor for it (the detailed FP reads poorly — rooms # bleed). This is the SINGLE injection point that feeds BOTH render # paths: legacy ``_process`` (``fp_paths_str[fp_first]``) and the # shot_aware substrate (``base_fp_str = fp_paths_str.get(fp_id)``). # ONLY the FP anchor PNG changes — the detailed FP stays the # shot_projection_card / edge-judge substrate, so the W21B-w5 partition # SOT is untouched. Flag OFF → this is a no-op (byte-identical path); # a missing/invalid light PNG falls back to the detailed FP per fp. light_fp_provenance = self._apply_light_fp_sidecar(fp_paths_str) # Phase 9.1: floor_plan_prompt cp의 camera_recommendations[]를 # bg_id → recommendation dict로 펼친다. 각 fp가 여러 bg를 커버할 수 # 있고 각 cr은 자기 bg_id를 명시하므로 bg_id 단위 평탄화가 자연스럽다. # 같은 bg_id가 여러 fp에 등장하면 first-wins (master_plan 기준 1:1). fp_prompts_map = ( ((fp_prompt_cp or {}).get("data", {}) or {}).get("floor_plans", {}) or {} ) camera_recs_by_bg: Dict[str, Dict[str, str]] = {} for fid, fp_entry in fp_prompts_map.items(): if not isinstance(fp_entry, dict) or fp_entry.get("status") != "ok": continue for cr in fp_entry.get("camera_recommendations") or []: if not isinstance(cr, dict): continue bg_id = cr.get("bg_id") or "" if not bg_id: continue if bg_id in camera_recs_by_bg: # master_plan 1:1 가정 위반 — 후속 floor_plan의 cr 무시 (first-wins) logger.warning( "background_render: duplicate camera_recommendation for bg_id=%r (fp=%r) — first-wins, dropping later", bg_id, fid, ) continue camera_recs_by_bg[bg_id] = { "camera_position": cr.get("camera_position", "") or "", "camera_height": cr.get("camera_height", "") or "", "lens_hint": cr.get("lens_hint", "") or "", "framing_notes": cr.get("framing_notes", "") or "", } # E2E11 ② (L13B01 계단 반전·L04B01/02 계단 누락): fp별 STRUCTURE # FACTS 소스 — floor_plan_prompt numbered_elements(dossier 파생 SOT) # + bg별 use 목록(exact integer join 계약). 선별(base_* ∩ use)은 # materialize 직전 select_structure_facts 가 수행 — state_overlay # 일시 요소는 기존 overlay 소비부 관할 유지(Codex BLOCKING-1). # flag OFF=None 전달=기존 byte-identical. structure_facts_on = bool(getattr( settings, "plate_structure_facts_enabled", False)) numbered_by_fp: Dict[str, List[Dict[str, Any]]] = {} facts_use_by_bg: Dict[str, List[Any]] = {} if structure_facts_on: for fid, fp_entry in fp_prompts_map.items(): if not isinstance(fp_entry, dict) or ( fp_entry.get("status") != "ok"): continue els = [ e for e in fp_entry.get("numbered_elements") or [] if isinstance(e, dict) ] if els: numbered_by_fp[fid] = els for cr in fp_entry.get("camera_recommendations") or []: if not isinstance(cr, dict): continue _fb = cr.get("bg_id") or "" if _fb and _fb not in facts_use_by_bg and isinstance( cr.get("use_numbered_elements"), list): facts_use_by_bg[_fb] = cr["use_numbered_elements"] # bg DAG 구성 (master_plan 기준; depends_on_bg[0]를 parent로 사용) # D6: marker-gated bg_id filter (BG_ID_RE strict if D6 cp, legacy regex else). bg_filter_re = _select_bg_id_filter(plans_cp) # T4-fix B4: D6 marker 모드면 invalid bg_id silent skip 안 함 — fail-fast. # legacy 모드는 기존 skip+log 패턴 (transitional cp 호환). d6_strict_mode = (bg_filter_re is BG_ID_RE) items: Dict[str, Dict[str, Any]] = {} order: List[str] = [] bg_specs: Dict[str, Dict[str, Any]] = {} for gid, entry in plans_map.items(): if entry.get("status") != "ok": continue plan = entry.get("plan") or {} for bg in plan.get("backgrounds") or []: bid = bg.get("bg_id") if not bid or not bg_filter_re.match(bid): if d6_strict_mode: raise ValueError( f"background_render: D6 marker cp 에 invalid bg_id " f"{bid!r} (group={gid!r}) — fail-fast. master_plan " f"post-processing handshake 미완료 또는 cp 손상. " f"force re-run background_master_plan 의무." ) logger.error( "background_render: unsafe bg_id %r — skip", bid ) continue if bid in items: # 중복 bg_id (different group) → first-wins continue deps = bg.get("depends_on_bg") or [] items[bid] = { "parent_id": deps[0] if deps else "", } order.append(bid) bg_specs[bid] = bg # 실제 렌더 가능한 bg = prompt 가 ok 인 것만 renderable = { bid for bid in order if prompts_map.get(bid, {}).get("status") == "ok" } # W22 W4b (2026-07-10, 설계 §7-2 확정 A): 직행 모드에선 야외 loc bg 의 # plate 렌더를 skip — 야외 배경은 장소 캐논(nb2 직행)이 담당. # ★D6 불변식: master_plan 카탈로그/consumed_hash 는 절대 건드리지 않고 # (StaleUpstreamError 0) render 단계에서만 per-bg skip. OFF = 조인 자체 # 미실행 → byte-identical. _direct_skipped: Dict[str, Dict[str, Any]] = {} if bool(getattr(settings, "outdoor_direct_compose_enabled", False)): _outdoor_locs = _outdoor_loc_ids_from_classify( self._load_prev_checkpoint("background_classify") ) if _outdoor_locs: renderable, _direct_skipped = _split_outdoor_direct_skip( renderable, bg_specs, _outdoor_locs, ) if _direct_skipped: logger.info( "background_render: W22 직행 — 야외 bg %d개 plate skip: %s", len(_direct_skipped), sorted(_direct_skipped), ) # 3a fix (2026-07-01, 정정): depth-1 star anchor **주입은 legacy w18j_overlap # 전용**. 그 모드에서만 depends_on_bg 가 실제 i2i 드라이버(prior_bg_paths, line # 560-564)이기 때문. shot_aware_plan 경로는 prior_bg 를 plan node(substrate # ref_tree_parents / adapter selected_refs)에서 뽑고 depends_on_bg 를 attach 에 # 안 쓰므로(워크플로 매핑 확증), 주입해도 실제 첨부를 못 바꾸고 parent_id 만 # 덮어써 mismatch 만 만든다 → shot_aware 에선 미주입. 3a 배경↔배경 엣지는 # 아래 _register_image_assets 의 '실제 첨부(prior_bg_ids) 기록'으로 복원. # OFF(default) 면 미호출 = byte-identical. star_anchor_by_bid: Dict[str, str] = {} if (bool(getattr(settings, "background_render_star_anchor_enabled", False)) and getattr(settings, "background_render_reference_mode", "") == "w18j_overlap"): star_anchor_by_bid = _inject_star_anchors( bg_specs, items, order, renderable) if star_anchor_by_bid: logger.info( "background_render: star anchor 주입 %d bg (legacy w18j only)", len(star_anchor_by_bid)) if not renderable: # D6 T5c + T5-fix B2: empty path 도 hash stamp (catalog + binding). # W22 W4b: 직행 skip 마커는 empty path 에도 보존 (가시성). plans_data_empty = (plans_cp or {}).get("data", {}) or {} return { "applicable_count": 1, "completed_count": 1, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": { "groups": dict(_direct_skipped), "consumed_bg_catalog_hash": plans_data_empty.get("bg_catalog_hash", "") or "", "consumed_shot_binding_hash": plans_data_empty.get("shot_binding_hash", "") or "", }, } from app.modules.pipeline._dag_levels import compute_dag_levels from app.modules.pipeline.background_render import render_one_background 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" / "background_chain" ) image_dir.mkdir(parents=True, exist_ok=True) image_dir_resolved = image_dir.resolve() rendered_paths: Dict[str, Path] = {} groups_out: Dict[str, Any] = {} failed = 0 from app.modules.pipeline._workers import resolve_workers max_workers = resolve_workers(default=4, cap=8) def _process( bid: str, snapshot: Dict[str, Path] ) -> Tuple[str, Dict[str, Any]]: spec = bg_specs[bid] prompt_entry = prompts_map.get(bid, {}) or {} t2i_prompt = prompt_entry.get("t2i_prompt", "") shot_guides = prompt_entry.get("shot_guides", []) or [] shot_ids = spec.get("applies_to_shots", []) or [] parent_id = (spec.get("depends_on_bg") or [""])[0] location_id = spec.get("loc_id", "") location_name = spec.get("sub_location", "") variant_label = spec.get("state_label", "") fp_first = (spec.get("depends_on_fp") or [""])[0] fp_path: Optional[Path] = None if fp_first and fp_paths_str.get(fp_first): fp_path = Path(fp_paths_str[fp_first]) prior_bg_paths: List[Path] = [] for dep in spec.get("depends_on_bg") or []: p = snapshot.get(dep) if p is not None and p.exists(): prior_bg_paths.append(p) out_path = image_dir / f"{bid}.png" # path traversal 2중 가드 (bg_id 정규식 + resolve relative_to) try: out_path.resolve().relative_to(image_dir_resolved) except (ValueError, OSError): logger.error( "background_render: out_path escapes image_dir for %s", bid ) return bid, { "status": "rejected_path", "location_id": location_id, "location_name": location_name, "png_path": "", "t2i_prompt": t2i_prompt, "shot_guides": shot_guides, "shot_ids": shot_ids, "scenes": [], "parent_id": parent_id, "ref_used": "text_only", "render_attempts": 0, "render_error": "out_path escapes image_dir", "variant_label": variant_label, "floor_plan_used": False, # Phase 9.1 "camera_recommendations": camera_recs_by_bg.get(bid, {}), # W21B-wave-1 — rejected_path 에서도 lineage diagnostic # 일관성. 실제 render 호출 전에 끊겼으므로 attached_ref_labels # 는 사실상 empty 가 정상 (path resolve 실패 → 첨부 없음). "attached_reference_lineage": _build_attached_reference_lineage( fp_id=fp_first, fp_path=None, prior_bg_paths=[], ref_used="text_only", ), } info = render_one_background( openai_client=client, image_model="gpt-image-2.5-sunburst", prompt=t2i_prompt, out_path=out_path, fp_path=fp_path, prior_bg_paths=prior_bg_paths, bg_id=bid, plate_multiroll_ctx=self._plate_multiroll_ctx(), ) ref_used = info.get("ref_used", "text_only") return bid, { "status": info.get("status", "failed"), "location_id": location_id, "location_name": location_name, "png_path": ( info.get("png_path", "") if info.get("status") == "ok" else "" ), "t2i_prompt": t2i_prompt, "shot_guides": shot_guides, "shot_ids": shot_ids, "scenes": [], "parent_id": parent_id, "ref_used": ref_used, "render_attempts": info.get("attempts", 0), "render_error": info.get("final_block_reason") or "", "variant_label": variant_label, "floor_plan_used": fp_path is not None and fp_path.exists(), # Phase 9.1: 매칭되는 floor_plan camera_recommendation (없으면 {}). "camera_recommendations": camera_recs_by_bg.get(bid, {}), # W21B-wave-1 — checkpoint-only attached reference lineage # diagnostic. DB / ImageAsset 비변경 (Codex 결정 2(c)). "attached_reference_lineage": _build_attached_reference_lineage( fp_id=fp_first, fp_path=fp_path, prior_bg_paths=prior_bg_paths, ref_used=ref_used, ), # 무인 계약 감사 3필드 CP 영속 (E2E10 실측: whitelist 탈락 # 으로 people_detected 분포 검증 불가). OFF/부재=빈 spread. **_unmanned_audit_fields(info), } # W19B-3 / W20C / W20E7-A: selector branch. Legacy path keeps the # DAG level ThreadPool + snapshot semantics verbatim. The # ``shot_aware_plan`` opt-in path consumes the W20B LLM-emitted # plan. The ``w18j_overlap`` opt-in selector value is W20E7-A # DEPRECATED — its branch fails closed for every renderable bg # with a stable error code (``w18j_overlap_deprecated`` / # ``background_render_reference_mode_deprecated``). The W19B-3 # deterministic reference planner module + its render queue # helper are deleted; zero image API calls, zero deterministic # reference decisions. catalog_dump: List[Dict[str, Any]] = [] # W-L: shot_aware 큐가 loc별 aerial 산출/재사용 진단을 채운다. # legacy/w18j 경로 또는 flag OFF 면 빈 dict 그대로(= cp shape # byte-identical — data key 자체를 안 쓴다). location_aerials: Dict[str, Any] = {} if settings.background_render_reference_mode == "w18j_overlap": for bid in order: if bid not in renderable: continue spec = bg_specs.get(bid) or {} prompt_entry = prompts_map.get(bid, {}) or {} groups_out[bid] = self._build_opt_in_failed_entry( bid=bid, spec=spec, prompt_entry=prompt_entry, camera_recs_by_bg=camera_recs_by_bg, error=( "background_render_reference_mode_deprecated: " "w18j_overlap_deprecated — the W19B-3 " "deterministic reference planner path is no " "longer an active production option (W20E7-A). " "Set background_render_reference_mode to " "'shot_aware_plan' (W20C) or leave the default " "'legacy'. Zero image calls were made." ), ) failed += 1 elif settings.background_render_reference_mode == "shot_aware_plan": plan_cp = self._load_prev_checkpoint( "shot_aware_bg_render_plan" ) plans_per_fp = ( ((plan_cp or {}).get("data", {}) or {}).get("per_fp", {}) or {} ) ( groups_out, rendered_paths, failed, catalog_dump, ) = self._run_shot_aware_plan_queue( bg_specs=bg_specs, prompts_map=prompts_map, fp_paths_str=fp_paths_str, camera_recs_by_bg=camera_recs_by_bg, plans_per_fp=plans_per_fp, renderable=renderable, order=order, image_dir=image_dir, image_dir_resolved=image_dir_resolved, client=client, # W-G: flag OFF/링크 없음 = {} → 큐 내부 조회가 전부 miss # (기존 경로 byte-identical). building_fp_by_loc=self._load_building_fp_by_loc(plans_cp), # W-K: flag OFF = {} → same-place 체이닝/재배열 전부 비활성 # (기존 경로 byte-identical). group_membership_by_loc=self._load_group_membership_by_loc(), # W-L: flag OFF = {} → aerial 생성/치환 전부 비활성 # (기존 경로 byte-identical). aerial_loc_context=self._load_aerial_loc_context(), aerial_diag_out=location_aerials, # E2E11 ②: fp별 STRUCTURE FACTS 소스+bg별 use 목록 # (flag OFF={} → 비활성) numbered_by_fp=numbered_by_fp, facts_use_by_bg=facts_use_by_bg, # W-L Codex NARROW: 자기 cp 의 loc별 aerial prompt_version — # cached 재사용은 버전 일치 시에만(프롬프트 의미 변경 후 # 이전 의미의 PNG 부활 방지). prior_location_aerials=( ((self._load_prev_checkpoint("background_render") or {}) .get("data", {}) or {}).get("location_aerials") or {} ), mode=mode, ) else: 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 so the runtime cap is # authoritative across the submit hop. _submit_process = bind_current_budget(_process) futures = { pool.submit(_submit_process, bid, snapshot): bid for bid in level } for fut in as_completed(futures): bid = futures[fut] try: _bid, res = fut.result() except Exception as exc: logger.error( "background_render: bg %s thread raised: %s", bid, exc, ) spec = bg_specs.get(bid, {}) prompt_entry = prompts_map.get(bid, {}) or {} res = { "status": "failed", "location_id": spec.get("loc_id", ""), "location_name": spec.get("sub_location", ""), "png_path": "", "t2i_prompt": prompt_entry.get("t2i_prompt", ""), "shot_guides": prompt_entry.get("shot_guides", []) or [], "shot_ids": spec.get("applies_to_shots", []) or [], "scenes": [], "parent_id": (spec.get("depends_on_bg") or [""])[0], "ref_used": "text_only", "render_attempts": 0, "render_error": str(exc)[:200], "variant_label": spec.get("state_label", ""), "floor_plan_used": False, # Phase 9.1: 실패 path에서도 cr 보존 (consumer drop 안 함). "camera_recommendations": camera_recs_by_bg.get(bid, {}), # W21B-wave-1 — thread-raise path 에선 _process # 안의 fp_path / prior_bg_paths 변수가 가시화되지 # 않으므로 spec 의 depends_on_fp 만 lineage 에 # 남긴다 (실제 첨부 여부는 알 수 없으니 빈 labels). "attached_reference_lineage": _build_attached_reference_lineage( fp_id=(spec.get("depends_on_fp") or [""])[0], fp_path=None, prior_bg_paths=[], ref_used="text_only", ), } groups_out[bid] = res if res.get("status") == "ok" and res.get("png_path"): rendered_paths[bid] = Path(res["png_path"]) else: failed += 1 # W22 W4b: 직행 skip 마커를 groups 에 합류 — status 가 'ok' 아니라 # 하류(load_background_chain_bg_map)가 자연 무시, 가시성만 제공. groups_out.update(_direct_skipped) # input order 보존 (Phase 6 호환 shape: data.groups[bg_id]) ordered = {bid: groups_out[bid] for bid in order if bid in groups_out} # ── ImageAsset UPSERT — main thread, after all renders ── # 실패 시 raise → step "failed" 마킹 (silent miss 차단, PID 0bb48ebf 회귀 가드). try: self._register_image_assets( ordered, order, location_aerials=location_aerials or None, ) except Exception as exc: ok_groups = sum( 1 for v in ordered.values() if v.get("status") == "ok" ) logger.error( "[BG_RENDER_SYNC_FAIL] background_render: chain_bg ImageAsset INSERT 실패. " "PNG는 디스크에 있으나 DB out of sync — downstream(scene_image_pipeline)이 " "참조 이미지를 못 찾음. groups_total=%d ok_status=%d exc=%s\n%s", len(ordered), ok_groups, exc, traceback.format_exc(), ) try: self.db.rollback() except Exception as rb_exc: logger.error( "[BG_RENDER_SYNC_FAIL] rollback also failed: %s", rb_exc ) raise # D6 T5c + T5-fix B2: catalog + binding 양쪽 stamp. # render 가 background_prompt 의 t2i_prompt / shot_guides 소비 (binding-derived) # + ImageAsset 의 shot_ids 도 applies_to_shots 에서 저장. # binding 변경 시 prompt 갱신 → render input 도 변함 → consumer 추적 의무. plans_data = (plans_cp or {}).get("data", {}) or {} data: Dict[str, Any] = { "groups": ordered, "consumed_bg_catalog_hash": plans_data.get("bg_catalog_hash", "") or "", "consumed_shot_binding_hash": plans_data.get("shot_binding_hash", "") or "", } # W19B-3 / W20C opt-in only: persist the ephemeral reference # catalog. Legacy path never writes this key — cp shape # byte-compatible. Both opt-in paths use the same key (the # catalog entries' ``source_kind`` distinguishes which path # produced them). if settings.background_render_reference_mode in ( "w18j_overlap", "shot_aware_plan", ): data["bg_reference_catalog"] = catalog_dump # W21B-w5 STEP5-B: persist the FP-anchor provenance only when the sidecar # produced swaps (non-empty). Flag OFF → empty → key absent → cp shape # byte-identical to the pre-STEP5-B path. if light_fp_provenance: data["light_fp_sidecar_provenance"] = light_fp_provenance # W-L: loc별 aerial 산출/재사용 진단 (flag OFF/비대상 = 빈 dict → key # 부재 → cp shape byte-identical). if location_aerials: data["location_aerials"] = location_aerials # P0-1 (W21B-w6, 06-03 worktree 구현의 main 포팅 — 2026-06-11 # fresh full E2E 실측 재발로 landing): per-bg 단위로 실제 ok 렌더 수 / # renderable 전체 수를 보고한다. 이전엔 step 전체를 1 unit 으로 취급하고 # completed_count 를 ``1 if failed == 0 else 0`` 로 이진화 → bg 1개라도 # 렌더 실패하면 completed=0 → step_runner 가 status='failed' (partial # 아님) → analysis_dispatch 즉시 break → 정상 렌더된 bg + 모든 # downstream 이 통째로 stop 됐다 ("L22 1건 실패가 L06·L08 까지 차단"의 # 상류 원인 = 06-02 audit P0-1). 완료 기준은 status=="ok" AND png_path # (rendered_paths 등록 기준과 동일 = downstream scene_image_pipeline 이 # 실제로 참조 가능한 PNG; reuse alias 도 status=ok + target png_path 라 # 포함). denominator 는 ``ordered`` 가 아니라 renderable set — groups_out # 에는 renderable bg 만 surface 하고 renderable set 이 원천 SOT. # no-renderable 분기는 위에서 별도 early return. ImageAsset DB sync # 실패는 위에서 raise 유지 (개별 render failure 아닌 DB/asset SOT 동기화 # 실패라 partial 금지). background_prompt (55db7f8) 와 동일 패턴. total_bgs = len(renderable) completed_bgs = sum( 1 for e in ordered.values() if isinstance(e, dict) and e.get("status") == "ok" and e.get("png_path") ) return { "applicable_count": max(1, total_bgs), "completed_count": completed_bgs, "failed_count": failed, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": data, } # ── opt-in render path helpers ── # # W20E7-A: the W19B-3 ``_run_sequential_overlap_queue`` method and # the deterministic reference planner module it imported are # deleted. The ``w18j_overlap`` selector value is now handled in # ``_execute`` with a fail-closed loop that emits a deprecated # error per renderable bg via ``_build_opt_in_failed_entry``. # The remaining helper below is still used by both the (fail-closed) # legacy w18j branch and the active ``shot_aware_plan`` (W20C) queue. def _build_opt_in_failed_entry( self, *, bid: str, spec: Dict[str, Any], prompt_entry: Dict[str, Any], camera_recs_by_bg: Dict[str, Dict[str, str]], error: str, status: str = "failed", decision_dict: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: entry: Dict[str, Any] = { "status": status, "location_id": spec.get("loc_id", ""), "location_name": spec.get("sub_location", ""), "png_path": "", "t2i_prompt": prompt_entry.get("t2i_prompt", ""), "shot_guides": prompt_entry.get("shot_guides", []) or [], "shot_ids": spec.get("applies_to_shots", []) or [], "scenes": [], "parent_id": (spec.get("depends_on_bg") or [""])[0], "ref_used": "text_only", "render_attempts": 0, "render_error": error[:200], "variant_label": spec.get("state_label", ""), "floor_plan_used": False, "camera_recommendations": camera_recs_by_bg.get(bid, {}), # W21B-wave-1 — opt-in failed path 에서도 lineage diagnostic 일관성. # 실제 render 호출 없이 끊겼으므로 attached_ref_labels 는 empty. "attached_reference_lineage": _build_attached_reference_lineage( fp_id=(spec.get("depends_on_fp") or [""])[0], fp_path=None, prior_bg_paths=[], ref_used="text_only", ), } if decision_dict is not None: entry["reference_decision"] = decision_dict return entry # ── W-G (2026-07-03): same-building indoor fp → outdoor plate 구조 참조 ── def _load_building_fp_by_loc( self, plans_cp: Optional[Dict[str, Any]], ) -> Dict[str, Dict[str, str]]: """flag ``outdoor_building_fp_ref_enabled`` (W-G) **또는** ``outdoor_building_anchor_ref_enabled`` (W-I) **또는** ``outdoor_aerial_reference_enabled`` (W-L) ON 일 때 outdoor loc → same-building indoor fp 컨텍스트를 구성한다. W-I 는 이 링크 맵을 outdoor 멤버십 + group_id 판정에 재사용하고, W-L 은 mixed 그룹 aerial 생성의 I2I fp ref 소스로 재사용한다(plate 에의 fp 첨부 여부는 큐의 W-G flag 게이트가 별도 결정). 반환: ``{outdoor_loc_sid: {"fp_id", "indoor_loc_sid", "group_id", "png_path", "fp_asset_id"}}`` — ``png_path`` 는 floor_plan_render cp 의 status=='ok' entry 만, ``fp_asset_id`` 는 ImageAsset(asset_type= 'floor_plan', variant_type=fp_id) 구조 조인(미해결이면 빈 문자열 — ref 첨부는 진행하되 UUID lineage 만 생략). 세 flag OFF/실패 = {} (비차단, 기존 경로 byte-identical). """ from app.core.config import settings # W22 W4b (2026-07-10): 직행 모드 — 야외 배경은 장소 캐논이 담당, # aerial 생성/참조 전부 비활성 (3-flag 무관 early return). OFF = 불변. if bool(getattr(settings, "outdoor_direct_compose_enabled", False)): return {} if not ( bool(getattr(settings, "outdoor_building_fp_ref_enabled", False)) or bool(getattr( settings, "outdoor_building_anchor_ref_enabled", False)) or bool(getattr( settings, "outdoor_aerial_reference_enabled", False)) ): return {} try: from app.modules.pipeline.building_fp_link import ( build_outdoor_building_fp_link, indoor_fp_ids_by_loc, ) bgc_cp = self._load_prev_checkpoint("background_classify") catalog = ( (plans_cp or {}).get("data", {}) or {} ).get("background_catalog") or {} link = build_outdoor_building_fp_link( ((bgc_cp or {}).get("data", {}) or {}).get("building_groups") or [], indoor_fp_ids_by_loc(catalog), ) if not link: return {} fpr_cp = self._load_prev_checkpoint("floor_plan_render") fpr_map = ( (fpr_cp or {}).get("data", {}) or {} ).get("floor_plans") or {} out: Dict[str, Dict[str, str]] = {} for oloc, lk in link.items(): entry = fpr_map.get(lk.get("fp_id")) or {} png = str(entry.get("png_path") or "") if entry.get("status") == "ok" and png: out[oloc] = {**lk, "png_path": png, "fp_asset_id": ""} if not out: return {} # fp_id → ImageAsset UUID (variant_type 구조 조인, 1쿼리 캐시맵). try: from app.models.project import ImageAsset needed = {lk["fp_id"] for lk in out.values()} for vt, aid in ( self.db.query(ImageAsset.variant_type, ImageAsset.id) .filter( ImageAsset.project_id == self.project_id, ImageAsset.episode_id == self.episode_id, ImageAsset.asset_type == "floor_plan", ImageAsset.variant_type.in_(sorted(needed)), ) .all() ): for lk in out.values(): if lk["fp_id"] == str(vt or ""): lk["fp_asset_id"] = str(aid) except Exception as exc: # noqa: BLE001 — UUID 미해결도 ref 는 첨부 logger.warning( "background_render: building fp asset UUID resolve 실패 " "(비차단, lineage 만 생략): %s", exc) logger.info( "background_render: building fp 링크 %d loc — %s", len(out), {k: v["fp_id"] for k, v in out.items()}, ) return out except Exception as exc: # noqa: BLE001 — 비차단 logger.warning( "background_render: building fp 링크 로드 실패 (비차단): %s", exc) return {} def _load_group_membership_by_loc(self) -> Dict[str, Dict[str, Any]]: """flag ``same_place_render_chain_enabled`` (W-K) ON 일 때 ``background_classify.building_groups`` 의 **전 멤버**(실내 포함) loc 멤버십을 로드한다. W-K 의 그룹 anchor 등록(outdoor ok 렌더만)·첨부 (그룹 전 멤버)·lane 재배열 정렬 키(mixed 그룹 outdoor 우선) 판정에 쓴다 — W-G 링크 맵(outdoor loc 만, fp png 필수)으로는 실내 첨부를 판정할 수 없다. flag OFF/실패 = {} (비차단, 기존 경로 byte-identical). 반환: ``{loc_sid: {"group_id", "is_indoor", "group_mixed"}}``. """ from app.core.config import settings if not bool(getattr( settings, "same_place_render_chain_enabled", False)): return {} try: from app.modules.pipeline.building_fp_link import ( build_group_membership_by_loc, ) bgc_cp = self._load_prev_checkpoint("background_classify") membership = build_group_membership_by_loc( ((bgc_cp or {}).get("data", {}) or {}).get("building_groups") or [] ) if membership: logger.info( "background_render: same-place 그룹 멤버십 %d loc — %s", len(membership), {k: v["group_id"] for k, v in membership.items()}, ) return membership except Exception as exc: # noqa: BLE001 — 비차단 logger.warning( "background_render: same-place 그룹 멤버십 로드 실패 " "(비차단): %s", exc) return {} # ── W-L (2026-07-03): 야외 loc 실사형 aerial establishing SOT ── def _load_aerial_loc_context(self) -> Dict[str, Dict[str, Any]]: """flag ``outdoor_aerial_reference_enabled`` (W-L) ON 일 때 ``background_classify.building_groups`` 에서 loc별 aerial 컨텍스트 (is_indoor/group_mixed/label/summary)를 로드한다. 야외 판정 (is_indoor==False)과 aerial 프롬프트의 location 데이터 소스. flag OFF/실패 = {} (비차단, 기존 경로 byte-identical). """ from app.core.config import settings # W22 W4b (2026-07-10, Codex NARROW_1 정정): 직행 모드 — 야외 배경은 # 장소 캐논이 담당, aerial 생성/참조 전부 비활성 (flag 무관 early # return). OFF = 불변. ※최초 반영이 _load_building_fp_context 에 # 잘못 들어갔었음 — 그쪽 가드도 유지(야외 plate 전용 컨텍스트라 무해). if bool(getattr(settings, "outdoor_direct_compose_enabled", False)): return {} if not bool(getattr( settings, "outdoor_aerial_reference_enabled", False)): return {} try: from app.modules.pipeline.location_aerial import ( build_loc_aerial_context, ) bgc_cp = self._load_prev_checkpoint("background_classify") ctx = build_loc_aerial_context( ((bgc_cp or {}).get("data", {}) or {}).get("building_groups") or [] ) outdoor = sorted( loc for loc, c in ctx.items() if c.get("is_indoor") is False ) if outdoor: logger.info( "background_render: aerial 컨텍스트 %d loc (outdoor %s)", len(ctx), outdoor, ) return ctx except Exception as exc: # noqa: BLE001 — 비차단 logger.warning( "background_render: aerial 컨텍스트 로드 실패 (비차단): %s", exc) return {} # ── W20C: opt-in shot_aware_plan render queue (LLM-emitted plan) ── def _run_shot_aware_plan_queue( self, *, bg_specs: Dict[str, Dict[str, Any]], prompts_map: Dict[str, Dict[str, Any]], fp_paths_str: Dict[str, str], camera_recs_by_bg: Dict[str, Dict[str, str]], plans_per_fp: Dict[str, Dict[str, Any]], renderable: set, order: List[str], image_dir: Path, image_dir_resolved: Path, client: Any, building_fp_by_loc: Optional[Dict[str, Dict[str, str]]] = None, group_membership_by_loc: Optional[Dict[str, Dict[str, Any]]] = None, aerial_loc_context: Optional[Dict[str, Dict[str, Any]]] = None, aerial_diag_out: Optional[Dict[str, Any]] = None, prior_location_aerials: Optional[Dict[str, Any]] = None, numbered_by_fp: Optional[Dict[str, List[Dict[str, Any]]]] = None, facts_use_by_bg: Optional[Dict[str, List[Any]]] = None, mode: str = "resume", ) -> Tuple[ Dict[str, Dict[str, Any]], Dict[str, Path], int, List[Dict[str, Any]], ]: """W20C opt-in render pass. Consumes the W20B ``shot_aware_bg_render_plan`` checkpoint and renders each plan node sequentially in ``node_index`` order. The reference graph SOT is the LLM-emitted plan; this method does NOT re-classify / re-score / fall back. ``render_one_background`` runs with ``max_attempts=2`` — normal path is still exactly one image call per BG (W20 contract); the 2nd attempt fires ONLY on a moderation rejection, where the render module sanitizes the prompt and retries once (2026-07-11 실측: 상태 bg safety 거부 시 산출 0 방지). W-L: ``aerial_loc_context`` (flag OFF = {} = 전부 비활성)가 주어지면 큐 시작 시 야외 loc(is_indoor==False)마다 실사형 aerial establishing 1장을 생성/재사용(mode=='force' 만 재생성)하고, 야외 bg 렌더의 ref 를 fp(평면도) 대신 aerial **1순위** 로 교체한다. ``aerial_diag_out`` (mutable dict)에 loc별 산출 진단을 채운다. """ from app.core.config import settings from app.modules.pipeline.background_render import ( BUILDING_ANCHOR_PLATE_GUIDANCE, BUILDING_FP_PLATE_GUIDANCE, BUILDING_INTERIOR_EXTERIOR_ANCHOR_GUIDANCE, INTERIOR_SAME_BUILDING_TEXT_GUIDANCE, SAME_PLACE_PLATE_GUIDANCE, render_one_background, ) from app.modules.pipeline.outdoor_structure_seed import ( SEED_SITE_GUIDANCE, ) from app.modules.pipeline.location_aerial import ( AERIAL_PROMPT_VERSION, AERIAL_SITE_GUIDANCE, compute_aerial_context_hash, decide_group_plate_order, render_location_aerial, ) from app.modules.pipeline.background_render_substrate import ( resolve_render_substrate, ) from app.modules.pipeline.shot_aware_bg_render_adapter import ( ShotAwareRenderAdapterError, is_plan_production_clear, materialize_decision, ordered_nodes, select_structure_facts, ) groups_out: Dict[str, Dict[str, Any]] = {} rendered_paths: Dict[str, Path] = {} failed = 0 catalog_dump: List[Dict[str, Any]] = [] # E2E11 ②: STRUCTURE FACTS+VIEW AUTHORITY prefix — flag OFF 또는 # 소스 미전달 = materialize 에 None = 기존 byte-identical. # 선별=base_* ∩ bg use 목록 (select_structure_facts, BLOCKING-1). structure_facts_on = bool(getattr( settings, "plate_structure_facts_enabled", False)) numbered_by_fp = numbered_by_fp or {} facts_use_by_bg = facts_use_by_bg or {} def _facts_for(bid: str, fid: str): if not structure_facts_on: return None return select_structure_facts( numbered_by_fp.get(fid, ()), facts_use_by_bg.get(bid), ) renderable_set = set(renderable) # Partition renderable bgs. W21B-wave-2: non-interior fp-less # surfaces bypass the W20B fp/dossier/geometry planner and render as # direct plates. interior_room without fp remains fail-closed. bgs_by_fp: Dict[str, List[str]] = {} direct_plate_bids: List[str] = [] for bid in order: if bid not in renderable_set: continue spec = bg_specs.get(bid) or {} deps_fp = spec.get("depends_on_fp") or [] if not deps_fp: surface_role = spec.get("surface_role", "interior_room") or "interior_room" if surface_role in {"exterior_plate", "transition_zone", "site_surface"}: direct_plate_bids.append(bid) continue prompt_entry = prompts_map.get(bid, {}) or {} groups_out[bid] = self._build_opt_in_failed_entry( bid=bid, spec=spec, prompt_entry=prompt_entry, camera_recs_by_bg=camera_recs_by_bg, error="shot_aware_plan: bg has no depends_on_fp", ) failed += 1 continue bgs_by_fp.setdefault(deps_fp[0], []).append(bid) ingestion_counter = 0 # W-G: outdoor loc → same-building indoor fp (양 flag OFF/링크 없음 = {}). # 로더는 W-G ∨ W-I 게이트 — fp "첨부"는 아래 W-G flag 가 별도 결정. _bfp_by_loc = building_fp_by_loc or {} _wg_fp_on = bool( getattr(settings, "outdoor_building_fp_ref_enabled", False)) def _building_fp_for(spec: Dict[str, Any]) -> Optional[Dict[str, str]]: """spec.loc_id 기준 building fp 컨텍스트 (W-G flag ON + png 실재).""" if not _wg_fp_on: return None lk = _bfp_by_loc.get(str(spec.get("loc_id") or "")) if not lk: return None p = Path(str(lk.get("png_path") or "")) return lk if p.is_file() else None # ── W-I (2026-07-03): building group anchor 렌더 체이닝 ── # 같은 building 그룹(실내·실외 공존)의 outdoor bg 들이 도면(fp)만 공유하고 # 렌더를 상호 참조하지 않아 plate 마다 건물 외관이 재발명되는 결함의 근본 # 대응. 그룹의 **첫 ok 렌더**(결정론 — 렌더 순서 자체가 결정론)를 anchor 로 # 등록하고, 이후 같은 그룹의 outdoor bg 렌더에 시각 정체성 ref 로 첨부한다. # outdoor 멤버십/group_id 판정은 W-G 링크 맵(_bfp_by_loc) 구조 필드 재사용. # # ── W-K (2026-07-03): same-place 렌더 체이닝 확장 ── # flag ``same_place_render_chain_enabled`` ON 이면: # - 그룹 anchor **등록**은 outdoor ok 렌더만(멤버십 맵 is_indoor 판정, # lane 재배열로 외부 establishing 이 선행), **첨부**는 그룹 전 멤버 # (실내 포함) + 실내↔실외 정합형 guidance. # - loc anchor 신설 — 모든 loc(실내 포함)의 첫 ok 렌더를 등록하고 # 이후 같은 loc 렌더에 장소 정체성 ref 로 첨부(전 lane). # - 추가 anchor cap 2(loc+group) — 같은 png 면 경로 dedup 으로 1. # OFF 면 아래 전부 기존 W-I 동작 그대로(byte-identical). _anchor_on = bool( getattr(settings, "outdoor_building_anchor_ref_enabled", False)) _wk_on = bool( getattr(settings, "same_place_render_chain_enabled", False)) # ── W-M (2026-07-03): 야외 plate 참조 단계화 ── # ON 이면 ① W-K 그룹 anchor 첨부를 야외 멤버 한정으로 좁힘(실내 # plate 에 건물 외부 렌더 사용 금지 — 사용자 계약, 실내↔실외 연속은 # INTERIOR_SAME_BUILDING_TEXT_GUIDANCE 문구만) ② aerial 이 첨부되는 # 렌더에 building fp(실내 도면) 동시 첨부 금지 ③ aerial ok 그룹의 # 야외 render_new_plate bg 를 그룹 체인 lane 으로 추출(첫 plate= # 배치도 1장, 이후=이전 ok plate+배치도 max 2, 순서=LLM). OFF = # 아래 전부 기존 W-I/W-K/W-L 동작 그대로(byte-identical). _wm_on = bool( getattr(settings, "outdoor_plate_stage_chain_enabled", False)) _membership_by_loc = group_membership_by_loc or {} # group_id → {"bg_id", "png_path"} (first-ok-render-wins). _anchor_by_group: Dict[str, Dict[str, str]] = {} # W-K: loc_id → {"bg_id", "png_path"} (first-ok-render-wins, 전 lane). _loc_anchor_by_loc: Dict[str, Dict[str, str]] = {} def _register_building_anchor( bid: str, spec: Dict[str, Any], png_path: str, ) -> None: """ok 렌더 직후 호출 — 그룹 무등록 시 이 bg 를 anchor 로 등록. W-K ON 이면 멤버십 맵 기반 + **outdoor 렌더만** 등록(실내 렌더가 건물 외관 anchor 가 되는 것을 구조적으로 차단). OFF 면 기존 W-I 게이트(_bfp_by_loc = mixed 그룹 outdoor loc) 그대로. """ if not ((_anchor_on or _wk_on) and png_path): return loc = str(spec.get("loc_id") or "") if _wk_on: m = _membership_by_loc.get(loc) gid = str((m or {}).get("group_id") or "") if not gid or (m or {}).get("is_indoor") is not False: return else: lk = _bfp_by_loc.get(loc) gid = str((lk or {}).get("group_id") or "") if not gid: return if gid in _anchor_by_group: return _anchor_by_group[gid] = {"bg_id": bid, "png_path": png_path} logger.info( "background_render: building anchor 등록 group=%s anchor=%s", gid, bid, ) def _register_loc_anchor( bid: str, spec: Dict[str, Any], png_path: str, ) -> None: """W-K: ok 렌더 직후 호출 — loc 무등록 시 이 bg 를 loc anchor 로.""" if not (_wk_on and png_path): return loc = str(spec.get("loc_id") or "") if not loc or loc in _loc_anchor_by_loc: return _loc_anchor_by_loc[loc] = { "bg_id": bid, "png_path": png_path, # E2E9 육안 #3 (2026-07-19): 소스 렌더 성격 기록 — 실내 # 렌더가 실외 plate 의 loc anchor 로 첨부되는 오염 차단 # 판정 입력 (_loc_anchor_for, W-M ① 의 대칭 규칙). "interior": _is_interior_render(spec), } logger.info( "background_render: same-place loc anchor 등록 loc=%s " "anchor=%s", loc, bid, ) def _building_anchor_for( bid: str, spec: Dict[str, Any], existing_ref_paths: List[Optional[Path]], ) -> Optional[Dict[str, Any]]: """이 bg 렌더에 첨부할 그룹 anchor (자기 자신/중복 경로 제외). 반환: ``{"bg_id", "group_id", "path": Path}`` 또는 None. dedup — anchor png 가 이미 ref 목록(자기 fp/substrate prior/building fp/loc anchor)에 있으면 None (같은 PNG 이중 첨부 방지, 경로 비교만 — 구조적). W-K ON 이면 첨부 대상이 그룹 **전 멤버**(실내 포함, 멤버십 맵 판정), OFF 면 기존 W-I(_bfp_by_loc outdoor)만. """ if not (_anchor_on or _wk_on): return None # W-M(1): 그룹 anchor(건물 외부 렌더) 첨부에서 **실내 렌더**를 # 제외한다 — 건물 외부 이미지를 건물 내부 렌더에 사용 금지 # (사용자 계약). 판정은 bg 의 surface_role 구조 필드(SOT): # loc 분류(is_indoor)가 아니라 렌더 성격 기준 — 실내 분류 # loc 에 걸린 외관 plate(site_surface/transition_zone, 예: # 같은 건물의 저녁 외관)는 계속 anchor 를 받아 W-K 간판/외관 # 연속을 유지한다. 실내↔실외 연속은 INTERIOR_SAME_BUILDING_ # TEXT_GUIDANCE(프롬프트 문구만)가 담당. W-K 의 실내 첨부 # 확장은 W-M OFF 일 때만 유지. if _wm_on and _is_interior_render(spec): return None loc = str(spec.get("loc_id") or "") if _wk_on: m = _membership_by_loc.get(loc) gid = str((m or {}).get("group_id") or "") else: lk = _bfp_by_loc.get(loc) gid = str((lk or {}).get("group_id") or "") reg = _anchor_by_group.get(gid) if gid else None if not reg or reg.get("bg_id") == bid: return None p = Path(str(reg.get("png_path") or "")) if not p.is_file(): return None existing = { str(x) for x in existing_ref_paths if x is not None } if str(p) in existing: return None return {"bg_id": reg["bg_id"], "group_id": gid, "path": p} def _loc_anchor_for( bid: str, spec: Dict[str, Any], existing_ref_paths: List[Optional[Path]], ) -> Optional[Dict[str, Any]]: """W-K: 이 bg 렌더에 첨부할 같은 loc anchor (자기/중복 경로 제외). 반환: ``{"bg_id", "loc_id", "path": Path}`` 또는 None. dedup — anchor png 가 이미 ref 목록(자기 fp/substrate prior/depends_on_bg prior)에 있으면 None (같은 loc prior 체인과의 이중 첨부 방지). """ if not _wk_on: return None loc = str(spec.get("loc_id") or "") reg = _loc_anchor_by_loc.get(loc) if loc else None if not reg or reg.get("bg_id") == bid: return None # E2E9 육안 #3 (2026-07-19, W-M ① 대칭): **실내 렌더** anchor 를 # **실외 렌더**(surface_role 비 interior_room, 예: 마트 외벽 # site_surface plate)에 첨부 금지 — 실내 사진이 외벽 plate 를 # 실내처럼 오염(L09B03 실측). 실내→실내/실외→any 는 기존 유지. if _wm_on and reg.get("interior") and not _is_interior_render( spec): return None p = Path(str(reg.get("png_path") or "")) if not p.is_file(): return None existing = { str(x) for x in existing_ref_paths if x is not None } if str(p) in existing: return None return {"bg_id": reg["bg_id"], "loc_id": loc, "path": p} def _is_interior_render(spec: Dict[str, Any]) -> bool: """W-M: '실내 렌더' 판정 — bg ``surface_role`` 구조 필드(SOT). interior_room(또는 필드 부재 = interior 기본 — W21B-wave-2 의 direct-plate 분기와 동일 규칙)만 실내 렌더로 본다. 실내 분류 loc 에 걸린 exterior_plate/transition_zone/site_surface bg (예: 실내 anchor 건물의 저녁 외관 plate)는 실내 렌더가 아니다 — loc is_indoor 로 판정하면 그런 외관 plate 의 anchor 연속이 끊긴다(간판/외관 연속 회귀). """ role = ( spec.get("surface_role", "interior_room") or "interior_room" ) return role == "interior_room" def _interior_text_guidance_for(spec: Dict[str, Any]) -> str: """W-M(1): mixed 그룹 **실내 렌더** bg 의 문구-only 연속 계약. 그룹 anchor(외부 렌더) 첨부에서 실내 렌더가 제외된 뒤, 실내 plate 의 실내↔실외 연속성은 이미지 ref 없이 이 문구만으로 유지한다. 판정 = 실내 렌더(surface_role) ∧ mixed 그룹 멤버십 (구조 데이터) — W-M ∧ W-K ON 일 때만(멤버십 맵이 W-K 로더 산출). """ if not (_wm_on and _wk_on): return "" if not _is_interior_render(spec): return "" m = _membership_by_loc.get(str(spec.get("loc_id") or "")) if m and m.get("group_mixed"): return INTERIOR_SAME_BUILDING_TEXT_GUIDANCE return "" # ── W-L (2026-07-03): 야외 loc 실사형 aerial establishing SOT ── # 큐 시작 시(어느 lane 실행보다 먼저) renderable bg 를 가진 야외 loc # (aerial 컨텍스트 is_indoor==False 구조판정)마다 aerial 1장을 # 생성/재사용한다. 이후 야외 bg 렌더는 fp(평면도) ref 를 제외하고 # aerial 을 1순위 ref 로 받는다(_aerial_for). aerial 생성 실패 loc 은 # 등록되지 않아 해당 loc 의 렌더가 기존 fp 경로 그대로 진행(fail-safe). _wl_on = bool(getattr( settings, "outdoor_aerial_reference_enabled", False)) _aerial_ctx = aerial_loc_context or {} # loc_id → {"png_path", "cached", "building_fp_used", "fp_asset_id"} # (ok 산출만 — 첨부 게이트). _aerial_by_loc: Dict[str, Dict[str, Any]] = {} # 2026-07-19 재설계 A: 야외 loc 공간·룩 참조=장소 seed(정면 실사, # aerial 대체 — 사용자 확정). flag ON+seed CP ok 그룹만 매핑, # 결손=기존 경로 fail-safe(비차단). _seed_plate_by_loc: Dict[str, Dict[str, str]] = {} if bool(getattr( settings, "outdoor_seed_plate_reference_enabled", False)): try: from app.modules.pipeline.outdoor_structure_seed import ( build_seed_plate_refs, ) _seed_cp = self._load_prev_checkpoint( "outdoor_structure_seed") _bgc_cp = self._load_prev_checkpoint("background_classify") _seed_plate_by_loc = build_seed_plate_refs( ((_seed_cp or {}).get("data", {}) or {}).get( "groups") or {}, ((_bgc_cp or {}).get("data", {}) or {}).get( "building_groups") or [], ) if _seed_plate_by_loc: logger.info( "background_render: seed 플레이트 참조 %d loc — %s", len(_seed_plate_by_loc), sorted(_seed_plate_by_loc), ) except Exception as exc: # noqa: BLE001 — 비차단 fail-safe logger.warning( "background_render: seed 플레이트 컨텍스트 로드 실패" " (비차단): %s", exc) _seed_plate_by_loc = {} _aerial_diag: Dict[str, Any] = ( aerial_diag_out if aerial_diag_out is not None else {} ) if _wl_on and _aerial_ctx: _wl_locs: set = set() for _b in direct_plate_bids: _wl_locs.add( str((bg_specs.get(_b) or {}).get("loc_id") or "")) for _bids in bgs_by_fp.values(): for _b in _bids: _wl_locs.add( str((bg_specs.get(_b) or {}).get("loc_id") or "")) # ★8차 정정: 산출 단위 = loc 가 아니라 **같은 장소 그룹** # (building_groups group_id) — 같은 그룹의 야외 loc 들(renderable # bg 보유)이 통합 배치도 1장을 공유한다. 구조판정: is_indoor 가 # 명시적 False 인 loc 만 그룹에 접는다(실내 loc 불변). _wl_groups: Dict[str, Dict[str, Any]] = {} for _loc in sorted(_wl_locs): ctx = _aerial_ctx.get(_loc) if not ctx or ctx.get("is_indoor") is not False: continue _gid = str(ctx.get("group_id") or "") if not _gid: continue _g = _wl_groups.setdefault(_gid, {"locs": [], "ctx": ctx}) _g["locs"].append(_loc) for _gid in sorted(_wl_groups): _grp_locs: List[str] = _wl_groups[_gid]["locs"] ctx = _wl_groups[_gid]["ctx"] # Codex 8차 MINOR: 파일 세그먼트는 group_id 를 그대로 쓰되 # 안전 문자 집합([A-Za-z0-9_-])이 아니면 해시로 평탄화 — # '/' 등 유입 시 nested path 로 producer 가 조용히 실패하는 # 경로 차단. 원본 group_id 는 diag/variant_type 에 보존. if re.fullmatch(r"[A-Za-z0-9_\-]+", _gid): _fn_gid = _gid else: import hashlib as _hashlib _fn_gid = "h" + _hashlib.sha256( _gid.encode("utf-8")).hexdigest()[:16] logger.warning( "background_render: aerial group id %r 비안전 문자 " "— 파일명 해시 평탄화 %s", _gid, _fn_gid) aerial_out = image_dir / f"aerial_{_fn_gid}.png" try: aerial_out.resolve().relative_to(image_dir_resolved) except (ValueError, OSError): logger.error( "background_render: aerial out_path escapes " "image_dir for group %s — skip", _gid) continue # mixed 그룹이면 같은 건물 indoor fp 를 I2I ref 로 footprint # 정합 (W-G 링크 맵 재사용 — 그룹 야외 loc 중 링크 보유 첫 # 것, png 실재는 로더가 검증). lk = None if ctx.get("group_mixed"): for _l in _grp_locs: if _bfp_by_loc.get(_l): lk = _bfp_by_loc.get(_l) break _fp_ref_path: Optional[Path] = None if lk is not None: _p = Path(str(lk.get("png_path") or "")) if _p.is_file(): _fp_ref_path = _p # Codex 8차 NARROW: 캐시 재사용 판정에 **입력 구조 컨텍스트 # 해시**를 함께 접는다 — 같은 group_id/버전이라도 그룹 멤버 # (label/summary/is_indoor)·anchor_loc·mixed fp 링크가 바뀐 # resume 에서 낡은 배치도가 부활하지 않게. _ctx_hash = compute_aerial_context_hash( group_id=_gid, members=list(ctx.get("group_members") or []), anchor_loc=str(ctx.get("anchor_loc") or ""), fp_sot=str((lk or {}).get("fp_id") or ""), ) # Codex NARROW(7차 정정): cached 재사용은 이전 run 의 aerial # 이 **같은 prompt_version + 같은 context_hash** 산출일 때만 # — 프롬프트 의미가 바뀌거나(실사→도면형, loc→그룹 통합) # 그룹 구조 데이터가 바뀐 뒤 이전 의미의 PNG 가 resume 에서 # 위상 SOT 로 부활하는 경로 차단. 이전 cp 에 기록이 없거나 # 다르면 mode 무관 재생성. _prior = (prior_location_aerials or {}).get(_gid) or {} _prior_ver = str(_prior.get("prompt_version") or "") _prior_ctx = str(_prior.get("context_hash") or "") if ( aerial_out.exists() and mode != "force" and _prior_ver == AERIAL_PROMPT_VERSION and _prior_ctx == _ctx_hash ): # 그룹당 1장 캐시 — resume/부분 재실행은 기존 aerial # 재사용(위상 SOT 안정성). force 만 재생성. 진단의 fp # 필드는 이 run 의 렌더 여부가 아니라 **현재 구조 SOT** # (mixed 그룹 fp 링크) 기준(Codex NARROW). for _l in _grp_locs: _aerial_by_loc[_l] = { "png_path": str(aerial_out), "cached": True, "group_id": _gid, } _aerial_diag[_gid] = { "status": "ok", "png_path": str(aerial_out), "cached": True, "building_fp_used": bool(_fp_ref_path), "fp_asset_id": ( str(lk.get("fp_asset_id") or "") if lk else ""), "prompt_version": AERIAL_PROMPT_VERSION, "context_hash": _ctx_hash, "locs": list(_grp_locs), "anchor_loc": str(ctx.get("anchor_loc") or ""), } continue info = render_location_aerial( openai_client=client, image_model="gpt-image-2.5-sunburst", place_id=_gid, members=list(ctx.get("group_members") or []), out_path=aerial_out, building_fp_path=_fp_ref_path, capture_input_image_ids=( [str(lk.get("fp_asset_id"))] if lk and lk.get("fp_asset_id") else None ), ) if info.get("status") == "ok" and info.get("png_path"): for _l in _grp_locs: _aerial_by_loc[_l] = { "png_path": info["png_path"], "cached": False, "group_id": _gid, } _aerial_diag[_gid] = { "status": "ok", "png_path": info["png_path"], "cached": False, "building_fp_used": bool( info.get("building_fp_used")), "fp_asset_id": ( str(lk.get("fp_asset_id") or "") if lk else ""), "attempts": info.get("attempts", 0), "prompt_used": info.get("prompt_used", ""), "prompt_version": str( info.get("prompt_version") or ""), "context_hash": _ctx_hash, "locs": list(_grp_locs), "anchor_loc": str(ctx.get("anchor_loc") or ""), } logger.info( "background_render: aerial 생성 group=%s locs=%s " "(fp_ref=%s)", _gid, _grp_locs, bool(_fp_ref_path)) else: # 비차단 — 이 그룹 loc 들의 렌더는 기존 fp 경로 그대로. _aerial_diag[_gid] = { "status": "failed", "cached": False, "building_fp_used": bool( info.get("building_fp_used")), "attempts": info.get("attempts", 0), "error": str( info.get("final_block_reason") or "")[:200], "locs": list(_grp_locs), "anchor_loc": str(ctx.get("anchor_loc") or ""), } logger.warning( "background_render: aerial 생성 실패 group=%s — " "기존 fp 경로로 진행(fail-safe): %s", _gid, info.get("final_block_reason")) def _aerial_for(spec: Dict[str, Any]) -> Optional[Dict[str, Any]]: """이 bg 렌더에 1순위로 첨부할 그룹 공간·룩 참조 (야외 loc 만). 2026-07-19 재설계 A: **장소 seed(정면 실사)가 우선** — 사용자 확정(위성/항공 배치도 형태 참조 금지). seed 없거나 flag OFF 면 기존 aerial 경로(레거시, 별도 flag). 반환: ``{"loc_id", "group_id", "path": Path, "kind": "seed"|"aerial"}`` 또는 None. 호출부는 kind 로 guidance 절을 분기한다. """ loc = str(spec.get("loc_id") or "") if loc and _seed_plate_by_loc: sreg = _seed_plate_by_loc.get(loc) if sreg: sp = Path(str(sreg.get("path") or "")) if sp.is_file(): return { "loc_id": loc, "group_id": str(sreg.get("group_id") or ""), "path": sp, "kind": "seed", } if not _wl_on: return None reg = _aerial_by_loc.get(loc) if loc else None if not reg: return None p = Path(str(reg.get("png_path") or "")) if not p.is_file(): return None return { "loc_id": loc, "group_id": str(reg.get("group_id") or ""), "path": p, "kind": "aerial", } def _render_direct_plate( bid: str, *, mode_label: str, decision_fp_id: str = "", degrade_meta: Optional[Dict[str, Any]] = None, ) -> Tuple[Dict[str, Any], str, List[str]]: """Render one bg via the direct-plate (text_only + prior-bg) path and build its group entry. Shared by the surface-role direct-plate lane and the TASK3-B missing-shot_aware_plan fallback. Returns ``(entry, png_path, source_bg_stems)`` — ``png_path`` is '' when the render did not succeed (failed / rejected path); the caller owns ``rendered_paths`` registration, catalog growth, ``ingestion_counter`` and ``failed``. ``mode_label`` stamps ``shot_aware_plan_mode`` + ``reference_decision.mode``; ``degrade_meta`` (optional) is merged into the entry for visible graceful-degradation marking. """ spec = bg_specs.get(bid) or {} prompt_entry = prompts_map.get(bid, {}) or {} t2i_prompt = prompt_entry.get("t2i_prompt", "") surface_role = spec.get("surface_role", "") or "" prior_bg_paths: List[Path] = [] for dep in spec.get("depends_on_bg") or []: p = rendered_paths.get(dep) if p is not None and p.exists(): prior_bg_paths.append(p) out_path = image_dir / f"{bid}.png" try: out_path.resolve().relative_to(image_dir_resolved) except (ValueError, OSError): logger.error( "background_render: out_path escapes image_dir for %s", bid ) rejected = self._build_opt_in_failed_entry( bid=bid, spec=spec, prompt_entry=prompt_entry, camera_recs_by_bg=camera_recs_by_bg, error="out_path escapes image_dir", status="rejected_path", ) return rejected, "", [] # W-G: same-building indoor fp 를 구조 참조로 첨부(outdoor plate 가 # 건물 규모/개구부를 근거 없이 상상하는 결함 억제). 링크 없으면 # 기존 text_only 경로 byte-identical. bfp = _building_fp_for(spec) effective_prompt = t2i_prompt # W-L: 야외 loc 의 aerial establishing 을 위상 SOT 로 1순위 첨부 # (direct lane 은 fp 가 원래 없음 — aerial 이 첫 ref). 등록 없으면 # 기존 경로 byte-identical. aerial = _aerial_for(spec) if _wm_on and aerial is not None: # W-M(2): 실외 도면(aerial)과 실내 도면(building fp)을 한 # 렌더에 동시 첨부 금지(사용자 계약) — fp 정합은 aerial # 생성 I2I 가 흡수. aerial 없으면 기존 fail-safe 그대로. bfp = None bfp_path: Optional[Path] = None if bfp is not None: bfp_path = Path(bfp["png_path"]) aerial_path: Optional[Path] = None if aerial is not None: aerial_path = aerial["path"] effective_prompt = effective_prompt + ( SEED_SITE_GUIDANCE if aerial.get("kind") == "seed" else AERIAL_SITE_GUIDANCE) # W-K: 같은 loc anchor 렌더를 장소 정체성 ref 로 첨부(prior 다음· # building anchor 앞 — render_one_background 계약). sp_anchor = _loc_anchor_for( bid, spec, [aerial_path, *prior_bg_paths, bfp_path]) sp_anchor_path: Optional[Path] = None if sp_anchor is not None: sp_anchor_path = sp_anchor["path"] effective_prompt = ( effective_prompt + SAME_PLACE_PLATE_GUIDANCE ) # W-I: 같은 building 그룹 anchor 렌더를 시각 정체성 ref 로 첨부 # (loc anchor 다음·building fp 앞). loc anchor 와 같은 png 면 # 경로 dedup 이 걸러 1장만 남는다(cap 2 의 "동일시 1"). anchor = _building_anchor_for( bid, spec, [aerial_path, *prior_bg_paths, bfp_path, sp_anchor_path]) anchor_path: Optional[Path] = None if anchor is not None: anchor_path = anchor["path"] effective_prompt = effective_prompt + ( BUILDING_INTERIOR_EXTERIOR_ANCHOR_GUIDANCE if _wk_on else BUILDING_ANCHOR_PLATE_GUIDANCE ) if bfp is not None: effective_prompt = ( effective_prompt + BUILDING_FP_PLATE_GUIDANCE ) # W-M(1) Codex MINOR: missing-plan degrade(TASK3-B)로 mixed # 그룹 **실내** bg 가 direct 경로로 렌더될 때도 문구-only 연속 # 계약을 유지한다(anchor 는 이미 실내 게이트로 차단됨). 통상 # direct lane 은 non-interior surface 라 "" (no-op). effective_prompt = ( effective_prompt + _interior_text_guidance_for(spec) ) info = render_one_background( openai_client=client, image_model="gpt-image-2.5-sunburst", prompt=effective_prompt, out_path=out_path, fp_path=None, prior_bg_paths=prior_bg_paths, bg_id=bid, # 4회차 E2E 실측(2026-07-11): 상태 bg(예: 시신 상태)가 OpenAI # safety 에 거부되면 산출 0 — 렌더 모듈의 moderation-전용 # sanitize 재시도(1회)가 살도록 2. 정상 경로는 여전히 1콜 # (W20 계약 유지 — 2번째 콜은 moderation 거부 시에만 발생). max_attempts=2, building_fp_path=bfp_path, building_anchor_path=anchor_path, same_place_anchor_path=sp_anchor_path, aerial_ref_path=aerial_path, plate_multiroll_ctx=self._plate_multiroll_ctx(), capture_input_image_ids=( [bfp["fp_asset_id"]] if bfp is not None and bfp.get("fp_asset_id") else None ), ) status = info.get("status", "failed") png_path = info.get("png_path", "") if status == "ok" else "" ref_used = info.get("ref_used", "text_only") source_bg_stems = [p.stem for p in prior_bg_paths] entry = { "status": status, "location_id": spec.get("loc_id", ""), "location_name": spec.get("sub_location", ""), "png_path": png_path, "t2i_prompt": t2i_prompt, "shot_guides": prompt_entry.get("shot_guides", []) or [], "shot_ids": spec.get("applies_to_shots", []) or [], "scenes": [], "parent_id": (spec.get("depends_on_bg") or [""])[0], "ref_used": ref_used, "render_attempts": info.get("attempts", 0), "render_error": info.get("final_block_reason") or "", "variant_label": spec.get("state_label", ""), "floor_plan_used": False, "camera_recommendations": camera_recs_by_bg.get(bid, {}), "reference_decision": { "mode": mode_label, "surface_role": surface_role, "fp_id": decision_fp_id, "source_bg_ids": source_bg_stems, }, "effective_render_prompt": effective_prompt, "shot_aware_plan_mode": mode_label, # 무인 계약 감사 3필드 CP 영속 (site 1 과 동일 계약) **_unmanned_audit_fields(info), "attached_reference_lineage": _build_attached_reference_lineage( fp_id=decision_fp_id, fp_path=None, prior_bg_paths=prior_bg_paths, ref_used=ref_used, building_fp_id=(bfp or {}).get("fp_id", ""), building_fp_path=bfp_path, building_anchor_bg_id=(anchor or {}).get("bg_id", ""), building_anchor_path=anchor_path, same_loc_anchor_bg_id=(sp_anchor or {}).get("bg_id", ""), same_loc_anchor_path=sp_anchor_path, aerial_loc_id=(aerial or {}).get("group_id", ""), aerial_path=aerial_path, ), } if aerial is not None and info.get("aerial_ref_attached"): # W-L 구조 필드 — DB sync 가 input_image_ids lineage 로 resolve. entry["aerial_ref"] = { "loc_id": aerial["loc_id"], "group_id": aerial["group_id"], } if bfp is not None and info.get("building_fp_attached"): # 구조 필드 — DB sync 가 input_image_ids lineage 로 resolve. entry["building_fp_ref"] = { "fp_id": bfp["fp_id"], "indoor_loc_sid": bfp.get("indoor_loc_sid", ""), "group_id": bfp.get("group_id", ""), "fp_asset_id": bfp.get("fp_asset_id", ""), } if anchor is not None and info.get("building_anchor_attached"): # W-I 구조 필드 — DB sync 가 input_image_ids lineage 로 resolve. entry["building_anchor_ref"] = { "anchor_bg_id": anchor["bg_id"], "group_id": anchor["group_id"], } if sp_anchor is not None and info.get("same_place_anchor_attached"): # W-K 구조 필드 — DB sync 가 input_image_ids lineage 로 resolve. entry["same_loc_anchor_ref"] = { "anchor_bg_id": sp_anchor["bg_id"], "loc_id": sp_anchor["loc_id"], } if degrade_meta: entry.update(degrade_meta) _register_building_anchor(bid, spec, png_path) _register_loc_anchor(bid, spec, png_path) return entry, png_path, source_bg_stems def _run_direct_plate_lane() -> None: """direct-plate lane — fp 없는 non-interior surface bg 렌더. W-K OFF: 기존 위치(fp lane 앞)에서 즉시 실행(byte-identical). W-K ON: fp lane **뒤**로 재배열 — 외부 establishing(fp lane 의 outdoor)이 먼저 렌더되어 loc/그룹 anchor 의 기준이 되도록. """ nonlocal failed, ingestion_counter for bid in direct_plate_bids: if bid in _stage_chain_handled: # W-M: 그룹 체인 lane 이 이미 렌더/처리 — 중복 방지. continue entry, png_path, source_bg_stems = _render_direct_plate( bid, mode_label="surface_role_direct_plate", ) groups_out[bid] = entry if entry.get("status") == "ok" and png_path: rendered_paths[bid] = Path(png_path) catalog_dump.append({ "bg_id": bid, "fp_id": "", "png_path": png_path, "mode": "surface_role_direct_plate", "node_index": None, "source_bg_ids": source_bg_stems, "ingestion_order": ingestion_counter, "source_kind": "surface_role_direct_plate", }) ingestion_counter += 1 else: failed += 1 # ── W-M (2026-07-03): 야외 같은 장소 그룹 plate 단계화 체인 lane ── # aerial ok 그룹의 야외 render_new_plate bg 를 기존 lane 에서 추출해 # 그룹 단위로 **어느 lane 보다 먼저** 순차 체인 렌더한다(이후 lane 은 # _stage_chain_handled 로 skip — 이중 렌더 방지). ref 구성(사용자 # 계약): 첫 plate = 그룹 통합 배치도(aerial) 1장만, 이후 = 이전 ok # plate + 배치도(최대 2장). 자기 fp/substrate prior/anchor/building # fp 는 첨부하지 않는다(도면 정합은 배치도 생성 I2I 가 흡수, 렌더 # 연속은 이전 plate 체인이 담당). 생성 순서는 LLM(background_plate_ # order) — 순열+depends_on 결정론 검증, 실패 시 구조 순서 fallback. # reuse alias/plan-미비/renderable 밖 bg 는 기존 lane 에 남는다. # 상태변형(depends_on_bg)은 그 base ok plate 를 prev 로 우선 선택. _stage_chain_handled: set = set() if _wm_on and _wl_on and _aerial_by_loc: def _chain_gid_for(spec: Dict[str, Any]) -> str: reg = _aerial_by_loc.get(str(spec.get("loc_id") or "")) return str((reg or {}).get("group_id") or "") if reg else "" # 후보 수집 — 구조 순서(결정론, LLM fallback 순서로도 사용): # fp lane(bgs_by_fp 원 순서 × plan node 순) → direct lane 순. _chain_cands: Dict[str, List[Dict[str, Any]]] = {} _chain_seen: set = set() for _c_fp_id, _c_bids in bgs_by_fp.items(): _c_plan = plans_per_fp.get(_c_fp_id) if not is_plan_production_clear(_c_plan): continue try: _c_nodes = ordered_nodes(_c_plan) except ShotAwareRenderAdapterError: continue _c_bid_set = set(_c_bids) for _c_node in _c_nodes: _c_bid = _c_node.get("bg_id") if ( not isinstance(_c_bid, str) or not _c_bid or _c_bid not in renderable_set or _c_bid not in _c_bid_set or _c_bid in _chain_seen ): continue if _c_node.get( "render_action", "render_new_plate" ) != "render_new_plate": continue _c_gid = _chain_gid_for(bg_specs.get(_c_bid) or {}) if not _c_gid: continue _chain_seen.add(_c_bid) _chain_cands.setdefault(_c_gid, []).append({ "bid": _c_bid, "kind": "node", "fp_id": _c_fp_id, "node": _c_node, }) for _c_bid in direct_plate_bids: if _c_bid in _chain_seen: continue _c_gid = _chain_gid_for(bg_specs.get(_c_bid) or {}) if not _c_gid: continue _chain_seen.add(_c_bid) _chain_cands.setdefault(_c_gid, []).append({ "bid": _c_bid, "kind": "direct", "fp_id": "", "node": None, }) for _c_gid in sorted(_chain_cands): cands = _chain_cands[_c_gid] cand_ids = [c["bid"] for c in cands] cand_by_bid = {c["bid"]: c for c in cands} cand_id_set = set(cand_ids) # LLM 순서 결정 입력 — 구조 필드 + t2i 전문(자르기 금지). _order_cands: List[Dict[str, Any]] = [] for c in cands: _o_spec = bg_specs.get(c["bid"]) or {} _o_ctx = _aerial_ctx.get( str(_o_spec.get("loc_id") or "")) or {} _order_cands.append({ "bg_id": c["bid"], "location_label": str(_o_ctx.get("label") or ""), "location_summary": str(_o_ctx.get("summary") or ""), "surface_role": str( _o_spec.get("surface_role") or ""), "state_label": str(_o_spec.get("state_label") or ""), "depends_on": [ d for d in (_o_spec.get("depends_on_bg") or []) if d in cand_id_set ], "t2i_prompt": ( prompts_map.get(c["bid"], {}) or {} ).get("t2i_prompt", ""), }) try: from app.modules.llm.llm_client import call_structured _order_res = decide_group_plate_order( group_id=_c_gid, candidates=_order_cands, call_structured_fn=call_structured, project_config=self.project_config, opik_metadata=self.build_opik_metadata( extra_metadata={"plate_order_group": _c_gid}), ) except Exception as exc: # noqa: BLE001 — 비차단 fallback logger.warning( "background_render: plate order 결정 실패 group=%s " "— 구조 순서 fallback: %s", _c_gid, exc) _order_res = { "order": list(cand_ids), "order_source": "fallback_structural", "first_reason": "", "error": str(exc)[:200], } _c_order = [ b for b in (_order_res.get("order") or cand_ids) if b in cand_by_bid ] logger.info( "background_render: stage chain group=%s order=%s " "(source=%s)", _c_gid, _c_order, _order_res.get("order_source")) _g_diag = _aerial_diag.get(_c_gid) if isinstance(_g_diag, dict): _g_diag["stage_chain"] = { "order": list(_c_order), "order_source": str( _order_res.get("order_source") or ""), "first_reason": str( _order_res.get("first_reason") or ""), "error": str(_order_res.get("error") or ""), } _chain_ok_paths: Dict[str, Path] = {} _last_ok_bid = "" for _c_idx, bid in enumerate(_c_order): cand = cand_by_bid[bid] _stage_chain_handled.add(bid) spec = bg_specs.get(bid) or {} prompt_entry = prompts_map.get(bid, {}) or {} t2i_prompt = prompt_entry.get("t2i_prompt", "") aerial = _aerial_for(spec) decision = None if cand["kind"] == "node": try: decision = materialize_decision( node=cand["node"], fp_id=cand["fp_id"], base_fp_png_str=( fp_paths_str.get(cand["fp_id"]) or None ), # 체인 lane 은 decision 을 프롬프트 prefix/ # audit 용으로만 소비 — ref 해석은 W-M 체인 # 이 소유(미해결 ref 는 graceful skip). catalog={ b: str(p) for b, p in _chain_ok_paths.items() }, validate_reference_catalog=False, # E2E11 ②: 구조 사실+VIEW AUTHORITY prefix structure_facts=_facts_for( bid, cand["fp_id"]), ) except ShotAwareRenderAdapterError as exc: logger.error( "background_render stage chain decision " "failed bg=%s: %s", bid, exc) groups_out[bid] = self._build_opt_in_failed_entry( bid=bid, spec=spec, prompt_entry=prompt_entry, camera_recs_by_bg=camera_recs_by_bg, error=f"shot_aware_plan: {exc}", ) failed += 1 continue # prev 선택 — 상태변형(depends_on_bg)의 base 가 이 체인 # 에서 이미 ok 면 그것을, 아니면 직전 ok plate. _prev_bid = "" _prev_source = "" for _dep in (spec.get("depends_on_bg") or []): if _dep in _chain_ok_paths: _prev_bid = str(_dep) _prev_source = "depends_on_bg" break if not _prev_bid and _last_ok_bid: _prev_bid = _last_ok_bid _prev_source = "chain" _prev_path = ( _chain_ok_paths.get(_prev_bid) if _prev_bid else None ) out_path = image_dir / f"{bid}.png" try: out_path.resolve().relative_to(image_dir_resolved) except (ValueError, OSError): logger.error( "background_render: out_path escapes image_dir " "for %s", bid) groups_out[bid] = self._build_opt_in_failed_entry( bid=bid, spec=spec, prompt_entry=prompt_entry, camera_recs_by_bg=camera_recs_by_bg, error="out_path escapes image_dir", status="rejected_path", ) failed += 1 continue if decision is not None: effective_prompt = ( decision.reference_guidance_prefix + t2i_prompt ) else: effective_prompt = t2i_prompt effective_prompt = effective_prompt + ( SEED_SITE_GUIDANCE if aerial.get("kind") == "seed" else AERIAL_SITE_GUIDANCE) _priors: List[Path] = [] if _prev_path is not None and _prev_path.exists(): _priors = [_prev_path] effective_prompt = ( effective_prompt + SAME_PLACE_PLATE_GUIDANCE ) info = render_one_background( openai_client=client, image_model="gpt-image-2.5-sunburst", prompt=effective_prompt, out_path=out_path, fp_path=None, prior_bg_paths=_priors, bg_id=bid, # 4회차 E2E 실측(2026-07-11): 상태 bg(예: 시신 상태)가 OpenAI # safety 에 거부되면 산출 0 — 렌더 모듈의 moderation-전용 # sanitize 재시도(1회)가 살도록 2. 정상 경로는 여전히 1콜 # (W20 계약 유지 — 2번째 콜은 moderation 거부 시에만 발생). max_attempts=2, aerial_ref_path=( aerial["path"] if aerial is not None else None ), plate_multiroll_ctx=self._plate_multiroll_ctx(), ) status = info.get("status", "failed") png_path = ( info.get("png_path", "") if status == "ok" else "" ) ref_used_audit = info.get("ref_used", "text_only") entry = { "status": status, "is_reuse": False, "render_action": "render_new_plate", "location_id": spec.get("loc_id", ""), "location_name": spec.get("sub_location", ""), "png_path": png_path, "t2i_prompt": t2i_prompt, "shot_guides": ( prompt_entry.get("shot_guides", []) or [] ), "shot_ids": spec.get("applies_to_shots", []) or [], "scenes": [], "parent_id": (spec.get("depends_on_bg") or [""])[0], "ref_used": ref_used_audit, "render_attempts": info.get("attempts", 0), "render_error": info.get("final_block_reason") or "", "variant_label": spec.get("state_label", ""), "floor_plan_used": False, "camera_recommendations": camera_recs_by_bg.get( bid, {}), "effective_render_prompt": effective_prompt, "attached_reference_lineage": ( _build_attached_reference_lineage( fp_id=cand["fp_id"], fp_path=None, prior_bg_paths=_priors, ref_used=ref_used_audit, aerial_loc_id=( (aerial or {}).get("group_id", "")), aerial_path=( aerial["path"] if aerial is not None else None ), ) ), # W-M 구조 필드 — 체인 위치/입력 audit(진단·캔버스). "stage_chain": { "group_id": _c_gid, "chain_index": _c_idx, "prev_bg_id": _prev_bid, "prev_source": _prev_source, "order_source": str( _order_res.get("order_source") or ""), }, # 무인 계약 감사 3필드 CP 영속 (site 1·2 동일 계약) **_unmanned_audit_fields(info), } if decision is not None: entry["reference_decision"] = decision.to_dict() entry["reference_guidance_prefix"] = ( decision.reference_guidance_prefix ) entry["camera_decision"] = dict( decision.camera_decision) entry["shot_aware_plan_node_index"] = ( decision.node_index ) entry["shot_aware_plan_mode"] = decision.mode entry["render_guidance"] = dict( decision.render_guidance) else: entry["reference_decision"] = { "mode": "surface_role_direct_plate", "surface_role": ( spec.get("surface_role", "") or ""), "fp_id": "", "source_bg_ids": ( [_prev_bid] if _prev_bid else []), } entry["shot_aware_plan_mode"] = ( "surface_role_direct_plate" ) if aerial is not None and info.get("aerial_ref_attached"): # W-L 구조 필드 — DB sync 가 input_image_ids 1순위를 # aerial UUID 로 resolve. 야외 bg 는 fp ref 가 # 의도적으로 제외됐음을 함께 표시(node lane 미러). entry["aerial_ref"] = { "loc_id": aerial["loc_id"], "group_id": aerial["group_id"], } if cand["kind"] == "node": entry["fp_ref_replaced_by_aerial"] = True groups_out[bid] = entry if status == "ok" and png_path: rendered_paths[bid] = Path(png_path) _chain_ok_paths[bid] = Path(png_path) _last_ok_bid = bid _register_building_anchor(bid, spec, png_path) _register_loc_anchor(bid, spec, png_path) catalog_dump.append({ "bg_id": bid, "fp_id": cand["fp_id"], "png_path": png_path, "mode": entry.get("shot_aware_plan_mode"), "node_index": entry.get( "shot_aware_plan_node_index"), "source_bg_ids": ( [_prev_bid] if _prev_bid else []), "ingestion_order": ingestion_counter, "source_kind": "stage_chain_plate", }) ingestion_counter += 1 else: failed += 1 # W-K: lane 재배열. 양 lane 내부는 mixed 그룹 outdoor 소속 우선의 # **안정정렬**(같은 키끼리는 원순서 보존 → 같은 loc 의 상대 순서 # 불변 = depends_on_bg same-loc 체인 안전. fp lane 의 substrate/ # reuse 참조는 fp-로컬 catalog 라 fp 그룹 간 순서 재배열과 무관). fp_lane_items = list(bgs_by_fp.items()) if _wk_on: def _is_mixed_outdoor_loc(b: str) -> bool: m = _membership_by_loc.get( str((bg_specs.get(b) or {}).get("loc_id") or "")) return bool( m and m.get("group_mixed") and m.get("is_indoor") is False) direct_plate_bids = sorted( direct_plate_bids, key=lambda b: 0 if _is_mixed_outdoor_loc(b) else 1) fp_lane_items = sorted( fp_lane_items, key=lambda kv: 0 if any( _is_mixed_outdoor_loc(b) for b in kv[1]) else 1) else: _run_direct_plate_lane() # TASK3-B: degrade a MISSING fp plan (plan is None) to the # direct-plate path so a single transiently-dropped fp does not # turn background_render ``partial`` and cascade to a whole-episode # scene_image_pipeline STALE_UPSTREAM block. Structural gate only # (plan-missing + a renderable direct-plate input) — never keyed on # location name / kind. The ``plan not production_clear`` case is # unaffected (always fails closed). missing_plan_fallback = bool( getattr( settings, "background_render_missing_shot_aware_plan_fallback_enabled", False, ) ) for fp_id, fp_bg_ids in fp_lane_items: plan = plans_per_fp.get(fp_id) if not is_plan_production_clear(plan): # W20F11 (2026-07-23 Codex 합의): camera geometry 구조 # 불가 typed not_applicable **만** direct-plate 경로 — # generic not_applicable(provider 미배선 등)/failed 는 # 기존 fail-closed 그대로. reason code 는 producer # (build_render_plan_for_fp)와 단일 SOT. from app.modules.pipeline.shot_aware_bg_render_plan import ( CAMERA_LOOKAT_NOT_APPLICABLE_REASON, ) camera_geometry_na = ( isinstance(plan, dict) and plan.get("shot_aware_bg_render_plan_status") == "not_applicable" and plan.get("not_applicable_reason_code") == CAMERA_LOOKAT_NOT_APPLICABLE_REASON ) if plan is None: reason = ( f"shot_aware_plan: plan missing for fp_id={fp_id!r}" ) elif camera_geometry_na: reason = ( f"shot_aware_plan: camera geometry not plannable " f"(fp_id={fp_id!r} reason_code=" f"{CAMERA_LOOKAT_NOT_APPLICABLE_REASON}) — " "direct-plate route" ) else: reason = ( f"shot_aware_plan: plan not production_clear " f"(fp_id={fp_id!r} " f"status={plan.get('shot_aware_bg_render_plan_status')!r} " f"production_clear={plan.get('production_clear')!r})" ) logger.error( "background_render shot_aware_plan gate failed fp=%s: %s", fp_id, reason, ) for bid in fp_bg_ids: spec = bg_specs.get(bid) or {} prompt_entry = prompts_map.get(bid, {}) or {} if ( camera_geometry_na and (prompt_entry.get("t2i_prompt") or "").strip() ): # typed reason 한정 direct plate — t2i_prompt # 결손은 아래 기존 fail-closed 로 낙하 entry, png_path, source_bg_stems = ( _render_direct_plate( bid, mode_label=( "camera_geometry_direct_plate" ), decision_fp_id="", degrade_meta={ "render_degraded": True, "fallback_reason": ( CAMERA_LOOKAT_NOT_APPLICABLE_REASON ), "source_fp_id": fp_id, }, ) ) groups_out[bid] = entry if entry.get("status") == "ok" and png_path: rendered_paths[bid] = Path(png_path) catalog_dump.append({ "bg_id": bid, "fp_id": "", "source_fp_id": fp_id, "png_path": png_path, "mode": "camera_geometry_direct_plate", "node_index": None, "source_bg_ids": source_bg_stems, "ingestion_order": ingestion_counter, "source_kind": ( "camera_geometry_direct_plate" ), }) ingestion_counter += 1 else: failed += 1 continue # TASK3-B: degrade ONLY a missing plan (plan is None) # AND only when the direct-plate input is present (a # non-empty t2i_prompt). Otherwise stay fail-closed. if ( missing_plan_fallback and plan is None and (prompt_entry.get("t2i_prompt") or "").strip() ): entry, png_path, source_bg_stems = _render_direct_plate( bid, mode_label="missing_plan_direct_plate_fallback", decision_fp_id="", degrade_meta={ "render_degraded": True, "fallback_reason": "shot_aware_plan_missing", "missing_fp_id": fp_id, }, ) groups_out[bid] = entry if entry.get("status") == "ok" and png_path: rendered_paths[bid] = Path(png_path) catalog_dump.append({ "bg_id": bid, "fp_id": "", "png_path": png_path, "mode": "missing_plan_direct_plate_fallback", "node_index": None, "source_bg_ids": source_bg_stems, "ingestion_order": ingestion_counter, "source_kind": "missing_plan_direct_plate_fallback", }) ingestion_counter += 1 else: failed += 1 continue groups_out[bid] = self._build_opt_in_failed_entry( bid=bid, spec=spec, prompt_entry=prompt_entry, camera_recs_by_bg=camera_recs_by_bg, error=reason, ) failed += 1 continue try: nodes = ordered_nodes(plan) except ShotAwareRenderAdapterError as exc: logger.error( "background_render shot_aware_plan node order failed fp=%s: %s", fp_id, exc, ) for bid in fp_bg_ids: spec = bg_specs.get(bid) or {} prompt_entry = prompts_map.get(bid, {}) or {} groups_out[bid] = self._build_opt_in_failed_entry( bid=bid, spec=spec, prompt_entry=prompt_entry, camera_recs_by_bg=camera_recs_by_bg, error=f"shot_aware_plan: {exc}", ) failed += 1 continue plan_bg_ids = { n.get("bg_id") for n in nodes if isinstance(n.get("bg_id"), str) and n.get("bg_id") } for bid in fp_bg_ids: if bid in plan_bg_ids: continue spec = bg_specs.get(bid) or {} prompt_entry = prompts_map.get(bid, {}) or {} groups_out[bid] = self._build_opt_in_failed_entry( bid=bid, spec=spec, prompt_entry=prompt_entry, camera_recs_by_bg=camera_recs_by_bg, error=( f"shot_aware_plan: bg_id={bid!r} missing from " f"plan graph for fp_id={fp_id!r}" ), ) failed += 1 base_fp_str: Optional[str] = fp_paths_str.get(fp_id) or None catalog: Dict[str, str] = {} for node in nodes: bid = node.get("bg_id") if not isinstance(bid, str) or not bid: continue if bid not in renderable_set: # Plan covers a bg that the upstream filter dropped # (e.g. background_prompt status != ok). Do not # synthesize a group entry — the W19F lock is that # only renderable bgs surface in groups_out. continue if bid in _stage_chain_handled: # W-M: 그룹 체인 lane 이 이미 렌더/처리 — 중복 방지. # (체인 plate 는 fp-로컬 catalog 에 넣지 않는다 — 남은 # 노드가 그것을 substrate parent 로 쓰면 graceful drop. # 실내 bg 가 외부 plate 를 참조하는 경로 차단과 정합.) continue spec = bg_specs.get(bid) or {} prompt_entry = prompts_map.get(bid, {}) or {} t2i_prompt = prompt_entry.get("t2i_prompt", "") # W21B-w3 Commit 2b — reuse_existing_plate: the 2a router # decided this bg shares an earlier plate. NO image call / # budget / new PNG; we copy-less-alias the target path. The # entry stays status='ok' + is_reuse=True so downstream scene # loaders (which gate on status=='ok') still resolve it. render_action = node.get("render_action", "render_new_plate") if render_action == "reuse_existing_plate": target_id = node.get("reuse_target_bg_id") or "" target_path = rendered_paths.get(target_id) if target_path is None or not Path(target_path).exists(): # Defensive fail-closed: 2a router forbids targeting a # non-render_new_plate / future node, but the target's # render may still have failed at this step. entry = self._build_opt_in_failed_entry( bid=bid, spec=spec, prompt_entry=prompt_entry, camera_recs_by_bg=camera_recs_by_bg, error=( f"shot_aware_plan: reuse target " f"{target_id!r} has no rendered plate" ), status="reuse_target_missing", ) entry["is_reuse"] = True entry["render_action"] = "reuse_existing_plate" entry["reuse_target_bg_id"] = target_id entry["reused_from_bg_id"] = target_id entry["shot_aware_plan_node_index"] = node.get( "node_index" ) entry["shot_aware_plan_mode"] = node.get("mode") groups_out[bid] = entry failed += 1 continue target_path_str = str(target_path) ref_audit = node.get("reference_decision") or {} groups_out[bid] = { "status": "ok", "is_reuse": True, "render_action": "reuse_existing_plate", "reuse_target_bg_id": target_id, "reused_from_bg_id": target_id, "location_id": spec.get("loc_id", ""), "location_name": spec.get("sub_location", ""), # copy-less alias — same plate path as the target. "png_path": target_path_str, "t2i_prompt": t2i_prompt, "shot_guides": prompt_entry.get("shot_guides", []) or [], "shot_ids": spec.get("applies_to_shots", []) or [], "scenes": [], "parent_id": target_id, "ref_used": "reused_plate", "render_attempts": 0, "render_error": "", "variant_label": spec.get("state_label", ""), "floor_plan_used": False, "camera_recommendations": camera_recs_by_bg.get( bid, {} ), "reference_decision": { "mode": node.get("mode"), "render_action": "reuse_existing_plate", "reuse_target_bg_id": target_id, "selected_refs": ref_audit.get( "selected_refs", [] ), }, "shot_aware_plan_node_index": node.get("node_index"), "shot_aware_plan_mode": node.get("mode"), "render_guidance": dict( node.get("render_guidance") or {} ), "attached_reference_lineage": ( _build_attached_reference_lineage( fp_id=fp_id, fp_path=None, prior_bg_paths=[Path(target_path_str)], ref_used="reused_plate", ) ), } # Surface the aliased plate to later catalog lookups (a # normal reference_derived node may still ref this bg; # reuse-of-reuse itself is router-forbidden). rendered_paths[bid] = Path(target_path_str) catalog[bid] = target_path_str catalog_dump.append({ "bg_id": bid, "fp_id": fp_id, "png_path": target_path_str, "mode": node.get("mode"), "node_index": node.get("node_index"), "source_bg_ids": [target_id], "ingestion_order": ingestion_counter, "source_kind": "shot_aware_plan_reuse", }) ingestion_counter += 1 continue substrate_on = bool(settings.bg_render_substrate_enabled) # W21B-w4 #4(C) Required 2 — substrate ON must fail closed on a # non-schema-9 plan node. is_plan_production_clear does not assert # the 3a/3b node shape, so a stale schema-8 production_clear plan # would otherwise be silently mis-rendered (fresh nodes treated as # no-parent fp_only). Require the mirror/DAG fields before any image. if substrate_on: missing_fields = [ f for f, ok in ( ("needs_new_plate", isinstance(node.get("needs_new_plate"), bool)), ("ref_tree_parents", isinstance(node.get("ref_tree_parents"), list)), ("ref_role_per_parent", isinstance(node.get("ref_role_per_parent"), dict)), ) if not ok ] if missing_fields: groups_out[bid] = self._build_opt_in_failed_entry( bid=bid, spec=spec, prompt_entry=prompt_entry, camera_recs_by_bg=camera_recs_by_bg, error=( "substrate_requires_schema9_plan_fields: " f"node bg_id={bid!r} missing/malformed " f"{missing_fields} — re-run shot_aware_bg_render_plan " f"(SCHEMA_VERSION>=9) before enabling the substrate " f"consumer" ), ) failed += 1 continue try: decision = materialize_decision( node=node, fp_id=fp_id, base_fp_png_str=base_fp_str, catalog=catalog, # substrate owns reference resolution (graceful missing- # parent drop); keep the shape/prefix gate but skip the # legacy catalog path-existence raise so it cannot pre-empt # the substrate fallback. validate_reference_catalog=not substrate_on, # E2E11 ②: 구조 사실+VIEW AUTHORITY prefix structure_facts=_facts_for(bid, fp_id), ) except ShotAwareRenderAdapterError as exc: logger.error( "background_render shot_aware_plan decision failed " "bg=%s: %s", bid, exc, ) groups_out[bid] = self._build_opt_in_failed_entry( bid=bid, spec=spec, prompt_entry=prompt_entry, camera_recs_by_bg=camera_recs_by_bg, error=f"shot_aware_plan: {exc}", ) failed += 1 continue substrate_decision: Optional[Dict[str, Any]] = None if substrate_on: # W21B-w4 #4(C) — resolve the render-input substrate from the # 3b reference DAG (ref_tree_parents canonical rendered-plate # anchors) instead of the adapter's FP|parents either-or refs: # FP first, then parents ordered space_continuity → style. A # parent plate not yet rendered is gracefully dropped (recorded # in fallback_reason) rather than blocking. PNG substrate only — # plate prose stays with the C4 projection-card boundary; the # adapter ``decision`` above is still the prompt-prefix / node # gate SOT. reuse-alias nodes never reach here (handled above). sub = resolve_render_substrate( plan_node=node, fp_id=fp_id, fp_png_by_fp={fp_id: base_fp_str}, rendered_plate_png_by_bg=catalog, ) fp_path: Optional[Path] = ( Path(sub["fp_path"]) if sub["fp_path"] else None ) prior_bg_paths: List[Path] = [ Path(p) for p in sub["ordered_prior_bg_paths"] ] substrate_decision = { k: sub[k] for k in ( "node_class", "ref_used", "fallback_reason", "missing_parent_bg_ids", "ordered_parent_bg_ids", "attached_ref_labels", ) } else: ref_path_objs = [Path(p) for p in decision.reference_paths] missing_refs = [ str(p) for p in ref_path_objs if not p.exists() ] if missing_refs: groups_out[bid] = self._build_opt_in_failed_entry( bid=bid, spec=spec, prompt_entry=prompt_entry, camera_recs_by_bg=camera_recs_by_bg, error=( f"shot_aware_plan: reference path unresolved: " f"{missing_refs}" ), decision_dict=decision.to_dict(), ) failed += 1 continue if decision.fp_included: fp_path = ref_path_objs[0] prior_bg_paths = [] else: fp_path = None prior_bg_paths = ref_path_objs out_path = image_dir / f"{bid}.png" try: out_path.resolve().relative_to(image_dir_resolved) except (ValueError, OSError): logger.error( "background_render: out_path escapes image_dir for %s", bid, ) groups_out[bid] = self._build_opt_in_failed_entry( bid=bid, spec=spec, prompt_entry=prompt_entry, camera_recs_by_bg=camera_recs_by_bg, error="out_path escapes image_dir", status="rejected_path", decision_dict=decision.to_dict(), ) failed += 1 continue effective_prompt = ( decision.reference_guidance_prefix + t2i_prompt ) # W-L: 야외 bg — fp(평면도) ref 를 제외하고 aerial establishing # 을 1순위 ref 로 교체(평면도 위상 지배 제거, aerial=위상 SOT). # substrate prior(LLM 체인)는 그대로 2순위. aerial 미등록 # loc(실내/생성 실패)은 기존 경로 byte-identical. aerial = _aerial_for(spec) aerial_path: Optional[Path] = None if aerial is not None: aerial_path = aerial["path"] fp_path = None effective_prompt = effective_prompt + ( SEED_SITE_GUIDANCE if aerial.get("kind") == "seed" else AERIAL_SITE_GUIDANCE) # W-G: outdoor loc 의 same-building indoor fp 를 추가 구조 # 참조로 첨부(ref 마지막). 자기 fp(shot_aware substrate)와 # 별개 채널 — 링크 없으면 기존 호출 byte-identical. bfp = _building_fp_for(spec) if _wm_on and aerial is not None: # W-M(2): 실외 도면(aerial)과 실내 도면(building fp)을 # 한 렌더에 동시 첨부 금지(사용자 계약) — fp 정합은 # aerial 생성 I2I 가 흡수. bfp = None bfp_path: Optional[Path] = None if bfp is not None: bfp_path = Path(bfp["png_path"]) # W-K: 같은 loc anchor 렌더 첨부(prior 다음·building anchor # 앞). substrate prior 에 이미 같은 png 가 있으면 dedup skip. sp_anchor = _loc_anchor_for( bid, spec, [aerial_path, fp_path, *prior_bg_paths, bfp_path]) sp_anchor_path: Optional[Path] = None if sp_anchor is not None: sp_anchor_path = sp_anchor["path"] effective_prompt = ( effective_prompt + SAME_PLACE_PLATE_GUIDANCE ) # W-I: 같은 building 그룹 anchor 렌더 첨부(loc anchor 다음· # building fp 앞). substrate prior/loc anchor 에 이미 같은 # png 가 있으면 dedup skip(cap 2 의 "동일시 1"). anchor = _building_anchor_for( bid, spec, [aerial_path, fp_path, *prior_bg_paths, bfp_path, sp_anchor_path]) anchor_path: Optional[Path] = None if anchor is not None: anchor_path = anchor["path"] effective_prompt = effective_prompt + ( BUILDING_INTERIOR_EXTERIOR_ANCHOR_GUIDANCE if _wk_on else BUILDING_ANCHOR_PLATE_GUIDANCE ) if bfp is not None: effective_prompt = ( effective_prompt + BUILDING_FP_PLATE_GUIDANCE ) # W-M(1): mixed 그룹 실내 bg — anchor(외부 렌더) 대신 # 문구-only 연속 계약(이미지 ref 없음). 야외/비대상 = "". effective_prompt = ( effective_prompt + _interior_text_guidance_for(spec) ) info = render_one_background( openai_client=client, image_model="gpt-image-2.5-sunburst", prompt=effective_prompt, out_path=out_path, fp_path=fp_path, prior_bg_paths=prior_bg_paths, bg_id=bid, # 4회차 E2E 실측(2026-07-11): 상태 bg(예: 시신 상태)가 OpenAI # safety 에 거부되면 산출 0 — 렌더 모듈의 moderation-전용 # sanitize 재시도(1회)가 살도록 2. 정상 경로는 여전히 1콜 # (W20 계약 유지 — 2번째 콜은 moderation 거부 시에만 발생). max_attempts=2, building_fp_path=bfp_path, building_anchor_path=anchor_path, same_place_anchor_path=sp_anchor_path, aerial_ref_path=aerial_path, plate_multiroll_ctx=self._plate_multiroll_ctx(), capture_input_image_ids=( [bfp["fp_asset_id"]] if bfp is not None and bfp.get("fp_asset_id") else None ), ) status = info.get("status", "failed") png_path = info.get("png_path", "") if status == "ok" else "" ref_used_audit = info.get("ref_used", "text_only") entry = { "status": status, # W21B-w3 Commit 2b — normal (newly-rendered) plate. "is_reuse": False, "render_action": "render_new_plate", "location_id": spec.get("loc_id", ""), "location_name": spec.get("sub_location", ""), "png_path": png_path, "t2i_prompt": t2i_prompt, "shot_guides": prompt_entry.get("shot_guides", []) or [], "shot_ids": spec.get("applies_to_shots", []) or [], "scenes": [], "parent_id": (spec.get("depends_on_bg") or [""])[0], "ref_used": ref_used_audit, "render_attempts": info.get("attempts", 0), "render_error": info.get("final_block_reason") or "", "variant_label": spec.get("state_label", ""), "floor_plan_used": ( fp_path is not None and fp_path.exists() ), "camera_recommendations": camera_recs_by_bg.get(bid, {}), # W20C audit fields. "reference_decision": decision.to_dict(), "reference_guidance_prefix": ( decision.reference_guidance_prefix ), "effective_render_prompt": effective_prompt, "camera_decision": dict(decision.camera_decision), "shot_aware_plan_node_index": decision.node_index, "shot_aware_plan_mode": decision.mode, # W20D: renderer-facing guidance pinned as a top-level # convenience field (also available inside # ``reference_decision`` via decision.to_dict()). "render_guidance": dict(decision.render_guidance), # W21B-wave-1 — checkpoint-only attached reference lineage # diagnostic. shot_aware path 의 ref_path_objs 가 이미 # decision 에서 검증된 실제 path 들이므로 그대로 사용 # (fp_included 일 때만 fp_path 전달, 나머지는 prior bg). "attached_reference_lineage": _build_attached_reference_lineage( fp_id=fp_id, fp_path=fp_path, prior_bg_paths=prior_bg_paths, ref_used=ref_used_audit, building_fp_id=(bfp or {}).get("fp_id", ""), building_fp_path=bfp_path, building_anchor_bg_id=(anchor or {}).get("bg_id", ""), building_anchor_path=anchor_path, same_loc_anchor_bg_id=( sp_anchor or {}).get("bg_id", ""), same_loc_anchor_path=sp_anchor_path, aerial_loc_id=(aerial or {}).get("group_id", ""), aerial_path=aerial_path, ), # 무인 계약 감사 3필드 CP 영속 (site 1·2 동일 계약) **_unmanned_audit_fields(info), } if aerial is not None and info.get("aerial_ref_attached"): # W-L 구조 필드 — DB sync 가 input_image_ids lineage resolve. # 야외 bg 는 fp ref 가 의도적으로 제외됐음을 함께 표시. entry["aerial_ref"] = { "loc_id": aerial["loc_id"], "group_id": aerial["group_id"], } entry["fp_ref_replaced_by_aerial"] = True if bfp is not None and info.get("building_fp_attached"): # W-G 구조 필드 — DB sync 가 input_image_ids lineage 로 resolve. entry["building_fp_ref"] = { "fp_id": bfp["fp_id"], "indoor_loc_sid": bfp.get("indoor_loc_sid", ""), "group_id": bfp.get("group_id", ""), "fp_asset_id": bfp.get("fp_asset_id", ""), } if anchor is not None and info.get("building_anchor_attached"): # W-I 구조 필드 — DB sync 가 input_image_ids lineage resolve. entry["building_anchor_ref"] = { "anchor_bg_id": anchor["bg_id"], "group_id": anchor["group_id"], } if sp_anchor is not None and info.get( "same_place_anchor_attached"): # W-K 구조 필드 — DB sync 가 input_image_ids lineage resolve. entry["same_loc_anchor_ref"] = { "anchor_bg_id": sp_anchor["bg_id"], "loc_id": sp_anchor["loc_id"], } # W21B-w4 #4(C) — JSON-safe substrate audit (only when the # consumer flag is ON; OFF leaves the entry byte-identical). if substrate_decision is not None: entry["substrate_decision"] = substrate_decision groups_out[bid] = entry if status == "ok" and png_path: rendered_paths[bid] = Path(png_path) catalog[bid] = png_path _register_building_anchor(bid, spec, png_path) _register_loc_anchor(bid, spec, png_path) catalog_dump.append({ "bg_id": bid, "fp_id": fp_id, "png_path": png_path, "mode": decision.mode, "node_index": decision.node_index, "source_bg_ids": list(decision.source_bg_ids), "ingestion_order": ingestion_counter, "source_kind": "shot_aware_plan", }) ingestion_counter += 1 else: failed += 1 # W-K: 재배열 시 direct-plate lane 은 fp lane 뒤에서 실행 — 같은 # loc/그룹의 fp lane 렌더(외부 establishing 포함)가 anchor 로 먼저 # 등록된 뒤 direct plate 들이 그것을 첨부받는다. if _wk_on: _run_direct_plate_lane() return groups_out, rendered_paths, failed, catalog_dump def _register_image_assets( self, groups: Dict[str, Dict[str, Any]], order: List[str], location_aerials: Optional[Dict[str, Dict[str, Any]]] = None, ) -> None: """Phase 7 — bg PNG를 ImageAsset(asset_type='chain_bg') UPSERT. Phase 5 ``background_chain_render_step._register_chain_bg_image_assets`` 패턴 미러. variant_index는 loc_id 단위 1+ counter (v00은 floor_plan). deterministic을 위해 ``order`` (master_plan gen_order에서 파생) 순서 대로 counter 증분. match key (application-level): (project_id, episode_id, asset_type, entity_id, variant_type=bg_id). W-L: ``location_aerials`` (loc별 aerial 진단, status=='ok' 만 대상)가 주어지면 aerial PNG 를 ImageAsset(asset_type='location_aerial', variant_type='aerial_') 로 먼저 UPSERT 하고, ``aerial_ref`` 구조 필드를 가진 bg row 의 input_image_ids **1순위** 를 loc 의 floor_plan UUID 대신 aerial UUID 로 교체한다(실제 첨부 SOT 정합 — 야외 bg 는 fp ref 가 의도적으로 제외됨). None(default) = 기존 경로 byte-identical. """ 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( "background_render: no location EntityCanon — skip ImageAsset DB sync" ) return # loc 단위 counter: v00은 floor_plan용 → bg는 1부터 시작. # order(=master_plan gen_order에서 도출)대로 deterministic. location_counter: Dict[str, int] = {} variant_idx_by_bg: Dict[str, int] = {} # 3a recording — 실제 첨부 prior_bg lineage 2-phase 기록. phase1(loop)=bg_id→ # UUID 맵 + 실제첨부 prior_bg_ids pending 수집, phase2(loop 후)=UUID resolve 해 # input_image_ids=[fp,*prior] 재기록. ★기록 정책 flag(생성 flag 와 분리, Codex # NARROW). OFF(default) 면 _star_pending 비어 no-op(byte-identical). _star_enabled = bool(getattr( settings, "background_render_record_prior_bg_lineage_enabled", False)) _star_uuid_by_bgid: Dict[str, Optional[str]] = {} # (row, first_uuid, prior_bg_ids: List[str], reuse: bool, # building_fp_uuid: str, anchors: List[(label_kind, anchor_bg_id)], # pre_unresolved: List[str]) # first_uuid = 1순위 lineage — 기본은 loc floor_plan UUID, W-L aerial # 첨부 bg 는 aerial UUID(phase1 과 동일 교체). W-I/W-K: anchor 는 # chain_bg 라 UUID 가 phase2 resolver 에서만 확정(렌더 순서상 anchor # row 가 loop 뒤에 생길 수 있음) — phase2 전용 병합. anchors 는 첨부 # 순서(same_loc_anchor → building_anchor) 보존 리스트. pre_unresolved # 는 phase1 에서 확정된 미해결 구조키(aerial 등) 이월분. _star_pending: List[ Tuple[Any, Optional[str], List[str], bool, str, List[Tuple[str, str]], List[str]] ] = [] for bid in order: res = groups.get(bid) or {} if res.get("status") != "ok" or not res.get("png_path"): continue loc_short = res.get("location_id", "") or "" if not loc_short: continue next_idx = location_counter.get(loc_short, 1) variant_idx_by_bg[bid] = next_idx location_counter[loc_short] = next_idx + 1 now = datetime.now(timezone.utc).isoformat() projects_root = Path(settings.projects_dir).parent registered = 0 # ── W-L: location aerial ImageAsset UPSERT (bg 보다 먼저 — bg lineage # 가 aerial UUID 를 1순위로 참조). ★8차 정정: 산출 단위=같은 장소 # 그룹 — diag key=group_id, entity=그룹 anchor_loc canon(부재 시 그룹 # 야외 첫 loc). match key = (project, episode, asset_type= # 'location_aerial', entity_id, variant_type='aerial_') — # 재실행 idempotent. None/{} = no-op. _aerial_uuid_by_loc: Dict[str, str] = {} for _gid in sorted((location_aerials or {}).keys()): _ae = (location_aerials or {}).get(_gid) or {} if _ae.get("status") != "ok" or not _ae.get("png_path"): continue _ae_locs = [str(x) for x in (_ae.get("locs") or []) if x] _canon = None for _cand in ([str(_ae.get("anchor_loc") or "")] + _ae_locs): if _cand and canon_id_by_short.get(_cand): _canon = canon_id_by_short[_cand] break if not _canon: logger.warning( "background_render: aerial group %s has no canon — " "skip DB sync", _gid) continue try: _rel = str(Path(str(_ae["png_path"])).relative_to( projects_root)) except ValueError: _rel = str(_ae["png_path"]) _vt = f"aerial_{_gid}" _row = ( self.db.query(ImageAsset) .filter_by( project_id=self.project_id, episode_id=self.episode_id, asset_type="location_aerial", entity_id=_canon, variant_type=_vt, ) .first() ) _is_new_row = _row is None if _row: _row.file_path = _rel # cached 재사용 run 은 prompt_used 미보유 — 기존 값 보존. _pu = str(_ae.get("prompt_used") or "") if _pu: _row.prompt_used = _pu _row.status = "generated" _row.generation_model = "gpt-image-2.5-sunburst" else: _row = ImageAsset( id=str(uuid.uuid4()), project_id=self.project_id, episode_id=self.episode_id, asset_type="location_aerial", entity_id=_canon, variant_index=0, variant_label="aerial", variant_type=_vt, file_path=_rel, prompt_used=str(_ae.get("prompt_used") or ""), generation_model="gpt-image-2.5-sunburst", status="generated", is_primary=0, created_at=now, ) self.db.add(_row) # lineage — mixed 그룹 aerial 은 indoor fp 를 I2I ref 로 생성. # cached 재사용 run 은 기존 row 의 input_image_ids 를 보존(None= # 불변)하되, row 자체가 없으면(파일만 잔존한 fresh DB) 구조 SOT # 기준 lineage 로 신규 기록한다 (Codex NARROW). _ae_fp = str(_ae.get("fp_asset_id") or "") _ae_cached = bool(_ae.get("cached")) annotate_generated_asset( _row, pipeline_role="location_aerial", input_image_ids=( None if (_ae_cached and not _is_new_row) else ( [_ae_fp] if (_ae.get("building_fp_used") and _ae_fp) else [] ) ), pipeline_metadata={ "group_id": _gid, "locs": _ae_locs, "anchor_loc": str(_ae.get("anchor_loc") or ""), "cached": _ae_cached, "building_fp_used": bool(_ae.get("building_fp_used")), # Codex NARROW: 산출 프롬프트 의미 버전+입력 구조 # 컨텍스트 해시 — cached 재사용 검증(큐)과 audit 공용. "prompt_version": str(_ae.get("prompt_version") or ""), "context_hash": str(_ae.get("context_hash") or ""), }, ) # 그룹 통합 aerial 1장 → 그룹 야외 loc 전부가 같은 UUID 를 # 1순위 lineage 로 참조(bg 루프의 aerial_ref.loc_id 조인용). for _l in _ae_locs: _aerial_uuid_by_loc[_l] = _row.id registered += 1 # persist-all Wave1 정련(Codex): FP→bg lineage resolve 를 per-bg N쿼리 → # per-step 캐시맵 1쿼리로. tie-breaker 는 기존과 동일(is_primary desc, # created_at desc) — 같은 canon 의 첫 floor_plan(=primary v00 우선)만 채택. _fp_id_by_canon: Dict[str, str] = {} for _fp in ( self.db.query(ImageAsset) .filter( ImageAsset.project_id == self.project_id, ImageAsset.asset_type == "floor_plan", ) .order_by(ImageAsset.is_primary.desc(), ImageAsset.created_at.desc()) .all() ): if _fp.entity_id and _fp.entity_id not in _fp_id_by_canon: _fp_id_by_canon[_fp.entity_id] = _fp.id for bg_id in order: res = groups.get(bg_id) or {} if res.get("status") != "ok": continue png_path_str = res.get("png_path", "") or "" if not png_path_str: continue loc_short = res.get("location_id", "") or "" canon_id = canon_id_by_short.get(loc_short) if not canon_id: logger.warning( "background_render: bg %s loc %s has no canon — skip DB sync", bg_id, loc_short, ) continue # Phase 3 컨벤션: file_path는 projects_root 기준 relative. try: rel_png_path = str(Path(png_path_str).relative_to(projects_root)) except ValueError: rel_png_path = png_path_str variant_index = variant_idx_by_bg.get(bg_id, 1) variant_label = res.get("variant_label", "") or f"v{variant_index:02d}" # t2i_guide — shot_guides[]의 guide 텍스트를 newline join (no truncation). shot_guides = res.get("shot_guides", []) or [] guide_lines: List[str] = [] for sg in shot_guides: if not isinstance(sg, dict): continue sid = sg.get("shot_id", "") or "" # schema field is "guide_text" — keep "guide" as legacy fallback # (Phase 7 pre-existing field name mismatch surfaced by Phase 8 review) guide = sg.get("guide_text") or sg.get("guide") or "" if guide: if sid: guide_lines.append(f"[{sid}] {guide}") else: guide_lines.append(guide) t2i_guide_text = "\n".join(guide_lines) if guide_lines else None existing = ( self.db.query(ImageAsset) .filter_by( project_id=self.project_id, episode_id=self.episode_id, asset_type="chain_bg", entity_id=canon_id, variant_type=bg_id, ) .first() ) # W19B-3 provenance: opt-in groups carry ``effective_render_prompt`` # (= reference_guidance_prefix + t2i_prompt) — that is the text # actually sent to images.edit. Prefer it so ImageAsset.prompt_used # matches what the model received. Legacy groups don't have this # field, so they fall through to the existing ``t2i_prompt`` value # (byte/behavior unchanged). prompt_used_value = ( res.get("effective_render_prompt") or res.get("t2i_prompt", "") or "" ) if existing: existing.file_path = rel_png_path existing.prompt_used = prompt_used_value existing.variant_index = variant_index existing.variant_label = variant_label existing.t2i_guide = t2i_guide_text existing.is_primary = 0 existing.status = "generated" existing.generation_model = "gpt-image-2.5-sunburst" _bg_row = existing else: _bg_row = ImageAsset( id=str(uuid.uuid4()), project_id=self.project_id, episode_id=self.episode_id, asset_type="chain_bg", entity_id=canon_id, variant_index=variant_index, variant_label=variant_label, variant_type=bg_id, # traceability + UPSERT 매치 키 t2i_guide=t2i_guide_text, file_path=rel_png_path, prompt_used=prompt_used_value, generation_model="gpt-image-2.5-sunburst", status="generated", is_primary=0, created_at=now, ) self.db.add(_bg_row) # persist-all Wave 1 — FP→bg lineage: location canon → floor_plan ImageAsset # 구조키(per-step 캐시맵 룩업, Wave1 정련). _fp_id_for_bg = _fp_id_by_canon.get(canon_id) # W-L: aerial 이 실제 첨부된 bg(구조 필드 aerial_ref)는 1순위 # lineage 를 loc floor_plan 대신 aerial UUID 로 교체 — 실제 첨부 # SOT 정합(야외 bg 는 fp ref 가 의도적으로 제외됨). 미해결이면 # unresolved 구조키(입력없음 [] 과 구분). _aerial_loc = str( (res.get("aerial_ref") or {}).get("loc_id") or "") _p1_unresolved: List[str] = [] if _aerial_loc: _first_uuid = _aerial_uuid_by_loc.get(_aerial_loc) if not _first_uuid: _p1_unresolved.append("location_aerial:" + _aerial_loc) else: _first_uuid = _fp_id_for_bg # W-G: 실제 첨부된 same-building indoor fp UUID(entry 구조 필드) 병합 # — 캔버스 fp→bg 자동 엣지. 링크 없으면 빈 문자열 = 기존과 동일. _bfp_uuid = str( (res.get("building_fp_ref") or {}).get("fp_asset_id") or "") _base_iids = list(dict.fromkeys( x for x in (_first_uuid, _bfp_uuid) if x)) if _base_iids: annotate_generated_asset( _bg_row, pipeline_role="background_render", input_image_ids=_base_iids, pipeline_metadata=( {"unresolved_inputs": _p1_unresolved} if _p1_unresolved else None ), ) else: # 못 찾으면 None(미상)+metadata.unresolved_inputs (입력없음[]과 구분, Codex). annotate_generated_asset( _bg_row, pipeline_role="background_render", pipeline_metadata={"unresolved_inputs": ( _p1_unresolved or ["floor_plan:" + str(canon_id)])}, ) # 3a phase1 (정정): bg_id→UUID 맵(★키=bg_id, NOT 잔여 bid) + **실제 첨부 # prior_bg** 수집(flag ON 만). 실제 첨부 SOT = attached_reference_lineage. # prior_bg_ids(구조 필드, FP 제외, prefix 없는 bg_id 리스트 — 라벨파싱 0). # parent_id(depends_on_bg[0])는 substrate 등에서 실제첨부와 diverge → 미사용. _star_uuid_by_bgid[bg_id] = _bg_row.id if _star_enabled: _lin = res.get("attached_reference_lineage") or {} # self 제외(bg 가 자기 자신 참조 금지) + dedup(stable order 보존). # FP 는 prior_bg_ids 에 애초에 없음(_build_attached_reference_lineage). _prior_ids: List[str] = [] for _b in (_lin.get("prior_bg_ids") or []): _bs = str(_b) if _bs and _bs != bg_id and _bs not in _prior_ids: _prior_ids.append(_bs) _reuse = (_lin.get("ref_used") == "reused_plate") # 구조 필드(copy-less alias) # W-I/W-K: 실제 첨부된 anchor 들의 bg_id (구조 필드 SOT) — # (label_kind, bg_id) 리스트, 첨부 순서(same_loc → building). _row_anchors: List[Tuple[str, str]] = [] _sp_bgid = str( (res.get("same_loc_anchor_ref") or {}).get("anchor_bg_id") or "") if _sp_bgid: _row_anchors.append(("same_loc_anchor", _sp_bgid)) _anchor_bgid = str( (res.get("building_anchor_ref") or {}).get("anchor_bg_id") or "") if _anchor_bgid: _row_anchors.append(("building_anchor", _anchor_bgid)) if _prior_ids or _row_anchors: # W-G: building fp UUID 도 함께 운반 — phase2 재기록이 # phase1 의 building lineage 를 유실하지 않게. W-L: # 1순위는 phase1 과 동일한 _first_uuid(aerial 교체 반영). _star_pending.append( (_bg_row, _first_uuid, _prior_ids, _reuse, _bfp_uuid, _row_anchors, _p1_unresolved)) registered += 1 # 3a phase2 (flag ON): 실제 첨부 prior_bg_ids → ImageAsset UUID resolve(배치맵 # 우선, 없으면 DB variant_type 구조 조인 fallback — 부분재렌더 대비) 후 input_ # image_ids=[fp, *prior] 재기록(캔버스 배경↔배경 엣지). 미해결은 unresolved_ # inputs 구조키. OFF 면 _star_pending 비어 no-op(byte-identical). if _star_pending: _needed = { p for _tup in _star_pending for p in _tup[2] } # W-I/W-K: anchor bg_id 도 chain_bg variant_type 으로 동일 resolve. _needed |= {a for _tup in _star_pending for _, a in _tup[5]} _resolver: Dict[str, Optional[str]] = dict(_star_uuid_by_bgid) _missing = [b for b in _needed if b not in _resolver] if _missing: for _vt, _aid in ( self.db.query(ImageAsset.variant_type, ImageAsset.id) .filter( ImageAsset.project_id == self.project_id, ImageAsset.episode_id == self.episode_id, ImageAsset.asset_type == "chain_bg", ImageAsset.variant_type.in_(_missing), ) .all() ): if _vt: _resolver[str(_vt)] = str(_aid) for (_bg_row, _fp_uuid, _prior_ids, _reuse, _row_bfp, _row_anchors, _pre_unresolved) in _star_pending: _iids, _meta = _compose_bg_input_image_ids( _prior_ids, _fp_uuid, _resolver, reuse=_reuse) # W-L: phase1 미해결 구조키(aerial 등) 이월 — 재기록이 # phase1 마킹을 유실하지 않게. for _u in _pre_unresolved: _meta.setdefault("unresolved_inputs", []).append(_u) # W-I/W-K: 실제 첨부된 anchor UUID 병합(dedup) — 첨부 순서 # (prior 다음·building fp 앞, same_loc → building)와 동일 # 위치. 미해결이면 unresolved_inputs 구조키(입력없음 [] 과 # 구분, label_kind 별). for _kind, _abgid in _row_anchors: _anchor_uuid = _resolver.get(_abgid) if _anchor_uuid and _anchor_uuid not in _iids: _iids = [*_iids, _anchor_uuid] elif not _anchor_uuid: _meta.setdefault("unresolved_inputs", []).append( _kind + ":" + _abgid) # W-G: 실제 첨부된 building fp UUID 를 마지막에 병합(dedup) — # phase1 annotate 를 덮어쓰는 재기록에서 유실 방지. if _row_bfp and _row_bfp not in _iids: _iids = [*_iids, _row_bfp] annotate_generated_asset( _bg_row, pipeline_role="background_render", input_image_ids=(_iids or None), pipeline_metadata=_meta, ) if registered: self.db.commit() logger.info( "background_render: registered %d ImageAssets " "(chain_bg + %d location_aerial)%s", registered, len(_aerial_uuid_by_loc), (f" (star anchor lineage: {len(_star_pending)})" if _star_pending else ""), ) def _compute_expected_bg_ids( self, plans_cp: Optional[Dict[str, Any]], prompts_cp: Optional[Dict[str, Any]], ) -> List[str]: """``_execute`` 와 동일한 필터로 렌더 대상 bg_id 리스트 산출 (helper). Codex I2 fix: verify_completion 의 expected 가 _execute 와 비대칭이면 prompt 실패 / duplicate / unsafe bg_id 를 expected에 포함해 false partial 이 발생한다. 이 helper로 양 sites 산출 로직을 통일한다. 필터: `_select_bg_id_filter(plans_cp)` (D6 marker → BG_ID_RE / legacy → `_LEGACY_SAFE_BG_RE`) + first-wins(중복 bg_id) + prompt status=='ok'. """ bg_filter_re = _select_bg_id_filter(plans_cp) plans_map = ((plans_cp or {}).get("data", {}) or {}).get("plans", {}) or {} prompts_map = ((prompts_cp or {}).get("data", {}) or {}).get("backgrounds", {}) or {} seen: set[str] = set() expected: List[str] = [] for entry in plans_map.values(): if not isinstance(entry, dict) or entry.get("status") != "ok": continue plan = entry.get("plan") or {} for bg in plan.get("backgrounds") or []: bid = bg.get("bg_id") if not bid or not bg_filter_re.match(bid): continue if bid in seen: continue # prompt 가 ok 인 bg만 실제 렌더 대상 (=_execute renderable 분기와 동일). if prompts_map.get(bid, {}).get("status") != "ok": continue seen.add(bid) expected.append(bid) return expected def _expected_bg_ids_from_self_cp(self) -> List[str]: """upstream cp 부재 시 자기 cp(background_render.data.groups)에서 expected 복구. Codex I1 fix: master_plan cp 가 손상/누락된 채 step_run.status='completed'면 자기 cp의 그룹 정보가 마지막 ground truth. 자기 cp도 없으면 noop이라 호출자가 빈 리스트를 보고 clean 반환하면 된다. D6: 자기 cp 의 bg_id 들도 marker-gated 검증. own_cp 가 D6 marker (bg_catalog_hash) 가지면 BG_ID_RE strict. """ own_cp = self._load_prev_checkpoint("background_render") # D6: own cp 자체가 marker 보유한 형식이면 BG_ID_RE strict. own_cp 의 data 에 # bg_catalog_hash 등 sibling 이 있으면 (background_render 가 master_plan # sibling 을 forward 한 경우) marker로 사용. 일반적으로 own cp 는 그룹 # 결과만 보유 → legacy 로 처리. consumer (master_plan) cp 와 정합 위해 # 외부 plans_cp 도 별도 검사하는 _compute_expected_bg_ids 가 1차 source. bg_filter_re = _select_bg_id_filter(own_cp) groups = ((own_cp or {}).get("data", {}) or {}).get("groups", {}) or {} expected: List[str] = [] for bid, gres in groups.items(): if not isinstance(gres, dict): continue if not bid or not bg_filter_re.match(bid): continue if gres.get("status") != "ok": continue expected.append(bid) return expected def verify_completion(self): """chain_bg 산출물 무결성 검증 (Phase 7 background_render). expected 산출 우선순위: 1. master_plan + background_prompt cp 가 모두 있으면 ``_execute`` 와 동일 필터(`_select_bg_id_filter(plans_cp)` + first-wins + prompt ok)로 expected 빌드. 2. 둘 중 하나라도 부재하면 자기 cp(`background_render.data.groups`)의 ``status=='ok'`` bg_id 로 fallback (Codex I1). 3. 자기 cp 마저 없으면 noop 으로 인정해 clean 반환. actual = ``ImageAsset(asset_type='chain_bg', variant_type IN expected)`` + 파일 stat(cwd 독립 ``resolve_image_path``). background_mode off면 step 자체가 noop → 항상 clean. cleanup_artifacts override 안 함 → default noop (사용자 caveat). """ 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 ImageAsset # background_mode off → 검증 면제 if settings.background_mode not in {"on", "floor_plan_anchored"}: return self._verify_clean_report(path="off") plans_cp = self._load_prev_checkpoint("background_master_plan") prompts_cp = self._load_prev_checkpoint("background_prompt") path_label = "phase7_master_plan" if plans_cp and prompts_cp: expected_bg_ids = self._compute_expected_bg_ids(plans_cp, prompts_cp) else: # upstream cp 부재 → 자기 cp 에서 복구 (silent fall-through 차단). logger.warning( "[VERIFY] background_render: upstream cp 부재 (master_plan=%s prompt=%s) " "→ self_cp fallback", bool(plans_cp), bool(prompts_cp), ) expected_bg_ids = self._expected_bg_ids_from_self_cp() path_label = "self_cp_fallback" expected = len(expected_bg_ids) if expected == 0: # 진짜 noop (mode=on 인데 plan 비어있거나 self_cp 도 0 ok groups). return self._verify_clean_report(path=path_label + "_empty") expected_set = set(expected_bg_ids) rows = ( self.db.query(ImageAsset) .filter( ImageAsset.project_id == self.project_id, ImageAsset.episode_id == self.episode_id, ImageAsset.asset_type == "chain_bg", ) .all() ) bgs_with_file: set[str] = set() extra_asset_bg_ids: set[str] = set() extra_asset_bg_ids_with_files: set[str] = set() for r in rows: bg_id = r.variant_type or "" p = resolve_image_path(r.file_path) has_file = bool(p and p.exists()) if bg_id in expected_set: if has_file: bgs_with_file.add(bg_id) elif bg_id: extra_asset_bg_ids.add(bg_id) if has_file: extra_asset_bg_ids_with_files.add(bg_id) found_set = bgs_with_file & expected_set found = len(found_set) missing_bgs = sorted(expected_set - found_set) image_dir = ( Path(settings.projects_dir) / self.project_id / "episodes" / self.episode_id / "images" / "background_chain" ) extra_file_bg_ids: List[str] = [] if image_dir.exists(): extra_file_bg_ids = sorted({ p.stem for p in image_dir.glob("*.png") if p.stem not in expected_set }) if found == expected: severity = "clean" elif found == 0: severity = "missing" else: severity = "partial" missing_msgs: List[str] = [] if missing_bgs: missing_msgs.append( f"{len(missing_bgs)} chain_bg images missing " f"(expected={expected}, found={found}): {missing_bgs[:20]}" ) if severity != "clean": logger.warning( "[VERIFY] background_render: path=%s expected=%d found=%d " "severity=%s missing=%s", path_label, expected, found, severity, missing_bgs[:10], ) extra_asset_bg_ids_sorted = sorted(extra_asset_bg_ids) extra_file_bg_ids_sorted = extra_file_bg_ids chain_bg_extra_overflow = ( len(extra_asset_bg_ids_sorted) > 50 or len(extra_file_bg_ids_sorted) > 50 ) # W21B-w3 Commit 2b — separate new-render vs reuse-mapping accounting. # Classification SOT is the self cp groups' render_action / is_reuse # (NOT ImageAsset status, which stays 'generated' for both). A reuse # bg is valid when its alias row resolves AND its target plate is # itself found. missing_reuse_targets surfaces broken aliases. own_cp = self._load_prev_checkpoint("background_render") own_groups = ( ((own_cp or {}).get("data", {}) or {}).get("groups", {}) or {} ) expected_new_render: List[str] = [] expected_reuse: List[str] = [] reuse_target_by_bg: Dict[str, str] = {} for bid in expected_set: g = own_groups.get(bid) or {} is_reuse = ( bool(g.get("is_reuse")) or g.get("render_action") == "reuse_existing_plate" ) if is_reuse: expected_reuse.append(bid) reuse_target_by_bg[bid] = ( g.get("reused_from_bg_id") or g.get("reuse_target_bg_id") or "" ) else: expected_new_render.append(bid) missing_reuse_targets: List[str] = [] valid_reuse = 0 for bid in expected_reuse: tgt = reuse_target_by_bg.get(bid) or "" if bid in found_set and tgt and tgt in found_set: valid_reuse += 1 else: missing_reuse_targets.append(bid) missing_reuse_targets_sorted = sorted(missing_reuse_targets) return CompletionReport( is_complete=(severity == "clean" and not missing_reuse_targets), missing=missing_msgs, severity=severity, metadata={ "chain_bg_expected": expected, "chain_bg_found": found, "path": path_label, # Commit 2b new/reuse split (diagnostic + clean gate). "expected_new_render_count": len(expected_new_render), "expected_reuse_count": len(expected_reuse), "valid_reuse_count": valid_reuse, "missing_reuse_targets": missing_reuse_targets_sorted[:50], "reuse_alias_bg_ids": sorted(expected_reuse)[:50], "reuse_target_bg_ids": sorted( { reuse_target_by_bg[b] for b in expected_reuse if reuse_target_by_bg.get(b) } )[:50], # Diagnostic-only: image steps do not delete stale artifacts on # force. Extra rows/files are ignored for completion but surfaced # so visual review and UI audits can filter to the active CP set. "chain_bg_extra_asset_rows": len(extra_asset_bg_ids), "chain_bg_extra_asset_bg_ids": extra_asset_bg_ids_sorted[:50], "chain_bg_extra_asset_rows_with_files": len( extra_asset_bg_ids_with_files ), "chain_bg_extra_file_count": len(extra_file_bg_ids), "chain_bg_extra_file_bg_ids": extra_file_bg_ids_sorted[:50], "chain_bg_extra_overflow": chain_bg_extra_overflow, }, ) def _verify_clean_report(self, *, path: str): """expected=0 / mode=off 등 정당한 noop 케이스의 clean 리포트 helper.""" from app.core.integrity_report import CompletionReport return CompletionReport( is_complete=True, missing=[], severity="clean", metadata={"chain_bg_expected": 0, "chain_bg_found": 0, "path": path}, )