"""씬 이미지 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 import and_, not_, or_ 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" #: 파생 참조의 표식 — `prompt_used` 머리말 (실측 138장: composite 계열 43 · #: outfit 30 · state_variant 계열 25 · 표식 없음 40). `auto_set_primary` 가 #: 합성/비합성을 가르는 데 쓰는 표식과 같은 계열이고, 물러난 판(`_old`· #: `_old2`·`_oldsrc`·`_retry`)도 같은 어간으로 시작해 함께 걸린다. DERIVED_REF_PREFIXES = ("[composite", "[outfit", "[state_variant", "[outlook_id") def is_base_identity_ref(prompt_used: Optional[str]) -> bool: """이 참조 자산이 **그 인물 자체**의 신원 그림인가 (파생본이 아닌가). ★파생본도 `asset_type='reference'` 에 `is_primary=1` 이다 (합성은 합성끼리 대표를 다툰다 — `auto_set_primary`). 그래서 **기본 신원**을 고르는 조회가 이걸 안 거르면, 옷 합성이 여러 벌 있는 인물의 「기본」이 그중 한 벌로 뽑힌다. 실측(컨트리로드 2판 2026-09-20): 찰리는 대표 참조가 4개(기본 1 + 합성 3)라 `order_by(created_at.desc()).first()` 가 **코트 합성본**을 기본으로 집었다. 그 결과 의상 배정이 `O00`(옷 없음)인 씬에서 찰리가 코트를 입고 나왔고, 우비 씬에도 코트가 붙었다 — 사용자 육안 지적 여러 건의 공통 뿌리. """ # ★SQL 판(`base_identity_ref_filter`)과 **정확히 같은 판단**이어야 한다 # (2026-09-20 Codex NON-BLOCK 1). SQL 은 앞 공백을 다듬지 않으므로 여기서도 # 다듬지 않는다 — 생산자는 표식을 머리 공백 없이 쓴다(실측 138장 중 0장). return not (prompt_used or "").startswith(DERIVED_REF_PREFIXES) def base_identity_ref_filter(): """위 판단의 SQL 판 — 기본 신원 참조만 남기는 조건. 표식이 없으면(`NULL`·빈 문자열) 기본이다 — 옛 자산은 머리말이 없다. """ return and_(*[ or_(ImageAsset.prompt_used.is_(None), # ★`autoescape` 없이는 표식의 `_` 가 LIKE 와일드카드가 된다 # (2026-09-20 Codex NON-BLOCK 1) — `[state_variant` 가 # `[stateXvariant` 까지 잡아 **파이썬 판단과 갈린다**. not_(ImageAsset.prompt_used.startswith(pre, autoescape=True))) for pre in DERIVED_REF_PREFIXES ]) 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, # ★파생본(합성·의상·상태 변형)은 기본이 아니다 base_identity_ref_filter(), ) .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() ) # ★★**지금 기본 참조에 맞는 합성**을 고른다 (Codex BLOCK, 2026-09-19). # 종전에는 (인물, 아웃룩)만 보고 **먼저 걸린 것**을 썼다. 기본 몸체를 # 다시 만들어도 옛 몸으로 그린 합성이 그대로 붙을 수 있었다. # 합성 키에 기본 참조 id 가 붙어 있으면(신판) 그것을 현재 기본 참조와 # 대조하고, 없으면(옛판) 마지막 후보로만 둔다. _cur_base: Dict[str, str] = {} _char_eids = [eid for eid, info in entity_lookup.items() if info.get("entity_type") == "character"] if _char_eids: for ba in (self._db.query(ImageAsset).filter( ImageAsset.project_id == self._project_id, ImageAsset.entity_id.in_(_char_eids), ImageAsset.asset_type == "reference", ImageAsset.is_primary == 1, # ★같은 규칙을 두 곳에 적지 않는다 (Codex 지적 # 2026-09-20). 여기 있던 `~like` 두 줄은 # `[outfit`·`[state_variant` 를 안 뺐고, prompt_used # 가 NULL 이면 `NOT (NULL LIKE …)` 가 NULL 이라 # **기본 참조 행을 통째로 떨궜다**. base_identity_ref_filter(), ).all()): _cur_base.setdefault(ba.entity_id, ba.id) _cands: Dict[str, list] = {} for ca in composite_assets: m = _re_comp.search( r'composite:([a-f0-9-]+):([a-f0-9-]+)(?::([a-f0-9-]+))?', ca.prompt_used or "") if not m: continue comp_key = f"composite:{m.group(1)}:{m.group(2)}" base_id = m.group(3) or "" _cands.setdefault(comp_key, []).append((ca, base_id, m.group(1))) for comp_key, rows in _cands.items(): if comp_key in scene_ref_image_map: continue def _rank(row): ca, base_id, char_id = row cur = _cur_base.get(char_id) or "" # 0) 지금 기본 참조로 그린 것 · 1) 기본 참조 표시가 없는 옛판 # 2) 다른(낡은) 기본 참조로 그린 것 tier = 0 if (base_id and cur and base_id == cur) else ( 1 if not base_id else 2) # ★`created_at` 칸은 **text** 다 (2026-09-20 실측). 종전의 # `ca.created_at.timestamp()` 가 매번 AttributeError 를 냈고, # 아래 넓은 except 가 경고 한 줄만 남긴 채 **합성·상태 변형 # 참조 전부**를 버렸다 — 09-19 본 주행 239샷이 옷 합성 없이 # 기본 참조만 받았다. 시각 칸 공용 해석기로 읽는다. from app.core.step_lock import parse_iso _ts = parse_iso(getattr(ca, "created_at", None)) return (tier, -(_ts.timestamp() if _ts else 0)) for ca, _b, _c in sorted(rows, key=_rank): fp = Path(ca.file_path) if fp.exists(): 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 break # 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]], *, episode_id: str) -> 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", "") # ★화는 **필수**다 (Codex BLOCK 2026-09-04). 기본값 `None` 을 두면 # 안 넘겨도 조용히 돌고, `episode_outlook_rows` 의 exact 조회가 # `episode_id IS NULL` 이 되어 legacy 행을 돌리거나 빈 배열이 된다. # 예외조차 아니라 아래 `except` 도 안 타고 `C##O##` 가 소리 없이 # 틀리거나 빈다. 그래서 **try 밖에서** 선다. if not episode_id: raise AppError( code="scene_reference.episode_required", message="build_entity_text_map 은 화를 받아야 한다 — " "안 주면 C##O## 합성 키가 조용히 비거나 틀린다", status_code=400) # C##O## 합성 키 (인물+아웃룩 설명 합침) try: # ★이 화의 배정만 (Codex BLOCK 2026-09-04). from app.core.entity_identity import episode_outlook_rows char_outlooks = episode_outlook_rows( self._db, self._project_id, episode_id) 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: # ★여기가 삼키는 자리다 — 이름 하나만 틀려도 합성 키가 통째로 # 안 만들어지고 아무도 모른다 (2026-09-04 실측). 그래서 무엇이 # 났는지 **종류까지** 적는다. logger.warning("Failed to build C##O## text map: %s: %s", exc.__class__.__name__, 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, # ★파생본을 거르지 않으면 옷 합성이 그 인물의 「기본」이 # 된다 — 의상 배정 O00(옷 없음) 씬에 옷이 실렸다. base_identity_ref_filter(), ) .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, base_identity_ref_filter(), ) .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 보존). """ # ★배경 sidecar 가 집행하는 갈래(location·location_part)는 **집행 계약**에서 # 받는다 — 종전엔 location 글자 하나만 빼서, 옛 LP 참조 자산이 있으면 실사 # sidecar 와 함께 새어 들어갔다 (Codex BLOCK 2026-09-02). 그 밖의 갈래 # (character·prop·outlook·종류 미상)는 종전과 같다. from app.modules.pipeline.grounding_entity_contract import sidecar_owner_types excluded = set(sidecar_owner_types()) 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") in excluded 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 반환. """ # ★보류(shelved) 제외 — 정본 하나 (2026-09-04). from app.core.entity_identity import active_episode_canon_ids ep_entity_ids = active_episode_canon_ids( self._db, self._project_id, episode_id) 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 # ★기본 `shot_dependency` CP(T2I 주석 없음)로 내려온 판 — ref_usage 가 원래 # 없다(빈 문자열은 malformed 가 아니라 **미주석**). close 샷은 declared no-ref # 와 같이 prev-shot 배경을 안 합성하고 None (실측 2026-09-02 stage2a S1_Shot2: # still_recipe off 라 t2i 스텝이 안 돌았는데 matrix 가 malformed 로 세웠다). if (_framing_scale == FRAMING_CLOSE and not bool((dep_detail_map.get(dep_key) or {}).get("t2i_annotated", True))): 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