"""씬 이미지 reference/entity 해결 서비스 — W5 F22 Phase B.8. SceneImageService에서 reference/visible-entity 매핑 로직을 이관. 공개 API는 facade가 유지하고 본 서비스는 구현을 담당. Codex B.6 조언(“reference resolution 성격은 validation과 분리”)을 반영한 별도 서비스. 공개 API (SceneImageService facade가 delegation): - `get_visible_entities(visible_entities_json)` — JSON → EntityCanon lookup - `get_reference_image_map(visible_entities)` — entity_id → file bytes - `build_scene_ref_image_map(ref_image_map, entity_lookup)` — location 제외 + outlook + composite + state_variant - `resolve_refs_for_prompt(t2i_prompt, ...)` — T2I ↔ ref 이미지 매칭 (short_id + 레거시 + state_variant). prop attach 는 Area D-min 이후 `required_refs` (Area B render_contracts SOT) 단일 결정자. Module-level stateless helpers (facade static wrapper): - `build_image_index(labeled_refs, entity_lookup)` — Image N 번호 부여 + sid→N 매핑 - `rewrite_t2i_with_image_refs(t2i_prompt, sid_to_img, sid_info)` — C##O##/P## → "from Image N" 치환 """ from __future__ import annotations import json import logging import re from pathlib import Path from typing import Any, Dict, List, Literal, Optional, Set, Tuple from sqlalchemy.orm import Session as OrmSession from app.core.errors import AppError from app.services.prompt_service import ( LabeledRefPayload, RefRoleError, make_labeled_ref_payload, ) from app.core.keep_elements import validate_keep_elements from app.core.subject_state import is_immobilized_state from app.models.project import CharacterOutlook, EntityCanon, EntityEpisodeLink, ImageAsset logger = logging.getLogger(__name__) def _identity_descriptor(info: Dict[str, Any]) -> str: """식별용 외형 descriptor (Inc2-a 메타-A): i2i 는 픽셀로 대상을 식별하므로 라벨에는 "이 대상이 화면상 어떻게 생겼나"만 넣는다. SOT 우선순위: entity ``description`` → ``visual_traits`` join → "" (generic). ``t2i_prompt`` 는 reference 이미지를 *생성*할 때의 지시문(passport / plain white background / pose 등)이라 식별 라벨에 부적합 — 절대 쓰지 않는다. 고유명사(이름)도 라벨에 넣지 않는다 (본문↔ref 연결은 rewrite_t2i_with_image_refs 의 "from Image N" 번호 메커니즘이 담당). ★ production lookup shape (Codex MINOR, 2026-06-28): 주 소비처 ``load_episode_entity_lookup`` 은 ``id/name/entity_type/short_id/description/ t2i_prompt`` 6필드만 반환한다 — ``visual_traits`` / ``stable_traits`` 미포함. 즉 production 에서 descriptor 는 사실상 ``description`` 기반이며, ``visual_traits`` 분기는 그 키를 주는 경로(테스트 등)를 위한 forward-compat 다. ``description`` 이 비고 ``stable_traits`` 만 외형 데이터를 가진 경우(현 lookup 엔 미노출) descriptor 가 generic("")으로 degrade 할 수 있으나, ``stable_traits`` JSON 파싱은 전샷 라벨 (Inc2-a)에 영향을 주는 별도 변경이라 본 wave 범위 밖으로 둔다 (generic 라벨이 실측 문제로 드러나면 그때 좁은 wave 로 stable_traits fallback 추가).""" desc = str((info or {}).get("description") or "").strip() if desc: return desc vt = (info or {}).get("visual_traits") or [] if isinstance(vt, list): joined = ", ".join(str(t).strip() for t in vt if str(t).strip()) if joined: return joined return "" def build_image_index( labeled_refs: list, entity_lookup: Dict[str, Dict], *, ref_roles: Optional[List[str]] = None, ref_role_metadata: Optional[List[Dict[str, Any]]] = None, attached_meta: Optional[List[Tuple[str, str]]] = None, ) -> tuple: """labeled_refs에 Image N 번호를 부여하고 sid→Image N 매핑 반환. W5 F22 Phase B.8.4: static helper로 추출 (상태 없음, stateless transformation). D5 §4.3 (2026-05-09): attached_meta keyword-only param 도입. label 만 rewrite, meta 는 평행 통과 (P1 강화 — label 변형이 meta 에 영향 X). caller 가 attached_meta=None 호출 시 빈 list 평행 반환 (backward 호환). Area #11 v1 W2 (2026-05-18+): ref_roles + ref_role_metadata parallel passthrough 추가 (4-tuple → 6-tuple). length parity fail-fast (RefRoleError). spec §3.2. Returns: (indexed_labeled_refs, sid_to_img_map, sid_info, indexed_ref_roles, indexed_ref_role_metadata, indexed_attached_meta) """ import re as _re # Area #11 v1 W2: length parity gate (No Silent Fallback). # caller 가 None 으로 호출 시 backward 호환 (D5 carry) — 모두 None 이면 빈 list 평행. _attached = list(attached_meta) if attached_meta is not None else [] _roles = list(ref_roles) if ref_roles is not None else [] _meta = list(ref_role_metadata) if ref_role_metadata is not None else [] if labeled_refs: n = len(labeled_refs) # 4 parallel list length parity (caller provides 3 sidecars; None 은 빈 list 로 fallback, # 단 production caller (W2 atomic) 는 항상 모두 제공해야 함). if attached_meta is not None and len(_attached) != n: raise RefRoleError( f"build_image_index: attached_meta length {len(_attached)} != labeled_refs length {n}" ) if ref_roles is not None and len(_roles) != n: raise RefRoleError( f"build_image_index: ref_roles length {len(_roles)} != labeled_refs length {n}" ) if ref_role_metadata is not None and len(_meta) != n: raise RefRoleError( f"build_image_index: ref_role_metadata length {len(_meta)} != labeled_refs length {n}" ) if not labeled_refs: return labeled_refs, {}, {}, _roles, _meta, _attached _sid_info: Dict[str, Dict] = {} for _einfo in entity_lookup.values(): sid = _einfo.get("short_id", "") if sid: _sid_info[sid] = _einfo indexed_refs = [] sid_to_img: Dict[str, int] = {} for i, (label, img_bytes) in enumerate(labeled_refs, 1): # Inc2-a (메타-A, Codex 합의): ref_roles/ref_role_metadata sidecar 를 primary # SOT 로 descriptor 라벨을 만든다. label 문자열 regex 재해석은 sidecar 부재 # (legacy caller) 일 때만 fallback. 어떤 경로든 이름·t2i_prompt(passport # 생성지시) 는 라벨에 넣지 않고, "Image N"/"Reference image N" prefix 도 # 붙이지 않는다 — prompt_service 가 role 별 단일 prefix SOT (이중 prefix 해소). _role = _roles[i - 1] if i - 1 < len(_roles) else "" _md = _meta[i - 1] if i - 1 < len(_meta) else {} _sid_full = str((_md or {}).get("sid") or "") _bm = _re.match(r'[CP]\d{2,3}', _sid_full) _base_sid = _bm.group(0) if _bm else "" if _role in ("character_ref", "outfit_ref_inline", "character_state_ref") and _base_sid: _desc = _identity_descriptor(_sid_info.get(_base_sid, {})) if _role == "character_state_ref": _state = str((_md or {}).get("state") or "") _head = (f"state-specific character appearance reference ({_state})" if _state else "state-specific character appearance reference") else: _head = "character appearance reference" new_label = f"{_head}: {_desc}" if _desc else _head sid_to_img[_sid_full] = i if _base_sid != _sid_full: sid_to_img[_base_sid] = i elif _role == "prop_ref" and _base_sid: _desc = _identity_descriptor(_sid_info.get(_base_sid, {})) new_label = f"object appearance reference: {_desc}" if _desc else "object appearance reference" sid_to_img[_base_sid] = i else: # legacy fallback (sidecar 부재 caller) — label regex 로 sid 만 추출, # descriptor 는 동일하게 description→visual_traits (이름·t2i_prompt 누출 0). _char_match = _re.search(r'character\s+(C\d{2,3}(?:O\d{2,3})?)', label) _prop_match = _re.search(r'object\s+(P\d{2,3})', label) if _char_match: _sid_full = _char_match.group(1) _char_sid = _re.match(r'C\d{2,3}', _sid_full).group(0) _desc = _identity_descriptor(_sid_info.get(_char_sid, {})) new_label = f"character appearance reference: {_desc}" if _desc else "character appearance reference" sid_to_img[_sid_full] = i if "O" in _sid_full and _sid_full != _char_sid: sid_to_img[_char_sid] = i elif _prop_match: _psid = _prop_match.group(1) _desc = _identity_descriptor(_sid_info.get(_psid, {})) new_label = f"object appearance reference: {_desc}" if _desc else "object appearance reference" sid_to_img[_psid] = i else: new_label = label # prefix 없이 그대로 — prompt_service 가 부여 indexed_refs.append((new_label, img_bytes)) # D5 §4.3 P1: attached_meta passthrough (label 변형 무관, 같은 인덱스 그대로). # Area #11 v1 W2: ref_roles + ref_role_metadata 도 동일 P1 passthrough. return indexed_refs, sid_to_img, _sid_info, _roles, _meta, _attached def rewrite_t2i_with_image_refs( t2i_prompt: str, sid_to_img: Dict[str, int], sid_info: Dict[str, Dict], ) -> str: """T2I의 C##O## / P##을 'the [desc] from Image N'으로 치환. W5 F22 Phase B.8.4: stateless helper. """ import re as _re if not sid_to_img or not t2i_prompt: return t2i_prompt rewritten = t2i_prompt for sid in sorted(sid_to_img.keys(), key=lambda x: -len(x)): img_n = sid_to_img[sid] _char_m = _re.match(r'C\d{2,3}', sid) if _char_m: replacement = f"the character from Image {img_n}" if "O" not in sid: pattern = r'\b' + _re.escape(sid) + r'(?:O\d{2,3})?\b' else: pattern = r'\b' + _re.escape(sid) + r'\b' elif sid.startswith("P"): _info = sid_info.get(sid, {}) _name = _info.get("name", "") replacement = f"the {_name} from Image {img_n}" if _name else f"the object from Image {img_n}" pattern = r'\b' + _re.escape(sid) + r'\b' else: continue rewritten = _re.sub(pattern, replacement, rewritten) return rewritten def apply_back_to_camera_constraints( attached_meta: List[Tuple[str, str]], ref_roles: List[str], ref_role_metadata: List[Dict[str, Any]], *, staging: Optional[Dict[str, Any]], visible_entities: List[Dict[str, Any]], entity_lookup: Dict[str, Dict], ) -> None: """B-run S15 sh5/sh7 실측 fix (2026-06-12) — back_to_camera 인물 ref 계약. shot_staging 의 structured ``character_angles[].angle == "back_to_camera"`` enum (글자 패턴 0) 인물이 character/outfit ref 로 부착돼 있으면 해당 ``ref_role_metadata`` 에 ``view_angle_constraint`` 를 stamp 한다 (in-place). prompt_service 의 character_ref / outfit_ref_inline 분기가 이를 읽어 "face must NOT be visible" 지시를 출력 — 정면 여권사진 ref + "match identity" 지시의 정면 bias 가 본문의 back-to-camera 작문을 누르는 실측 결함의 대응. name→sid resolve 는 staged-character backstop 과 동일 규약: structured name exact equality 만, ambiguous → skip+log. staging 부재/해당 enum 부재 still 은 no-op (기존 경로 byte-identical). resolve_refs_for_prompt 에 넣지 않고 별도 helper 인 이유: staging 전달이 휴면 중인 W2 backstop 전체를 활성화하지 않도록 좁게 분리 (Codex FEEDBACK3 합의). """ if not staging: return from app.core.subject_reference_policy import normalize_subject_id btc_sids: set = set() for ca in staging.get("character_angles") or []: if ca.get("angle") != "back_to_camera": continue ca_name = ca.get("character", "") if not ca_name: continue cands: set = set() for ve in visible_entities: if (ve.get("name") == ca_name and ve.get("entity_type") == "character" and ve.get("short_id")): cands.add(ve["short_id"]) for info in entity_lookup.values(): if (info.get("name") == ca_name and info.get("entity_type") == "character" and info.get("short_id")): cands.add(info["short_id"]) if len(cands) == 1: btc_sids.add(next(iter(cands))) elif len(cands) > 1: logger.warning( "back_to_camera constraint skip: name %r ambiguous (%d entities)", ca_name, len(cands), ) if not btc_sids: return for i, (kind, sid) in enumerate(attached_meta): if i >= len(ref_roles) or i >= len(ref_role_metadata): break if ref_roles[i] not in ("character_ref", "outfit_ref_inline"): continue if kind == "character": base_sid = sid elif kind == "character_outlook": base_sid = normalize_subject_id( sid, warn_on_outlook=False, where="apply_back_to_camera_constraints", ) else: continue if base_sid in btc_sids and isinstance(ref_role_metadata[i], dict): ref_role_metadata[i]["view_angle_constraint"] = "back_to_camera" class SceneReferenceService: """씬 이미지 생성 시 reference/visible-entity 해결 서비스 (W5 F22 Phase B.8).""" def __init__(self, db: OrmSession, project_id: str) -> None: self._db = db self._project_id = project_id def get_visible_entities(self, visible_entities_json: Optional[str]) -> List[Dict[str, Any]]: """SceneStill.visible_entities_json을 파싱해서 EntityCanon 상세 dict 목록 반환. JSON 형식 허용: - 리스트 요소가 dict: `{"id": ...}` 또는 `{"entity_id": ...}` - 리스트 요소가 str: entity_id 자체 """ if not visible_entities_json: return [] try: visible_ids = json.loads(visible_entities_json) except json.JSONDecodeError: return [] visible_entities = [] for v in visible_ids: if isinstance(v, dict): eid = v.get("id") or v.get("entity_id", "") elif isinstance(v, str): eid = v else: continue entity = ( self._db.query(EntityCanon) .filter(EntityCanon.id == eid) .first() ) if entity: visible_entities.append({ "id": entity.id, "name": entity.name, "entity_type": entity.entity_type, "short_id": entity.short_id or "", "description": entity.description or "", "stable_traits": entity.stable_traits or "{}", }) return visible_entities def resolve_chain_bg_asset_id( self, episode_id: Optional[str], bg_id: Optional[str], ) -> Optional[str]: """P0 (2026-07-01): chain_bg plate 의 실제 ImageAsset UUID 를 구조키로 resolve. ``bg_id`` (예: ``L04B06``) 는 ImageAsset UUID 가 아니라 ``ImageAsset.variant_type`` 구조키다(Codex BLOCKING). scene 의 실제 입력 lineage(input_image_ids)에는 반드시 실제 UUID 를 넣어야 generated_input 엣지가 dangling 하지 않는다. asset_type=='chain_bg' AND variant_type==bg_id 로 조회(is_primary 우선, 최신). 인스턴스 캐시로 샷 반복 조회 회피. """ if not episode_id or not bg_id: return None _cache = getattr(self, "_chain_bg_uuid_cache", None) if _cache is None: _cache = {} self._chain_bg_uuid_cache = _cache _key = (episode_id, bg_id) if _key in _cache: return _cache[_key] row = ( self._db.query(ImageAsset.id) .filter( ImageAsset.project_id == self._project_id, ImageAsset.episode_id == episode_id, ImageAsset.asset_type == "chain_bg", ImageAsset.variant_type == bg_id, ) .order_by(ImageAsset.is_primary.desc(), ImageAsset.created_at.desc()) .first() ) _uuid = row[0] if row else None _cache[_key] = _uuid return _uuid def build_scene_ref_image_map( self, ref_image_map: Dict[str, bytes], entity_lookup: Dict[str, Dict], *, out_asset_id_map: Optional[Dict[str, str]] = None, ) -> Dict[str, bytes]: """ref_image_map에서 location 제외 + outlook 단독 + composite + state_variant refs 포함. W5 F22 Phase B.8.2 이관. generate_images의 씬 이미지 생성 경로에서 사용. Keys in returned map: - entity_id (character/prop): ref_image_map 그대로 (location 제외) - outlook_id: 아웃룩 ref (단독) - `composite:{char_id}:{outlook_id}`: 캐릭터 + 아웃룩 합성 - `state_variant:{char_id}:{state}`: 죽은/다친 인물 state variant P0 (2026-07-01): ``out_asset_id_map`` 이 주어지면 **동일 key** → 실제 ImageAsset UUID 를 병렬로 채운다(실제 첨부 lineage SOT). base entity 는 primary reference asset 을 자체 조회(load_entity_reference_images 와 동일 criteria), outlook/composite/state 는 아래 loop 의 asset row 에서 id 회수. None 이면 기존 경로 byte-identical (out-param, return 불변). """ import re as _re_comp # P0: base entity 의 primary reference asset_id 자체 조회(out 요청 시만). _base_asset_ids: Dict[str, str] = {} if out_asset_id_map is not None: _base_eids = [ eid for eid in ref_image_map if (entity_lookup.get(eid, {}) or {}).get("entity_type") != "location" ] if _base_eids: _base_rows = ( self._db.query(ImageAsset.entity_id, ImageAsset.id) .filter( ImageAsset.project_id == self._project_id, ImageAsset.entity_id.in_(_base_eids), ImageAsset.asset_type == "reference", ImageAsset.is_primary == 1, ) .order_by(ImageAsset.created_at.desc()) .all() ) for _eid, _aid in _base_rows: _base_asset_ids.setdefault(_eid, _aid) scene_ref_image_map: Dict[str, bytes] = {} for eid, img_bytes in ref_image_map.items(): entity_info = entity_lookup.get(eid, {}) if entity_info.get("entity_type") == "location": continue scene_ref_image_map[eid] = img_bytes if out_asset_id_map is not None and eid in _base_asset_ids: out_asset_id_map[eid] = _base_asset_ids[eid] try: # 아웃룩 단독 (키: outlook_id) outlook_eids = [eid for eid, info in entity_lookup.items() if info.get("entity_type") == "outlook"] if outlook_eids: outlook_assets = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.entity_id.in_(outlook_eids), ImageAsset.asset_type == "reference", ImageAsset.is_primary == 1, ) .all() ) for oa in outlook_assets: fp = Path(oa.file_path) if fp.exists() and oa.entity_id not in scene_ref_image_map: scene_ref_image_map[oa.entity_id] = fp.read_bytes() if out_asset_id_map is not None: out_asset_id_map[oa.entity_id] = oa.id # 합성 이미지 (키: composite:{char_id}:{outlook_id}) composite_assets = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.asset_type == "reference", ImageAsset.prompt_used.like("%composite:%"), ) .all() ) for ca in composite_assets: m = _re_comp.search(r'composite:([a-f0-9-]+):([a-f0-9-]+)', ca.prompt_used or "") if m: comp_key = f"composite:{m.group(1)}:{m.group(2)}" fp = Path(ca.file_path) if fp.exists() and comp_key not in scene_ref_image_map: scene_ref_image_map[comp_key] = fp.read_bytes() if out_asset_id_map is not None: out_asset_id_map[comp_key] = ca.id # state variant refs (죽은/다친 인물) state_v_assets = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.asset_type == "reference", ImageAsset.prompt_used.like("%state_variant:%"), ) .all() ) for sva in state_v_assets: m = _re_comp.search(r'state_variant:([a-f0-9-]+):(\w+)', sva.prompt_used or "") if m: sv_key = f"state_variant:{m.group(1)}:{m.group(2)}" fp = Path(sva.file_path) if fp.exists() and sv_key not in scene_ref_image_map: scene_ref_image_map[sv_key] = fp.read_bytes() if out_asset_id_map is not None: out_asset_id_map[sv_key] = sva.id logger.info("Loaded outfit + composite + state_variant refs into scene_ref_image_map") except Exception as exc: logger.warning("Failed to load outfit/composite/state_variant refs: %s", exc) return scene_ref_image_map def resolve_refs_for_prompt_set( self, t2i_prompts: List[str], visible_entities: list, scene_ref_image_map: Dict[str, bytes], entity_lookup: Dict[str, Dict], state_variant_sids: Optional[Dict[str, Dict]] = None, *, required_refs: Optional[List[Dict[str, Any]]] = None, staging: Optional[Dict[str, Any]] = None, force_character_names: Optional[Set[str]] = None, scene_ref_asset_id_map: Optional[Dict[str, str]] = None, ) -> LabeledRefPayload: """Multiple t2i_prompts 의 ID 매칭 union 으로 ref/meta build. 2026-05-10 — 24 shot deterministic ref-contract fail fix (Fix A): 기존 coordinator 가 first variation 의 t2i_prompt 만 보고 attached_meta build 하던 결함 차단. target_variations[*] 모든 prompt 의 ID union 으로 attach — 모든 variation 이 동일 ref set 공유 (Gemini image_edit 은 사용 안 하는 ref 무시 OK). 구현: prompts 를 ``\\n`` join 후 기존 resolve_refs_for_prompt 위임. char/outlook short_id regex 매칭은 join 텍스트 전체에서 작동, _used_ref_ids set 이 dedup 보장. 빈 list / 모두 빈 string → 빈 payload 반환. invariant 1 보존. Area D-min (2026-05-14): prop attach 는 required_refs keyword arg 단일 SOT — join 텍스트의 P##/prop name 매칭은 더 이상 prop attach 신호가 아님. Area #11 v1 W2 (2026-05-18+): return shape 2-tuple → LabeledRefPayload. producer 가 ref_roles + ref_role_metadata 동반 emit. consumer (resolve_ref_roles) 는 enum dispatch only (substring branch 0). """ if not t2i_prompts: return make_labeled_ref_payload( labeled_refs=[], ref_roles=[], ref_role_metadata=[], attached_meta=[], ) joined = "\n".join(p for p in t2i_prompts if p) if not joined: return make_labeled_ref_payload( labeled_refs=[], ref_roles=[], ref_role_metadata=[], attached_meta=[], ) return self.resolve_refs_for_prompt( t2i_prompt=joined, visible_entities=visible_entities, scene_ref_image_map=scene_ref_image_map, entity_lookup=entity_lookup, state_variant_sids=state_variant_sids, required_refs=required_refs, staging=staging, force_character_names=force_character_names, scene_ref_asset_id_map=scene_ref_asset_id_map, ) def _collect_ref_roles_for_prompt_set( self, *, labeled_refs: List[Tuple[str, Any]], attached_meta: List[Tuple[str, str]], ) -> Tuple[List[str], List[Dict[str, Any]]]: """Area #11 v1 W1 placeholder helper — W2 implementation pending. Codex iter 2 Important 1 fix — silent fallback stub 폐기. Production unused (W2 atomic switch 까지). W1 호출 시 RefRoleError raise (No Silent Fallback gate 일관). W2 atomic 에서 producer append site 와 동시 implementation rewrite. """ from app.services.prompt_service import RefRoleError raise RefRoleError( "_collect_ref_roles_for_prompt_set not production-ready " "(W1 placeholder; W2 implementation required). No Silent Fallback gate." ) def resolve_refs_for_prompt( self, t2i_prompt: str, visible_entities: list, scene_ref_image_map: Dict[str, bytes], entity_lookup: Dict[str, Dict], state_variant_sids: Optional[Dict[str, Dict]] = None, *, required_refs: Optional[List[Dict[str, Any]]] = None, staging: Optional[Dict[str, Any]] = None, force_character_names: Optional[Set[str]] = None, scene_ref_asset_id_map: Optional[Dict[str, str]] = None, ) -> LabeledRefPayload: """T2I 프롬프트에서 참조 이미지 매칭 — short_id(C01O02) + 레거시([[name]+[outlook]]) + state_variant. prop attach 는 Area D-min 이후 `required_refs` (Area B render_contracts producer SOT) 단일 결정자. 옛 prop text matching 분기 (P## word-boundary, prop name word-boundary/CJK substring) 는 폐기. W5 F22 Phase B.8.3 이관. state_variant_sids: 죽은/다친 인물 short_id → {key, state} — state variant ref 제공. D5 §4.2 + §4.3 (2026-05-09): return 을 (labeled_refs, attached_meta) 2-tuple. attached_meta = list[(kind, id)] — append 시점의 source-of-truth (regex 매칭 결과 / entity_lookup short_id) 만 사용 (P1: 라벨 추론 X). composite 부재 fallback 은 ("character", char_sid) 별 kind 로 기록 — character_outlook 위조 금지 (P2). invariant 1: len(labeled_refs) == len(attached_meta). Area D-min (2026-05-14): prop attach 는 required_refs keyword arg 단일 SOT. """ import re as _re labeled_refs: List[Tuple[str, bytes]] = [] attached_meta: List[Tuple[str, str]] = [] # Area #11 v1 W2: parallel sidecar — ref_roles + ref_role_metadata # spec §3.1 / §3.2. invariant: len == len(labeled_refs) at return. ref_roles: List[str] = [] ref_role_metadata: List[Dict[str, Any]] = [] _used_ref_ids: set = set() # P0 (2026-07-01): 실제 첨부한 image asset UUID 를 attach 시점(정확한 # scene_ref_image_map key 가 손에 있는 지점)에 metadata 로 stamp. 최종 scene # asset input_image_ids lineage SOT (라벨 파싱 금지). scene_ref_asset_id_map # 이 None(기존 caller) 이면 stamp 스킵 → byte-identical. def _stamp(meta: Dict[str, Any], key: str, role: str) -> Dict[str, Any]: if scene_ref_asset_id_map is not None: _aid = scene_ref_asset_id_map.get(key) if _aid: meta["asset_id"] = _aid meta["pipeline_role"] = role return meta # short_id → UUID 매핑 _sid_to_uuid = {ve.get("short_id", ""): ve.get("id", "") for ve in visible_entities if ve.get("short_id")} _sid_to_uuid.update({info.get("short_id", ""): eid for eid, info in entity_lookup.items() if info.get("short_id")}) def _find_char_outlook_ids(char_ref, outlook_ref): """char/outlook 참조에서 UUID 찾기 (short_id 우선, 이름 fallback).""" char_id = _sid_to_uuid.get(char_ref) outlook_id = _sid_to_uuid.get(outlook_ref) if not char_id: char_id = next( (ve.get("id", "") for ve in visible_entities if ve.get("name") == char_ref and ve.get("entity_type") == "character"), None, ) if not outlook_id: outlook_id = next( (ve.get("id", "") for ve in visible_entities if ve.get("name") == outlook_ref and ve.get("entity_type") == "outlook"), None, ) if not outlook_id: outlook_id = next( (eid for eid, info in entity_lookup.items() if info.get("name") == outlook_ref and info.get("entity_type") == "outlook"), None, ) return char_id, outlook_id _sv_sids = state_variant_sids or {} # 1) short_id 패턴: C01O02 for match in _re.finditer(r'(C\d{2,3})(O\d{2,3})', t2i_prompt): sid = match.group(0) # e.g. C05O07 char_sid = match.group(1) # e.g. C05 outlook_sid = match.group(2) # e.g. O07 # state variant가 있는 인물 → variant ref 제공 if char_sid in _sv_sids: sv = _sv_sids[char_sid] if sv["key"] not in _used_ref_ids and sv["key"] in scene_ref_image_map: labeled_refs.append(( f"character {sid} — {sv['state']} state reference", scene_ref_image_map[sv["key"]] )) # D5 P2: state_variant 가 character_outlook 만족시키면 안 됨 — 별 kind. attached_meta.append(("character_state", f"{char_sid}:{sv['state']}")) # Area #11 v1 W2: parallel sidecar (state-variant → character_state_ref) ref_roles.append("character_state_ref") ref_role_metadata.append(_stamp( {"sid": char_sid, "state": sv["state"]}, sv["key"], "character_state_variant")) _used_ref_ids.add(sv["key"]) char_id = _sid_to_uuid.get(char_sid) if char_id: _used_ref_ids.add(char_id) outlook_id = _sid_to_uuid.get(outlook_sid) if outlook_id: _used_ref_ids.add(outlook_id) continue # O00 (Null Outlook) — composite 없이 캐릭터 base ref 이미지 직접 사용 if outlook_sid == "O00": char_id = _sid_to_uuid.get(char_sid) if char_id and char_id in scene_ref_image_map and char_id not in _used_ref_ids: labeled_refs.append((f"character {char_sid} identity", scene_ref_image_map[char_id])) # D5 P2: O00 base — character kind. character_outlook 위조 X. attached_meta.append(("character", char_sid)) # Area #11 v1 W2: parallel sidecar (O00 base → character_ref) ref_roles.append("character_ref") ref_role_metadata.append(_stamp( {"sid": char_sid}, char_id, "reference_face")) _used_ref_ids.add(char_id) continue char_id, outlook_id = _find_char_outlook_ids(match.group(1), match.group(2)) if not char_id: continue composite_key = f"composite:{char_id}:{outlook_id}" if outlook_id else None if composite_key and composite_key in scene_ref_image_map and composite_key not in _used_ref_ids: labeled_refs.append((f"character {sid} in outfit", scene_ref_image_map[composite_key])) # D5 P1 SOT: regex 매칭 결과 sid (= "C01O02") 사용. attached_meta.append(("character_outlook", sid)) # Area #11 v1 W2: parallel sidecar (composite → outfit_ref_inline) ref_roles.append("outfit_ref_inline") ref_role_metadata.append(_stamp( {"sid": sid, "outfit_kind": "composite"}, composite_key, "reference_composite")) _used_ref_ids.add(composite_key) _used_ref_ids.add(char_id) if outlook_id: _used_ref_ids.add(outlook_id) continue if char_id in scene_ref_image_map and char_id not in _used_ref_ids: labeled_refs.append((f"character {char_sid} identity", scene_ref_image_map[char_id])) # D5 P2 strict: composite 부재 → base attach but 별 kind. # character_outlook 위조 금지 (S8 production 결함의 직접 fix). attached_meta.append(("character", char_sid)) # Area #11 v1 W2: parallel sidecar (composite 부재 fallback → character_ref) ref_roles.append("character_ref") ref_role_metadata.append(_stamp( {"sid": char_sid}, char_id, "reference_face")) _used_ref_ids.add(char_id) # outfit 단독 제공 금지 — composite 없으면 face만 (outfit 단독은 2명으로 해석됨) if outlook_id: _used_ref_ids.add(outlook_id) # outfit-only fallback 차단 # 2) 레거시 [[name]+[outlook]] 패턴 for match in _re.finditer(r'\[\[([^\]]+)\]\+\[([^\]]+)\]\]', t2i_prompt): char_name, outlook_name = match.group(1), match.group(2) char_id, outlook_id = _find_char_outlook_ids(char_name, outlook_name) if not char_id: continue # D5 P1 SOT (I5 fix): legacy 경로 char_sid 추출은 entity_lookup 의 short_id # 단일 source-of-truth. 부재 시 ref + meta 동시 skip (silent forgery 차단, # length match invariant 보존). _char_info = entity_lookup.get(char_id) or {} _legacy_char_sid = _char_info.get("short_id", "") if not _legacy_char_sid: logger.warning( "legacy ref skip: char_id=%s entity_lookup short_id 부재 — " "P1 SOT 위반 회피 (D5)", str(char_id)[:8] if char_id else "", ) continue if outlook_name == "미지정": if char_id in scene_ref_image_map and char_id not in _used_ref_ids: labeled_refs.append(("character identity", scene_ref_image_map[char_id])) # D5 P2: legacy 미지정 → ('character', short_id) 별 kind. attached_meta.append(("character", _legacy_char_sid)) # Area #11 v1 W2: parallel sidecar (legacy 미지정 → character_ref) ref_roles.append("character_ref") ref_role_metadata.append(_stamp( {"sid": _legacy_char_sid}, char_id, "reference_face")) _used_ref_ids.add(char_id) continue # 1) 합성 이미지 우선 (얼굴+옷 합친 전신) composite_key = f"composite:{char_id}:{outlook_id}" if outlook_id else None if composite_key and composite_key in scene_ref_image_map and composite_key not in _used_ref_ids: labeled_refs.append((f"character {char_name} in outfit", scene_ref_image_map[composite_key])) # D5 P1 SOT: outlook short_id 도 entity_lookup 에서 read. _legacy_outlook_sid = (entity_lookup.get(outlook_id) or {}).get("short_id", "") if outlook_id else "" if _legacy_outlook_sid: attached_meta.append(("character_outlook", f"{_legacy_char_sid}{_legacy_outlook_sid}")) else: # outlook short_id 부재 — composite attach 했지만 outlook 식별 불가 → # P2: character_outlook 위조 금지, base meta 로 (id 식별 X). attached_meta.append(("character", _legacy_char_sid)) # Area #11 v1 W2: parallel sidecar (legacy composite → outfit_ref_inline) ref_roles.append("outfit_ref_inline") ref_role_metadata.append(_stamp( {"sid": _legacy_char_sid, "outfit_kind": "composite"}, composite_key, "reference_composite")) _used_ref_ids.add(composite_key) _used_ref_ids.add(char_id) # face-only fallback 차단 if outlook_id: _used_ref_ids.add(outlook_id) # outfit-only fallback 차단 continue # 2) fallback: 얼굴만 (outfit 단독 제공 금지) if char_id in scene_ref_image_map and char_id not in _used_ref_ids: labeled_refs.append(("character identity", scene_ref_image_map[char_id])) # D5 P2: legacy fallback → ('character', short_id) 별 kind. attached_meta.append(("character", _legacy_char_sid)) # Area #11 v1 W2: parallel sidecar (legacy fallback → character_ref) ref_roles.append("character_ref") ref_role_metadata.append(_stamp( {"sid": _legacy_char_sid}, char_id, "reference_face")) _used_ref_ids.add(char_id) if outlook_id: _used_ref_ids.add(outlook_id) # outfit-only 차단 # 2.5) character base attach — FINDING 9 W2 (Cat2): required_refs(kind= # 'character') force-attach. base_id_required subject 는 t2i_prompt 에 # bare C## (W4b canonicalization) 를 쓰므로 section 1 의 composite regex # (C##)(O##) 가 영구 미매칭 → 이 branch 가 base ref attach 의 단일 SOT. # prop branch(아래 3)와 대칭. visible 부재 / ref image 부재는 skip — # Tier validator 가 image stage 직전 fail-fast (RefContractError). _required_char_sids: List[str] = [] if required_refs: for _r in required_refs: if isinstance(_r, dict) and _r.get("kind") == "character": _cid = _r.get("id") or "" if _cid: _required_char_sids.append(_cid) for _char_sid in _required_char_sids: _char_uuid = _sid_to_uuid.get(_char_sid) if not _char_uuid: continue # visible 부재 / short_id 매핑 부재 (Tier 1 validator) if _char_uuid in _used_ref_ids: continue # 이미 attach (regex/legacy branch) — intra-loop dedup # P8 Inc2-b (2026-06-22): bare C## immobilized subject 의 state-variant # ref 우선. base_id_required 인 시신/부상자는 t2i 에 bare C## (outlook # 없음) 를 쓰므로 section 1 composite (C##)(O##) state-variant 분기를 안 # 탄다 → 여기가 SOT. base passport bytes 대신 dead/injured variant bytes # 를 attach (section 1 565-584 과 대칭). attached_meta=("character_state", # "C##:state") + ref_role=character_state_ref → build_image_index 가 # "state-specific character appearance reference (state)" 라벨 부여 # (Inc2-a). validator step 3b 가 character_state 를 required character # 충족으로 인정(Codex 경고 대응: character_outlook 거짓 우회 X, validator # 명시 인정). state_variant_sids 는 detect_state_variant_sids 가 이 shot 의 # immobilized 판정(staging.subject_state)+ref 존재 시에만 채우므로 alive # 샷에 dead bytes 가 붙는 일 없음. _sv = _sv_sids.get(_char_sid) if (_sv and _sv.get("key") and _sv["key"] not in _used_ref_ids and _sv["key"] in scene_ref_image_map): labeled_refs.append(( f"character {_char_sid} — {_sv['state']} state reference", scene_ref_image_map[_sv["key"]], )) attached_meta.append(("character_state", f"{_char_sid}:{_sv['state']}")) ref_roles.append("character_state_ref") ref_role_metadata.append(_stamp( {"sid": _char_sid, "state": _sv["state"]}, _sv["key"], "character_state_variant")) _used_ref_ids.add(_sv["key"]) _used_ref_ids.add(_char_uuid) # base face dedup — variant 가 대체 continue if _char_uuid not in scene_ref_image_map: continue # ref image 부재 (Tier 3 가 잡음) labeled_refs.append(( f"character {_char_sid} identity", scene_ref_image_map[_char_uuid], )) # D5 P2: base ref — character kind (character_outlook 위조 금지). attached_meta.append(("character", _char_sid)) # Area #11 v1 W2: parallel sidecar (character base required → character_ref) ref_roles.append("character_ref") ref_role_metadata.append(_stamp( {"sid": _char_sid}, _char_uuid, "reference_face")) _used_ref_ids.add(_char_uuid) # 2.6) staged-character backstop — feedback6-B (2026-06-11 B-run S14 sh7 # 실측, Codex APPROVE_RUNTIME_BACKSTOP_NARROW). VE(shot_director_ve_map) # 가 staged 인물을 누락하면 policy 필터→required_refs→2.5 체인 전체가 # 눈멂 → 본문 generic 묘사 인물의 의상/정체성 발명. backstop 조건(전부 # 구조 데이터, 글자 패턴 0): # - staging.character_angles membership = staged 신호 1차 SOT # - structured name exact equality 로만 sid resolve (ambiguous → skip+log) # - explicit generic_descriptor_allowed 존중 (S15 sh7 by-design) # - 이미 attach 된 인물(_used_ref_ids) skip / bytes 부재 skip+log # - base identity ref 만 (outlook/composite 강제 X — Codex 권고 5) # producer-level fix(RPC visible 확장) 승격용 diagnostic metadata 동반. if staging: from app.core.subject_reference_policy import normalize_subject_id _ve_ids = {ve.get("id", "") for ve in visible_entities if ve.get("id")} _policy_by_sid: Dict[str, str] = {} for _p in staging.get("subject_reference_policy") or []: if isinstance(_p, dict) and _p.get("subject_id"): _base_sid = normalize_subject_id( _p["subject_id"], warn_on_outlook=False, where="resolve_refs_for_prompt.staged_backstop", ) _policy_by_sid[_base_sid] = _p.get("policy", "") for ca in staging.get("character_angles") or []: ca_name = ca.get("character", "") if not ca_name: continue # structured exact name resolve — visible_entities 우선, entity_lookup 보강 _cands: Dict[str, str] = {} # id → sid for ve in visible_entities: if ve.get("name") == ca_name and ve.get("entity_type") == "character" and ve.get("id"): _cands[ve["id"]] = ve.get("short_id", "") for eid, info in entity_lookup.items(): if info.get("name") == ca_name and info.get("entity_type") == "character": _cands.setdefault(eid, info.get("short_id", "")) if len(_cands) != 1: if len(_cands) > 1: logger.warning( "staged-character backstop skip: name %r ambiguous " "(%d entities) — diagnostic only", ca_name, len(_cands), ) continue _bs_char_id, _bs_char_sid = next(iter(_cands.items())) if not _bs_char_sid: continue # W21B-W8 option C: outdoor composition continuity group 멤버에서 # forced 된 staged character 는 generic_descriptor_allowed 라도 # attach (같은 연속 그룹의 동일 피사체 — anchor frame·staging 양쪽에 # 존재. step 시점 this-shot ∩ anchor_source staged 교집합으로 좁게 # 산출돼 source 에 없는 인물 강제 0 — Codex 안전 가드). _forced = ca_name in (force_character_names or set()) # explicit generic 선언 존중. 부재 = SOT default(id_and_outlook_required) 계열 → attach 대상. if (not _forced and _policy_by_sid.get(_bs_char_sid, "") == "generic_descriptor_allowed"): continue if _bs_char_id in _used_ref_ids: continue if _bs_char_id not in scene_ref_image_map: logger.warning( "staged-character backstop skip: %s staged but no ref " "bytes in scene_ref_image_map", _bs_char_sid, ) continue labeled_refs.append(( f"character {_bs_char_sid} identity", scene_ref_image_map[_bs_char_id], )) attached_meta.append(("character", _bs_char_sid)) ref_roles.append("character_ref") ref_role_metadata.append(_stamp({ "sid": _bs_char_sid, "staged_character_backstop_attached": True, "reason": ( "staged_character_in_continuity_group" if _forced else "staged_character_missing_from_visible_entities" if _bs_char_id not in _ve_ids else "staged_character_ref_not_attached" ), "policy_source": "staging_character_angles", **({"outdoor_continuity_forced_character_ref": True} if _forced else {}), }, _bs_char_id, "reference_face")) _used_ref_ids.add(_bs_char_id) logger.warning( "staged-character backstop attach: %s (%s) — staged in " "character_angles but no ref attached by token/required_refs " "(feedback6-B)", _bs_char_sid, ca_name, ) # 3) prop attach — Area D-min: required_refs(kind='prop') 가 prop attach 의 # 단일 SOT. legacy text matching (옛 (a) P## word-boundary + (b) prop_name # word-boundary/CJK substring) 분기는 Area D-min 에서 폐기 — Area B 의 # render_contracts → required_refs_from_render_contracts → asset_requirements # → coordinator → 본 keyword arg 까지 producer SOT 일관. # # RPC producer SOT — required_refs(kind='prop') 만 attach 결정 신호. # visible_entities 미등재 prop / ref image 부재 prop 은 skip — Tier 3 # validate_attached_refs 가 image stage 직전 fail-fast (RefContractError). _required_prop_sids: List[str] = [] if required_refs: for _r in required_refs: if isinstance(_r, dict) and _r.get("kind") == "prop": _pid = _r.get("id") or "" if _pid: _required_prop_sids.append(_pid) for _prop_sid in _required_prop_sids: _target_eid = next( (ve.get("id", "") for ve in visible_entities if ve.get("short_id") == _prop_sid and ve.get("entity_type") == "prop"), None, ) if not _target_eid: continue # visible 부재 (Tier 1 validator 가 잡음) if _target_eid in _used_ref_ids: continue # intra-loop dedup — required_refs 중복 prop (T5 검증) + 미래 branch defense if _target_eid not in scene_ref_image_map: continue # ref image 부재 (Tier 3 가 잡음) labeled_refs.append(( f"object {_prop_sid} (required)", scene_ref_image_map[_target_eid], )) attached_meta.append(("prop", _prop_sid)) # Area #11 v1 W2: parallel sidecar (prop required → prop_ref) ref_roles.append("prop_ref") ref_role_metadata.append(_stamp( {"sid": _prop_sid}, _target_eid, "reference_prop")) _used_ref_ids.add(_target_eid) # Area #11 v1 W2: payload return (substring branch 0 producer SOT). return make_labeled_ref_payload( labeled_refs=labeled_refs, ref_roles=ref_roles, ref_role_metadata=ref_role_metadata, attached_meta=attached_meta, ) def build_entity_text_map(self, entities: List[Dict[str, Any]]) -> Dict[str, str]: """Build short_id → text description map (+ C##O## composite keys). W5 F22 Phase B.16 (2026-04-22): generate_images 배치 준비 단계의 entity text map 빌드 블록을 이관. short_id가 지정된 각 entity를 'description' 맵에 등록하고, CharacterOutlook 링크별로 C##O## composite 키로 ', wearing ' 문자열을 추가. CharacterOutlook 쿼리 실패 시 warning log만 찍고 기본 short_id map만 반환한다 (non-fatal). fallback 순서: description → name. ★계약 (2026-07-02): entity `t2i_prompt` 는 reference 이미지 생성 전용 프롬프트(스타일 래퍼: "Photorealistic product photo, isolated object, plain neutral background" 등 포함)라 in-scene 치환 텍스트로 절대 사용 금지 — 씬 프롬프트 문장 중간에 스플라이스되면 스타일 지시 충돌로 부양 오브젝트/패널 콜라주 등 렌더 오염을 일으킨다. Returns: {short_id: description, 'C##O##': composite, ...} """ text_map: Dict[str, str] = {} entity_by_id = {e["id"]: e for e in entities} for e in entities: sid = e.get("short_id", "") if sid: text_map[sid] = e.get("description") or e.get("name", "") # C##O## 합성 키 (인물+아웃룩 설명 합침) try: char_outlooks = self._db.query(CharacterOutlook).filter( CharacterOutlook.project_id == self._project_id, ).all() for co in char_outlooks: char = entity_by_id.get(co.character_id) outfit_ent = self._db.query(EntityCanon).filter( EntityCanon.id == co.outlook_id, ).first() if char and outfit_ent and char.get("short_id") and outfit_ent.short_id: ck = f"{char['short_id']}{outfit_ent.short_id}" char_desc = char.get("description") or char.get("name", "") outfit_desc = outfit_ent.description or outfit_ent.name text_map[ck] = f"{char_desc}, wearing {outfit_desc}" except Exception as exc: logger.warning("Failed to build C##O## text map: %s", exc) return text_map def load_entity_reference_images(self, entities: List[Dict[str, Any]]) -> Dict[str, bytes]: """generate_images 초기 단계에서 전체 entity 목록의 primary reference 로드. W5 F22 Phase B.12: scene_image_service.generate_images의 참조 이미지 로드 루프를 이관. get_reference_image_map과 달리 fallback 없음 — primary만 포함. Returns: {entity_id: image_bytes} (파일이 없는 primary는 skip). P0 (2026-07-01): base entity 의 실제 첨부 asset UUID 는 build_scene_ref_image_map (out_asset_id_map) 이 동일 criteria(is_primary==1, created_at desc)로 self-query 한다 — 여기서 별도 out-param 을 노출하지 않는다(dead code 회피, Codex minor). """ ref_image_map: Dict[str, bytes] = {} for entity in entities: primary_asset = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.entity_id == entity["id"], ImageAsset.asset_type == "reference", ImageAsset.is_primary == 1, ) .order_by(ImageAsset.created_at.desc()) .first() ) if primary_asset: fp = Path(primary_asset.file_path) if fp.exists(): ref_image_map[entity["id"]] = fp.read_bytes() logger.info( "Loaded %d existing reference images for %d entities", len(ref_image_map), len(entities), ) return ref_image_map def get_reference_image_map(self, visible_entities: List[Dict[str, Any]]) -> Dict[str, bytes]: """visible_entities의 각 entity_id에 대해 primary(없으면 latest) reference 이미지 bytes. Returns: {entity_id: image_bytes} """ ref_image_map: Dict[str, bytes] = {} for entity_data in visible_entities: eid = entity_data["id"] # Get primary reference image, or latest if none is primary ref_img = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.entity_id == eid, ImageAsset.asset_type == "reference", ImageAsset.is_primary == 1, ) .first() ) if not ref_img: ref_img = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.entity_id == eid, ImageAsset.asset_type == "reference", ) .order_by(ImageAsset.created_at.desc()) .first() ) if ref_img: fp = Path(ref_img.file_path) if fp.exists(): ref_image_map[eid] = fp.read_bytes() return ref_image_map def get_ref_image_map_excluding_locations( self, visible_entities: List[Dict[str, Any]], ) -> Dict[str, bytes]: """visible_entities의 ref image map에서 location 엔티티 제외. W5 F22 Phase B.23.3 (2026-04-23): generate_single_scene_image의 all_ref → location 필터링 블록(~10 LOC)을 이관. 처리: 1. `get_reference_image_map(visible_entities)`로 전체 map 조회 2. 각 entry에서 entity_type=='location'인 eid 제외 location 판정은 visible_entities 리스트에서 id 매칭으로만 수행 (literal 보존). """ all_ref_image_map = self.get_reference_image_map(visible_entities) return { eid: img_bytes for eid, img_bytes in all_ref_image_map.items() if not any( e["id"] == eid and e.get("entity_type") == "location" for e in visible_entities ) } def load_episode_entity_lookup(self, episode_id: str) -> Dict[str, Dict[str, Any]]: """Load full episode entity lookup (canon_id → 6-field dict) for T2I matching. W5 F22 Phase B.21.2 (2026-04-22): scene_image_service.generate_single_scene_image의 entity_lookup 구성 블록을 이관. EntityEpisodeLink로 에피소드 전체 엔티티 id를 조회하고, EntityCanon에서 메타데이터를 로드해 6 필드 dict로 매핑한다. generate_single_scene_image은 T2I 프롬프트에서 non-visible 캐릭터 short_id 매칭을 위해 visible뿐 아니라 에피소드 전체 엔티티가 필요하다. 필드는 id / name / entity_type / short_id / description / t2i_prompt (stable_traits 제외). 에피소드에 연결된 entity가 없으면 빈 dict 반환. """ episode_links = ( self._db.query(EntityEpisodeLink) .filter( EntityEpisodeLink.project_id == self._project_id, EntityEpisodeLink.episode_id == episode_id, ) .all() ) ep_entity_ids = [lnk.canon_id for lnk in episode_links] ep_entities = ( self._db.query(EntityCanon).filter(EntityCanon.id.in_(ep_entity_ids)).all() if ep_entity_ids else [] ) return { ee.id: { "id": ee.id, "name": ee.name, "entity_type": ee.entity_type, "short_id": ee.short_id or "", "description": ee.description or "", "t2i_prompt": ee.t2i_prompt or "", } for ee in ep_entities } def build_prev_shot_background_ref( self, *, best_prev_bytes: Optional[bytes], bytes_source_kind: Literal["dep_scene", "location_history", "none"], still_data: Dict[str, Any], visible_entities: List[Dict[str, Any]], current_location_ids: List[str], dep_scene_id: Optional[str], stills: List[Dict[str, Any]], location_scene_history: Dict[str, Any], dep_detail_map: Dict[str, Any], staging: Optional[Dict[str, Any]], state_variant_sids: Dict[str, Any], entity_lookup: Dict[str, Any], ) -> Optional[Tuple[str, bytes, str, str, Dict[str, Any]]]: """Build (label, bytes, loc_id, ref_role, ref_role_metadata) for prev-shot, or None. Area #11 v1 W2 (2026-05-18+): return 3-tuple → 5-tuple. ref_role 은 기존 structured `ref_usage` (`zoom_in_detail` / `atmosphere_reference` / `exact_background` / else) → enum 1:1 매핑 (spec §3.2). ref_role_metadata 안 pre-parsed keep_elements / ignore / remove_hints — consumer (resolve_ref_roles) 가 prose split 안 함 (substring branch 0). W5 F22 Phase B.22.2 (2026-04-22): scene_image_service._generate_one_scene의 prev-shot background ref 구성 블록(~90 LOC)을 이관. 처리: 1. best_prev_bytes 없으면 None 반환 (Layer 1 early return) 2. _cur_vis_ids: 현재 샷 visible entity UUID 집합 3. _prev_still_data 결정: **bytes_source_kind 기반 metadata selection branching (no cross-source fallback)** — - "dep_scene" → dep_scene_id matching still 만 lookup - "location_history" → current_location_ids × location_scene_history 매칭 만 lookup - mismatch / no match → Layer 2.5 / 2.6 fail-fast (RefContractError) 4. _remove_hints: prev에 있던 character/prop 중 현재에서 빠진 것 (state_variant 제외) 5. ref_usage 분기 (zoom_in_detail / atmosphere_reference / exact_background / fallback) 6. fallback은 staging에 죽은 캐릭터 있는지로 라벨 분기 D5 §4.2.2 (2026-05-09): return 을 (label, bytes, loc_id) 3-tuple 로 확장. 2026-05-15 zoom_in_detail source provenance hardening (spec docs/superpowers/specs/2026-05-15-zoom-in-detail-source-provenance-design.md §4.5): loc_id 결정도 bytes_source_kind 기반 (no cross-source fallback): - bytes_source_kind == "dep_scene" → loc_id_from_dep (dep_scene 의 prev_still 안 entity_type='location' 첫 항의 short_id) - bytes_source_kind == "location_history" → loc_id_from_history (current_location_ids x location_scene_history 매칭 key) 결정 실패 시 빈 문자열 "". P2 strict 는 validator (T3) 에서 enforce — empty loc_id 가 어떤 required bg_id 도 만족 X. """ if not best_prev_bytes: return None # Layer 2 — bytes present 시 source kind valid 의무 (No Silent Fallback Gate). # "none" 은 no-bytes early return 전용 — bytes 있을 때 "none" 은 invalid. # spec docs/superpowers/specs/2026-05-15-zoom-in-detail-source-provenance-design.md §4. if bytes_source_kind not in {"dep_scene", "location_history"}: from app.core.ref_contract_validator import RefContractError raise RefContractError( f"prev_ref_source_missing: bytes present but bytes_source_kind invalid " f"(S{still_data.get('scene_index')}_Shot{still_data.get('shot_index')}, " f"got={bytes_source_kind!r})" ) # Layer 2.5 — dep_scene source consistency guard. # coordinator 가 dep_scene_id 없이 "dep_scene" 명시 시 차단 (Layer 2.6 와 대칭). if bytes_source_kind == "dep_scene" and not dep_scene_id: from app.core.ref_contract_validator import RefContractError raise RefContractError( f"prev_ref_source_missing: dep_scene source requires dep_scene_id " f"(S{still_data.get('scene_index')}_Shot{still_data.get('shot_index')}, " f"dep_scene_id={dep_scene_id!r})" ) cur_vis_ids = set() for ve in visible_entities: ve_id = ve.get("id", "") if ve_id: cur_vis_ids.add(ve_id) # D5 §4.2.2 — loc_id 결정 변수. dep_scene 의 prev_still 또는 history 매칭. loc_id_from_dep: str = "" loc_id_from_history: str = "" # §4.5 — Metadata selection branching (bytes_source_kind 기반). # spec §4.5: bytes provenance 와 metadata provenance 일치 의무. # bytes_source_kind="dep_scene" → dep_scene_id stills 매칭만 (location_history fallback 금지). # bytes_source_kind="location_history" → current_location_ids x location_scene_history 매칭만 (dep_scene_id stills 우선 금지). prev_still_data = None if bytes_source_kind == "dep_scene": prev_still_data = next( (s for s in stills if s.get("id") == dep_scene_id), None ) elif bytes_source_kind == "location_history": for _hist_loc_id in current_location_ids: if _hist_loc_id in location_scene_history: _, prev_still_data = location_scene_history[_hist_loc_id] loc_id_from_history = _hist_loc_id break # Layer 2.6 — location_history source consistency guard (Layer 2.5 와 대칭). # §4.5 selection 후 검증 — coordinator 가 location match 없이 "location_history" # 명시 또는 future caller 가짜 source_kind 통과 시도 fail-fast. if bytes_source_kind == "location_history" and not loc_id_from_history: from app.core.ref_contract_validator import RefContractError raise RefContractError( f"prev_ref_source_missing: location_history source requires " f"current_location_ids match in location_scene_history " f"(S{still_data.get('scene_index')}_Shot{still_data.get('shot_index')}, " f"current_location_ids={current_location_ids!r})" ) remove_hints: List[str] = [] # W3 (2026-06-11): zoom subject-coherence 검사용으로 함수 스코프 호이스트. prev_vis_ids: set = set() if prev_still_data: try: prev_vis_raw = json.loads(prev_still_data.get("visible_entities_json", "[]")) except (json.JSONDecodeError, TypeError): prev_vis_raw = [] for pv in prev_vis_raw: pv_id = "" if isinstance(pv, dict): pv_id = pv.get("id") or pv.get("entity_id", "") elif isinstance(pv, str): pv_id = pv if pv_id: prev_vis_ids.add(pv_id) # 이전 샷에는 있었지만 현재 샷에 없는 요소 → 제거 대상 # state_variant 인물(죽은/의식불명)은 별도 ref로 제공되므로 제거 대상에서 제외 gone_ids = prev_vis_ids - cur_vis_ids for gid in gone_ids: ent = entity_lookup.get(gid) if ent and ent.get("entity_type") in ("character", "prop"): ent_name = ent.get("name", gid) ent_sid = ent.get("short_id", "") if ent_sid and ent_sid in state_variant_sids: continue remove_hints.append(ent_name) # D5 §4.2.2 — dep_scene 가 사용된 경우 loc_id 결정 (prev_still 의 # visible entities 중 entity_type='location' 첫 항의 short_id). # P1: entity_lookup 의 short_id 단일 SOT (label/regex 추론 X). # 2026-05-15 §4.5 source provenance hardening — bytes_source_kind="dep_scene" # 에서만 dep loc lookup 실행 (split-provenance 차단). location_history bytes # 면 loc_id_from_history 만 사용. if bytes_source_kind == "dep_scene" and dep_scene_id and not loc_id_from_dep: for pv in prev_vis_raw: pv_id = ( pv.get("id") or pv.get("entity_id", "") if isinstance(pv, dict) else (pv if isinstance(pv, str) else "") ) if not pv_id: continue ent = entity_lookup.get(pv_id) or {} if ent.get("entity_type") == "location": loc_sid = ent.get("short_id", "") if loc_sid: loc_id_from_dep = loc_sid break # ref_usage 기반 라벨 구성 dep_key = f"{still_data.get('scene_index', 0)}_{still_data.get('shot_index', 0)}" dep_info = dep_detail_map.get(dep_key, {}) ref_usage = dep_info.get("ref_usage", "") ignore = dep_info.get("ignore_elements", "") # Layer 3 — 핵심 invariant (close 무관, ref_usage 기준). # spec docs/superpowers/specs/2026-05-15-zoom-in-detail-source-provenance-design.md §4 Layer 3: # declared ref_usage 와 실제 bytes provenance cross-check. if ref_usage == "zoom_in_detail" and bytes_source_kind != "dep_scene": from app.core.ref_contract_validator import RefContractError raise RefContractError( f"zoom_in_detail_source_violation: ref_usage='zoom_in_detail' requires " f"bytes from dep_scene, got source={bytes_source_kind!r} " f"(S{still_data.get('scene_index')}_Shot{still_data.get('shot_index')}, " f"dep_scene_id={dep_scene_id!r})" ) # close × ref_usage matrix v1 (framing_scale enum SOT v1 / spec §4.9) # — Gate 4 (No Silent Fallback). framing_scale=close + ref_usage != # zoom_in_detail → RefContractError. best_prev_bytes 존재 + attach # 후보 있을 때만 enforce (early return :725 후). from app.core.framing_scale import ( FRAMING_CLOSE, get_framing_scale_or_raise, ) from app.core.ref_contract_validator import RefContractError _framing_scale = get_framing_scale_or_raise( staging, where=( f"scene_reference_service.matrix " f"S{still_data.get('scene_index')}_Shot{still_data.get('shot_index')}" ), ) # FINDING 9 W3 (Cat1 sub-cause B) — declared no-ref vs malformed 구분. # dep_detail_map 에 shot key 부재 = shot_dependency_t2i 가 location_refs=[] # 를 emit (producer 의 정상 verdict). 이는 "location_ref object 존재 + # ref_usage 빈 문자열"(malformed) 과 다른 상태다. close shot 이 declared # location_ref 를 안 가지면 best_prev_bytes(location_history) 로 prev-shot # background 를 합성하지 않고 return None — declared no-ref 를 fallback 으로 # 되살리지 않는다. 실제 declared ref 의 close×ref_usage matrix invariant 는 # 아래에서 그대로 enforce (malformed entry 는 dep_key 존재 → matrix raise). if _framing_scale == FRAMING_CLOSE and dep_key not in dep_detail_map: return None if _framing_scale == FRAMING_CLOSE and ref_usage != "zoom_in_detail": raise RefContractError( f"close_ref_usage_violation: close framing requires " f"ref_usage='zoom_in_detail' but got {ref_usage!r} " f"(S{still_data.get('scene_index')}_" f"Shot{still_data.get('shot_index')}). Allowed: close + " f"zoom_in_detail only." ) # W3 (2026-06-11 fresh full E2E S29 실측, Codex 합의): zoom subject # coherence — 현재 샷에 dep 프레임에 없던 캐릭터가 등장하면 '같은 프레임 # 재구도'가 성립 불가 (새 인물을 그려야 하므로 exact reuse 는 프레임 복제 # 오류를 낳는다). environment continuity 로 강등 + diagnostic. # close × ref_usage matrix(위)는 declared usage 기준이라 순서상 이후 적용. zoom_downgraded_new_subjects: List[str] = [] if ref_usage == "zoom_in_detail" and prev_still_data is not None: for ve in visible_entities: ve_id = ve.get("id", "") ent = entity_lookup.get(ve_id) or {} if ent.get("entity_type") == "character" and ve_id not in prev_vis_ids: zoom_downgraded_new_subjects.append(ent.get("short_id") or ve_id) if zoom_downgraded_new_subjects: logger.warning( "Scene %d Shot %d: zoom_in_detail downgraded to continuity — " "subjects not in dep frame: %s (W3 exact-reuse 금지 diagnostic)", still_data.get("scene_index", 0), still_data.get("shot_index", 0), zoom_downgraded_new_subjects, ) ref_usage = "atmosphere_reference" # feedback6-C (2026-06-11 B-run S12 실측): 줌인 샷 VE 의 prop 이 dep # 프레임 VE 에 없으면 '같은 순간 재구도' 모순이 작문(VE/scene_detail) # 단계에서 baked in — prop 줌인은 continuity 유지가 맞으므로 W3 캐릭터 # 강등과 달리 강등하지 않고 diagnostic 으로만 보존한다. zoom_props_missing_in_dep: List[str] = [] if ref_usage == "zoom_in_detail" and prev_still_data is not None: for ve in visible_entities: ve_id = ve.get("id", "") ent = entity_lookup.get(ve_id) or {} if ent.get("entity_type") == "prop" and ve_id not in prev_vis_ids: zoom_props_missing_in_dep.append(ent.get("short_id") or ve_id) if zoom_props_missing_in_dep: logger.warning( "Scene %d Shot %d: zoom_in_detail props not in dep frame: %s " "(feedback6-C diagnostic — 줌아웃/줌인 작문 모순 의심, upstream " "VE/scene_detail 점검 대상)", still_data.get("scene_index", 0), still_data.get("shot_index", 0), zoom_props_missing_in_dep, ) if ref_usage == "zoom_in_detail": # W3 문구 완화 (Codex 합의): 'reuse this exact frame'/'Do NOT change # the pose' 절대 지시가 본문을 눌러 프레임 복제를 유발 (S29 sh11). # 환경/구도 연속성 기준 + 피사체·포즈는 본문 프롬프트 SOT 로 명시. label = "previous shot at same location (SAME MOMENT, zoomed-in reframing) — continuity reference. Keep the environment, lighting, and spatial layout consistent with this frame; the camera moved closer to a detail region of the same moment. Render the subjects and their poses as described in the prompt text (for a true zoom they match this frame). Do NOT duplicate body parts." if ignore: label += f" {ignore}" keep = dep_info.get("keep_elements", []) if keep: # Area D-next — dict shape enum check (L3 defensive, L2 loader # 가 이미 차단). label 합성: entry.label 만 사용. ref_usage 별 # kind allowlist 차이 없음 (consumer 가 의미 판단 안 함). validate_keep_elements( keep, label_source=f"S{still_data.get('scene_index')}_Shot" f"{still_data.get('shot_index')} zoom_in_detail", error_code_entry="step.scene_reference.keep_elements_kind_invalid", ) label += f" Keep: {', '.join(e['label'] for e in keep)}" if remove_hints: label += f" Also ignore: {', '.join(remove_hints)}" elif ref_usage == "atmosphere_reference": label = "previous shot at same location (DIFFERENT ROOM/ANGLE) — use ONLY as style, lighting, and atmosphere reference. Do NOT copy the exact background layout, furniture positions, or wall details from this image." elif ref_usage == "exact_background": label = "previous shot at same location (SAME ROOM) — use this background as-is." if ignore: label += f" {ignore}" keep = dep_info.get("keep_elements", []) if keep: # Area D-next — dict shape enum check (L3 defensive). label 합성: # entry.label 만 사용. ref_usage 별 kind allowlist 차이 0. validate_keep_elements( keep, label_source=f"S{still_data.get('scene_index')}_Shot" f"{still_data.get('shot_index')} exact_background", error_code_entry="step.scene_reference.keep_elements_kind_invalid", ) label += f" Keep: {', '.join(e['label'] for e in keep)}" if remove_hints: label += f" Also ignore: {', '.join(remove_hints)}" else: # fallback: staging 의 immobilized subject_state 인물 존재 여부로 라벨 분기 # Area #2 W5 — subject_state SOT (legacy mixed gaze field literal 폐기). Gate 1 helper-only. has_immobilized_chars = False if staging: for ca in staging.get("character_angles", []): state = ca["subject_state"] # required by v13 schema, KeyError = schema violation (Gate 4 fail-fast) if is_immobilized_state(state): has_immobilized_chars = True break if has_immobilized_chars: label = "previous shot at same location — use ONLY the background/environment from this image. Ignore all standing/moving people and human silhouettes. Keep registered immobilized-state figures exactly as they are. Keep architecture, furniture, lighting, and environment." else: label = "previous shot at same location — use ONLY the background/environment from this image. Ignore all people, figures, and human silhouettes. Keep only architecture, furniture, lighting, and environment." if remove_hints: label += f" Also ignore: {', '.join(remove_hints)}" # D5 §4.2.2 — loc_id 우선순위: dep_scene > history. 둘 다 없으면 빈 문자열. # P2 strict 는 validator (T3) 에서 enforce — empty loc_id 가 어떤 required # bg_id 도 만족 X (silent forgery 차단). loc_id_resolved = loc_id_from_dep or loc_id_from_history or "" # Area #11 v1 W2: ref_usage → role enum 1:1 매핑 (spec §3.2). # consumer (resolve_ref_roles) 가 enum dispatch only — prose 안 split 안 함. if ref_usage == "zoom_in_detail": ref_role = "previous_shot_same_frame_zoomed" elif ref_usage == "exact_background": ref_role = "previous_shot_same_room" elif ref_usage == "atmosphere_reference": ref_role = "previous_shot_continuity" else: ref_role = "previous_shot_continuity" # default conservative ref_role_metadata: Dict[str, Any] = { "keep_elements": dep_info.get("keep_elements", []), "ignore": ignore or "", "remove_hints": list(remove_hints) if remove_hints else [], "ref_usage": ref_usage, } # W3 diagnostic — 강등 사실을 metadata 에 보존 (validator/검증 surface). if zoom_downgraded_new_subjects: ref_role_metadata["zoom_downgraded_new_subjects"] = ( zoom_downgraded_new_subjects ) # feedback6-C (2026-06-11 B-run S12 sh7↔sh12 실측): zoom 프레임의 # immobilized(subject_state enum SOT — dead/unconscious/severely_injured) # 피사체는 본문 재서술로 자세가 흔들리면 '같은 순간'이 깨진다(시신이 # 움직임). consumer 가 image1 을 해당 피사체의 pose SOT 로 렌더하도록 # metadata 로 전달 — mobile 피사체는 W3 완화(본문 pose SOT) 유지. # 강등(ref_usage 변경) 후 평가라 downgraded zoom 에는 미발화. if ref_usage == "zoom_in_detail" and staging: _immobilized_subjects: List[Dict[str, str]] = [] for ca in staging.get("character_angles", []): _ca_state = ca["subject_state"] # required by v13 schema, KeyError = schema violation (Gate 4 fail-fast) if is_immobilized_state(_ca_state): _immobilized_subjects.append({ "character": ca.get("character", ""), "state": _ca_state, }) if _immobilized_subjects: ref_role_metadata["immobilized_subjects"] = _immobilized_subjects if zoom_props_missing_in_dep: ref_role_metadata["zoom_props_missing_in_dep"] = ( zoom_props_missing_in_dep ) logger.info( "Scene %d Shot %d: prev-shot background ref injected (remove: %s, loc_id: %s, role: %s)", still_data.get("scene_index", 0), still_data.get("shot_index", 0), remove_hints if remove_hints else "none", loc_id_resolved if loc_id_resolved else "", ref_role, ) return (label, best_prev_bytes, loc_id_resolved, ref_role, ref_role_metadata) def load_identity_family_by_sid(self, episode_id: str) -> Dict[str, Set[str]]: """entity_relation 체크포인트 → character identity-variant family 맵 (인스턴스 캐시). detect_state_variant_sids 의 identity-variant aware 매칭 입력 (2026-07-02, S12 시신: staging=base 표기 vs VE=variant 표기 어긋남 수정). cp 부재/파손 시 빈 dict — 소비자는 기존 exact 매칭으로 환원. """ cache: Dict[str, Dict[str, Set[str]]] = getattr( self, "_identity_family_cache", {}) if episode_id in cache: return cache[episode_id] fam: Dict[str, Set[str]] = {} try: from app.core.config import settings from app.modules.pipeline.visual_continuity_anchor_plan import ( build_identity_family_by_sid, ) from app.services.scene_checkpoint_loaders import _ep_checkpoint_path p = _ep_checkpoint_path( settings.projects_dir, self._project_id, episode_id, "entity_relation") if p.exists(): relations = ((json.loads(p.read_text(encoding="utf-8")).get("data") or {}) .get("relations") or []) fam = build_identity_family_by_sid(relations) except Exception as exc: # noqa: BLE001 — 비차단(기존 동작 환원) logger.warning("identity family 로드 실패 (episode=%s): %s", episode_id, exc) cache[episode_id] = fam self._identity_family_cache = cache return fam def detect_state_variant_sids( self, visible_entities: List[Dict[str, Any]], entity_lookup: Dict[str, Dict[str, Any]], scene_ref_image_map: Dict[str, bytes], staging: Optional[Dict[str, Any]], identity_family_by_sid: Optional[Dict[str, Set[str]]] = None, ) -> Dict[str, Dict[str, str]]: """Detect state_variant short_ids from staging subject_state. W5 F22 Phase B.22.1 (2026-04-22): scene_image_service._generate_one_scene의 state_variant 감지 블록(~26 LOC)을 이관. staging.character_angles를 훑어 subject_state 가 immobilized (unconscious/dead/severely_injured) 인 캐릭터를 찾고, scene_ref_image_map에 해당하는 state_variant ref가 있으면 short_id를 key로 {"key": "state_variant:UUID:STATE", "state": STATE} 매핑을 반환. Area #2 W5 (2026-05-17): legacy gaze field literal 판정 폐기 — ``is_immobilized_state(ca["subject_state"])`` 단일 SOT helper 사용 (Gate 1). staging이 None이면 빈 dict 반환. skip 로직 없음 — C##이 t2i에 있으면 무조건 ref 제공 (판단은 scene_detail 책임). """ state_variant_sids: Dict[str, Dict[str, str]] = {} if not staging: return state_variant_sids fam_map = identity_family_by_sid or {} sid_to_uuid = { ve.get("short_id", ""): ve.get("id", "") for ve in visible_entities if ve.get("short_id") } sid_to_uuid.update({ info.get("short_id", ""): eid for eid, info in entity_lookup.items() if info.get("short_id") }) # 이름→sid (episode 전체 lookup, exact equality) — staging 이 base 표기, # VE 가 variant 표기(별도 EntityCanon)일 때 identity family 로 잇는 브리지. name_to_sids: Dict[str, Set[str]] = {} for _eid, info in entity_lookup.items(): nm, s = info.get("name"), info.get("short_id") if nm and s: name_to_sids.setdefault(nm, set()).add(s) for ca in staging.get("character_angles", []): state = ca["subject_state"] # required by v13 schema, KeyError = schema violation (Gate 4 fail-fast) if not is_immobilized_state(state): continue ca_name = ca.get("character", "") char_sid = "" for ve in visible_entities: if ve.get("name") == ca_name and ve.get("short_id"): char_sid = ve["short_id"] break if not char_sid and fam_map: # ★identity-variant aware (2026-07-02): staging 인물이 VE 에 variant # EntityCanon 으로 존재하면 같은 인물로 매칭 (모호 이름 2+ 는 미적용). cands = name_to_sids.get(ca_name) or set() if len(cands) == 1: fam = fam_map.get(next(iter(cands))) or set() for ve in visible_entities: if ve.get("short_id") and ve["short_id"] in fam: char_sid = ve["short_id"] break if not char_sid: continue # state variant 자산 소유자는 family 내 다른 멤버(base EntityCanon)일 수 # 있다 — VE sid 우선, 이후 family 순회로 sv_key 를 찾는다. for cand_sid in [char_sid] + sorted( (fam_map.get(char_sid) or set()) - {char_sid}): cand_uuid = sid_to_uuid.get(cand_sid, "") sv_key = f"state_variant:{cand_uuid}:{state}" if cand_uuid and sv_key in scene_ref_image_map: state_variant_sids[char_sid] = {"key": sv_key, "state": state} break return state_variant_sids def build_custom_labeled_refs( self, visible_entities: List[Dict[str, Any]], ref_image_map: Dict[str, bytes], *, chain_bg_entry: Optional[Dict[str, Any]] = None, ) -> List[Tuple[str, bytes]]: """Build labeled reference list for custom-prompt scene generation. W5 F22 Phase B.21.3 (2026-04-22): scene_image_service.generate_single_scene_image의 custom_prompt 경로 labeled_refs 구성 블록을 이관. 각 visible entity에 대해: - location은 제외 (custom_prompt 경로는 캐릭터/소품 참조만 사용) - ref_image_map에 bytes가 있는 항목만 수집 - 라벨: character → "character identity", 그 외 → "object appearance" Phase 3 정련 (2026-06-11 S10 sh5 v2 실측): chain_bg_entry(해당 shot 의 bg map entry — space plate/chain bg)가 주어지고 image_bytes 가 있으면 ★첫 ref 로 부착★ — custom_prompt 라도 환경 정체성 ref 가 없으면 모델이 벽 재질·구조를 텍스트만으로 발명한다(자동 경로 5a 와 정합). no-plate sentinel(image_bytes 없음)은 부착하지 않는다. 예외는 suppress하고 warning log + 빈 리스트 반환 (non-fatal — ref 없이 진행). """ labeled_refs: List[Tuple[str, bytes]] = [] try: entity_lookup = {e["id"]: e for e in visible_entities} # noqa: F841 (원본 동작 보존) for e in visible_entities: eid = e["id"] etype = e.get("entity_type", "") if etype != "location" and eid in ref_image_map: label = "character identity" if etype == "character" else "object appearance" labeled_refs.append((label, ref_image_map[eid])) if chain_bg_entry and chain_bg_entry.get("image_bytes"): labeled_refs.insert( 0, ( chain_bg_entry.get("label") or "background environment ref", chain_bg_entry["image_bytes"], ), ) except Exception as exc: logger.warning("labeled_refs 구성 실패: %s — ref 없이 진행", exc) return labeled_refs