"""Scene generation coordinator — scene_image_service의 bound method 집합을 분리. W5 F22 Phase B.25 (2026-04-23): SceneImageService의 9개 bound method(~720 LOC) + `_get_style_context` utility를 SceneGenerationCoordinator로 이관. facade는 self._coord를 주입받아 해당 메서드들에 위임. 모든 메서드는 literal lift (동작 보존). ## 내부 helper 메서드 - `_init_scene_gen_clients` — 4-client 초기화 (gemini/openai/sanitizer/tracker) - `_generate_scene_in_loop` — 씬 단위 N variation 생성 (thread pool) - `_run_fal_angle_pipeline` — fal.ai 앵글 N+1 이미지 생성 - `_finalize_and_track_primary` — primary 마킹 + 4-map tracking + checkpoint - `_finalize_scene_with_variations` — A/B variation 완료 후 recommended + commit + log - `_maybe_generate_variation_slot` — A/B variation 슬롯 기반 i2i - `_build_single_scene_prompt_and_refs` — non-custom_prompt 경로 prompt+refs - `_finalize_single_scene` — single scene 생성 후처리 (lineage + save + log + dict) - `_generate_variation_in_loop` — variation 1개 생성 (retry + sanitize) - `_get_style_context` — 프로젝트 세계관 → T2I prefix string 모든 메서드는 facade와 동일한 (db, project_id, actor_id, logger, 5 svc)를 주입받음. """ from __future__ import annotations import json import logging import uuid from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable, Dict, List, Literal, Optional, Tuple from sqlalchemy.orm import Session as OrmSession from app.core.config import settings from app.core.errors import AppError from app.core.file_paths import to_relative_image_path from app.core.framing_scale import FRAMING_CLOSE, get_framing_scale_or_raise from app.core.image_call_budget import bind_current_budget from app.core.ref_contract_validator import RefContractError from app.logging.activity_logger import ActivityLogger from app.models.project import ProjectSettings from app.modules.generation_tracker import GenerationTracker from app.modules.image_validator import ImageValidator from app.modules.llm.gemini_image_client import GeminiImageClient, ModerationError from app.modules.llm.openai_client import OpenAIClient from app.modules.prompt_sanitizer import PromptSanitizer from app.modules.semantic_contract_router import build_semantic_contract from app.services.fal_angle_helpers import ( apply_fal_angle as _apply_fal_angle, select_and_recommend_angle as _select_and_recommend_angle, ) from app.services.image_service_helpers import ( build_lineage_fields, image_to_dict, load_project_llm_config, ) from app.services.prompt_service import ( LabeledRefPayload, RefRoleError, build_final_scene_prompt as _build_final_scene_prompt, make_labeled_ref_payload, ) from app.services.scene_checkpoint_loaders import load_shot_t2i_variations from app.services.scene_persistence_service import ScenePersistenceService from app.services.scene_provenance_service import SceneProvenanceService from app.services.scene_reference_service import ( SceneReferenceService, apply_back_to_camera_constraints, build_image_index as _build_image_index_helper, rewrite_t2i_with_image_refs as _rewrite_t2i_helper, ) from app.services.scene_validation_service import SceneValidationService from app.services.scene_variation_service import ( SceneVariationService, set_variant_primary as _set_variant_primary_helper, ) __all__ = ["SceneGenerationCoordinator"] logger = logging.getLogger(__name__) def _now() -> str: return datetime.now(timezone.utc).isoformat() def _new_id() -> str: return str(uuid.uuid4()) def collect_actual_attached_refs(payload: Any) -> Dict[str, Any]: """P0 (2026-07-01): final scene payload 에서 **실제 Gemini 에 첨부된** image asset UUID + 상세를 수집(라벨 파싱 금지, 구조 metadata SOT). ref_role_metadata[i] 의 ``asset_id`` 만 실입력 lineage(UUID SOT) 로 본다. ★ ``bg_id`` 는 ImageAsset UUID 가 아니라 구조키(``L04B01`` variant_type) 이므로 UUID fallback 으로 쓰지 않는다(Codex BLOCKING) — bg 는 attach 시점에 chain_bg ImageAsset UUID 를 resolve 해 ``asset_id`` 로 stamp 한 경우만 lineage 에 포함. labeled_refs[i][0] 은 표시/감사용 라벨(edge SOT 아님). asset_id 없는 ref(prev_shot / bg UUID 미해결 등)는 unresolved 로 분리(input_image_ids 미포함). Returns dict: {image_ids: [uuid...], refs: [{asset_id, role, label}...], unresolved: [{role, label}...]} """ labeled = list(getattr(payload, "labeled_refs", None) or []) metas = list(getattr(payload, "ref_role_metadata", None) or []) roles = list(getattr(payload, "ref_roles", None) or []) image_ids: List[str] = [] refs: List[Dict[str, Any]] = [] unresolved: List[Dict[str, Any]] = [] for i, meta in enumerate(metas): if not isinstance(meta, dict): continue aid = meta.get("asset_id") # ★ bg_id 는 UUID 아님 → fallback 금지(Codex BLOCKING) role = meta.get("pipeline_role") or (roles[i] if i < len(roles) else "") label = labeled[i][0] if i < len(labeled) and labeled[i] else "" if aid: if aid not in image_ids: image_ids.append(aid) refs.append({"asset_id": aid, "role": role, "label": label}) else: # P0 (2026-07-01, Codex 합의): unresolved 에 구조필드 보존 → # persistence 가 라벨없이 post-hoc resolve(registered_pose_guide 등). # A5 (2026-07-02): source_still_id/anchor_source 추가 — immobilized # prev-frame ref 의 post-hoc resolve 구조키 (scene primary lookup). _u: Dict[str, Any] = {"role": role, "label": label} for _k in ("pipeline_role", "group_id", "bg_key", "guide_hash", "visible_focus", "subject_state", "source_still_id", "anchor_source"): if meta.get(_k) is not None: _u[_k] = meta.get(_k) unresolved.append(_u) return {"image_ids": image_ids, "refs": refs, "unresolved": unresolved} def _attached_lineage_fields(payload: Any) -> Dict[str, Any]: """collect_actual_attached_refs → var_result 키로 변환(P0).""" _c = collect_actual_attached_refs(payload) return { "actual_attached_image_ids": _c["image_ids"], "actual_attached_refs": _c["refs"], "unresolved_attached_refs": _c["unresolved"], } # ────────────────────────────────────────────────────────────────────── # lookup_render_prompt_card — single regen RPC inject (D1 patch) # 단건 generate_single_scene_image 경로가 scene_detail/manifest.json 의 # render_prompt_card 를 들고 와야 validate_attached_refs 의 required_refs # fail-fast 가 작동한다. lookup 실패 시 운영자-readable 메시지로 fail-fast. # ────────────────────────────────────────────────────────────────────── def lookup_render_prompt_card( project_id: str, episode_id: str, scene_index: int, shot_index: Optional[int], ) -> Dict[str, Any]: """scene_detail/manifest.json 에서 (scene_index, _shot_index) 매칭 RPC 반환. Manifest shape: data.scenes[*] (실제로 shot 단위 list). 매칭 키: scene_index + _shot_index (legacy None == None 허용). 실패 시 RefContractError (HTTP 422) — 운영자-readable msg. silent skip 금지. """ from app.core.ref_contract_validator import RefContractError cp_path = ( Path(settings.projects_dir) / project_id / "checkpoints" / "episodes" / episode_id / "scene_detail" / "manifest.json" ) msg_suffix = ( " — rerun scene_detail step or use an explicit custom_prompt to bypass." ) if not cp_path.exists(): raise RefContractError( f"scene_detail render_prompt_card missing for " f"scene={scene_index} shot={shot_index} (manifest file absent)" + msg_suffix ) try: cp = json.loads(cp_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise RefContractError( f"scene_detail render_prompt_card load failed for " f"scene={scene_index} shot={shot_index}: {exc}" + msg_suffix ) # iter1 IMPORTANT (Codex): cp/data/scenes/sc 구조가 expected dict/list 가 # 아니면 raw AttributeError 대신 RefContractError 변환 (4 failure-mode 가드 일관성). if not isinstance(cp, dict): raise RefContractError( f"scene_detail render_prompt_card malformed for " f"scene={scene_index} shot={shot_index} (manifest root must be " f"dict, got {type(cp).__name__})" + msg_suffix ) data = cp.get("data") if data is not None and not isinstance(data, dict): raise RefContractError( f"scene_detail render_prompt_card malformed for " f"scene={scene_index} shot={shot_index} (manifest 'data' must be " f"dict, got {type(data).__name__})" + msg_suffix ) scenes = (data or {}).get("scenes") if scenes is None: scenes = [] if not isinstance(scenes, list): raise RefContractError( f"scene_detail render_prompt_card malformed for " f"scene={scene_index} shot={shot_index} (manifest 'data.scenes' " f"must be list, got {type(scenes).__name__})" + msg_suffix ) for idx, sc in enumerate(scenes): if not isinstance(sc, dict): raise RefContractError( f"scene_detail render_prompt_card malformed for " f"scene={scene_index} shot={shot_index} (manifest " f"'data.scenes[{idx}]' must be dict, got " f"{type(sc).__name__})" + msg_suffix ) if sc.get("scene_index") != scene_index: continue if sc.get("_shot_index") != shot_index: continue rpc = sc.get("render_prompt_card") if not isinstance(rpc, dict): raise RefContractError( f"scene_detail render_prompt_card missing for " f"scene={scene_index} shot={shot_index} (shot row found but " f"render_prompt_card field absent or not a dict)" + msg_suffix ) return rpc raise RefContractError( f"scene_detail render_prompt_card missing for " f"scene={scene_index} shot={shot_index} (no matching shot row in manifest)" + msg_suffix ) # ────────────────────────────────────────────────────────────────────── # build_chain_bg_lookup — D5 §4.5 chain_bg lineage helper (T3) # T1 의 bg_map entry 확장 (bg_id + location_id) 을 활용해 추가 IO 0 으로 # bg_id → location_id 역참조 lambda 생성. validator 가 prev_shot lineage # substitute 판정에 사용. # ────────────────────────────────────────────────────────────────────── def waive_required_background_if_no_plate( rpc: Optional[Dict[str, Any]], background_chain_bg_map: Dict[str, Dict[str, Any]], still_data: Dict[str, Any], ) -> Optional[Dict[str, Any]]: """space_set_bg no-plate policy ↔ render_prompt_card required background 의 runtime reconciliation — validator 약화가 아니다 (Codex 합의 문구 그대로). W1-A (2026-06-11 fresh full E2E S10 실측): space_set_bg 가 해당 shot 을 unassigned/connector_no_plate 로 진단하면 loader 가 legacy chain bg 를 sentinel(`suppress_background_required`) 로 대체한다. 그 shot 의 render_prompt_card 가 여전히 legacy bg(L##B##) exact attach 를 요구하면 validator 가 차단하므로, 검증용 RPC '사본'에서 background required 만 제거한다 — "이 shot 은 background plate 를 쓰지 않는 것이 더 안전" 이라는 space policy 결론의 반영. persisted RPC 는 mutate 하지 않는다. silent waiver 금지 — 제거 내역을 warning 으로 기록. """ if not rpc: return rpc key = f"{still_data.get('scene_index', 0)}_{still_data.get('shot_index', 0)}" entry = (background_chain_bg_map or {}).get(key) or {} # W2-A (2026-06-11 S12 sh12 실측): dep_scene continuity 우선으로 bg map 을 # skip 한 shot 도 동일 reconciliation — required L## 가 미렌더 bg(L06 cascade) # 면 prev_shot lineage lookup 자체가 불가해 validator 가 차단한다. policy # 결정(prev_shot 이 이 shot 의 배경)의 반영이지 validator 약화가 아니다. _dep_waiver = bool(still_data.get("_dep_continuity_bg_waiver")) if not entry.get("suppress_background_required") and not _dep_waiver: return rpc asset_req = rpc.get("asset_requirements") if not isinstance(asset_req, dict): return rpc rr = asset_req.get("required_refs") import copy as _copy rpc2 = _copy.deepcopy(rpc) ar2 = rpc2["asset_requirements"] removed: List[str] = [] if isinstance(rr, list): kept = [] for e in ar2.get("required_refs") or []: if isinstance(e, dict) and e.get("kind") == "background": removed.append(str(e.get("id"))) else: kept.append(e) ar2["required_refs"] = kept remaining = kept elif isinstance(rr, dict): removed = [str(x) for x in (ar2.get("required_refs", {}).get("background") or [])] ar2["required_refs"] = { k: v for k, v in (ar2.get("required_refs") or {}).items() if k != "background" } remaining = [v for vs in ar2["required_refs"].values() for v in (vs or [])] else: return rpc if not removed: return rpc # required 가 전부 비면 readiness_policy=block_if_missing 이 validator 의 # drift 검사(step 5)에 걸린다 — waiver 결과로 빈 것이므로 None 으로 해제. if not remaining and ar2.get("readiness_policy") == "block_if_missing": ar2["readiness_policy"] = None _reason = ( entry.get("reason") if entry.get("suppress_background_required") else "dep_continuity_priority" ) logger.warning( "space_set_bg_no_plate background required waived (shot=%s reason=%s " "removed=%s) — runtime reconciliation, persisted RPC unchanged", key, _reason, removed, ) return rpc2 def resolve_dep_target_still_id( stills: Any, t2i_scene_index: Any, t2i_shot_index: Any, ) -> Optional[str]: """dep_detail(shot_dependency_t2i)이 명시한 (scene,shot) 타깃의 still id resolve. W3 (2026-06-11 fresh full E2E S29 실측): bytes 공급원(scene_still. dependent_scene_id = shot_dependency 산출)과 지시문 공급원(shot_dependency_t2i) 이 서로 다른 샷을 가리킬 수 있다 — S29 sh11 은 dependent_scene_id=sh4(사진 인서트), dep_detail=sh7(zoom) 로 'SAME FRAME' 지시가 sh4 프레임에 적용돼 프레임 복제가 발생했다. 지시문의 타깃 still 을 bytes 로 써야 정합. still 항목은 dict(서비스 계약) 또는 ORM 양쪽 지원. 미발견 시 None. """ if t2i_scene_index is None or t2i_shot_index is None: return None for s in stills or []: if isinstance(s, dict): si, sh, sid = s.get("scene_index"), s.get("shot_index"), s.get("id") else: si = getattr(s, "scene_index", None) sh = getattr(s, "shot_index", None) sid = getattr(s, "id", None) if si == t2i_scene_index and sh == t2i_shot_index and sid: return sid return None def build_chain_bg_lookup(background_chain_bg_map: Dict[str, Dict[str, Any]]): """T1 entry 확장 (`bg_id`, `location_id`) 활용 — 추가 IO 없이 bg_id→loc lambda 반환. Entry 가 T1 확장 전 (bg_id 또는 location_id 부재) 이면 lookup 에서 자연스럽게 제외 — 사용자 binding G3 와 정합 (lineage 확인 불가 시 prev_shot 통과 X). """ bg_to_loc = { entry["bg_id"]: entry["location_id"] for entry in background_chain_bg_map.values() if entry.get("bg_id") and entry.get("location_id") } return lambda bg_id: bg_to_loc.get(bg_id) # ────────────────────────────────────────────────────────────────────── # _normalize_background_phrase_kind — FINDING C W2 (Category A) # ────────────────────────────────────────────────────────────────────── def _normalize_background_phrase_kind( reference_phrase_kinds: List[str], attached_meta: List[Tuple[str, str]], rpc: Optional[Dict[str, Any]], ) -> List[str]: """FINDING C W2 (Category A) — background over-declaration consumer normalization. scene_detail LLM 이 [L##: ...] free-form location 묘사 블록을 보고 per-variation sidecar `reference_phrase_kinds` 에 'background' 를 over-declare 할 수 있다. background ref attach 여부는 coordinator runtime SOT (chain_bg map lookup + build_prev_shot_background_ref) 이므로 producer 는 예측 불가 — attached_meta 가 확정된 consumer boundary 에서 normalize. 'background' strip 3-조건 (모두 만족 시): 1. reference_phrase_kinds 에 'background' 있음. 2. attached_meta 에 ('background', *) / ('background_prev_shot', *) 없음. 3. rpc.asset_requirements.required_refs 에 kind='background' 없음. 조건 3 = genuine missing REQUIRED background 는 mask 안 함 — required background 이 있는데 미attach 면 validator step 4 가 step 6 전에 fail-fast. validator step 6 (sidecar phantom guard) 자체는 무변경 — over-declaration 만 producer/runtime 불일치로 보고 consumer 가 정합화한다. 약화 0. """ if "background" not in reference_phrase_kinds: return list(reference_phrase_kinds) _attached_kinds = {k for k, _ in attached_meta} if "background" in _attached_kinds or "background_prev_shot" in _attached_kinds: return list(reference_phrase_kinds) _required_refs = ( (rpc or {}).get("asset_requirements") or {} ).get("required_refs") or [] _has_bg_required = any( isinstance(r, dict) and r.get("kind") == "background" for r in _required_refs ) if _has_bg_required: return list(reference_phrase_kinds) return [k for k in reference_phrase_kinds if k != "background"] # ────────────────────────────────────────────────────────────────────── # build_scene_attached_refs — Task 1 of single-vs-batch reference contract fix # spec: docs/superpowers/specs/2026-05-08-single-batch-reference-contract-design.md §4.1 # # 단건/배치 양쪽 ref builder 공통화. chain_bg / prev_shot_ref / state_variant / # close framing skip 정책 일관 적용. # ────────────────────────────────────────────────────────────────────── def _make_osl_source_bytes_resolver( stills: Optional[List[Any]], scene_paths_by_index_by_id: Dict[str, Path], ): """W21B-W8 option C — composition continuity_anchor 의 anchor_source (si,shi) → 그 still 의 **현재-run primary bytes** 해석 클로저. `scene_paths_by_index_by_id` 는 메인 스레드가 batch barrier 에서 갱신하는 still_id → primary path 맵 (dep edge 가 source 선생성 보장). source still 의 현재-run 프레임이 없으면 None 반환 → consumer 가 stale fallback 없이 no-op (option C 계약). list/path read 만 — db 접근 0 (worker thread 안전).""" _key_to_sid: Dict[Tuple[int, int], str] = {} for _s in stills or []: try: _si, _shi = _s.get("scene_index"), _s.get("shot_index") except AttributeError: continue if isinstance(_si, int) and isinstance(_shi, int) and _s.get("id"): _key_to_sid[(_si, _shi)] = _s["id"] def _resolve(src_key: Tuple[int, int]) -> Optional[bytes]: sid = _key_to_sid.get((src_key[0], src_key[1])) if not sid: return None path = scene_paths_by_index_by_id.get(sid) if path is None or not path.exists(): return None try: return path.read_bytes() except Exception: return None return _resolve def _direct_resolver_prompts( od: Optional[Dict[str, Any]], fallback_prompts: List[str], ) -> List[str]: """W22 직행 — resolver 입력 = [9절] + scene_detail 원 프롬프트 union. W4a NARROW_1 은 [9절] 단독이었으나 fresh E2E 에서 직행 샷 전건 REF_CONTRACT_VIOLATION 실측(2026-07-10, 19 shots): resolver 의 캐릭터 매칭은 복합 short_id 정규식 (C\\d{2,3})(O\\d{2,3}) 로 프롬프트를 스캔하는데 9절은 설계상 ID-free 라 캐릭터/아웃룩 ref 가 구조적으로 0건 첨부됐다. ref 계약(reference_phrase_kinds)의 원천이 scene_detail 이므로 캐릭터/prop 해석도 ID 를 보유한 scene_detail 프롬프트가 담당하고, 9절은 생성 프롬프트(var_t2i)로만 쓴다 — s33 계약(ref=[실사+맵]+passport)과 정합. """ if od and od.get("prompt"): return [od["prompt"], *fallback_prompts] return fallback_prompts def _insert_outdoor_canon_refs( labeled_refs: List, attached_meta: List, ref_roles: List, ref_role_metadata: List, od: Dict[str, Any], *, scene_index: Any, shot_index: Any, where: str, ) -> None: """W22 직행 — 캐논 2장(index0=실사 마스터[룩 SOT], index1=맵[배치 전용]) 주입. 단건/배치 공용 helper (Codex W4a MINOR — drift 방지). 4-list 평행 insert (length-parity invariant). attached kind="background" (value `outdoor_canon:`) — declared/required background 충족 (ref_contract_validator waiver 참조). asset_id metadata stamp → 최종 scene asset input_image_ids lineage. """ from app.modules.pipeline.outdoor_direct_compose import ( LOCATION_PHOTO_LABEL, SITE_PLAN_LABEL, ) _bg_value = f"outdoor_canon:{od['place_id']}" labeled_refs.insert(0, (SITE_PLAN_LABEL, od["map_bytes"])) attached_meta.insert(0, ("background", _bg_value)) ref_roles.insert(0, "outdoor_canon_map_ref") _map_meta: Dict[str, Any] = { "pipeline_role": "outdoor_canon_map", "place_id": od["place_id"], } if od.get("map_asset_id"): _map_meta["asset_id"] = od["map_asset_id"] ref_role_metadata.insert(0, _map_meta) labeled_refs.insert(0, (LOCATION_PHOTO_LABEL, od["master_bytes"])) attached_meta.insert(0, ("background", _bg_value)) ref_roles.insert(0, "outdoor_canon_photo_ref") _photo_meta: Dict[str, Any] = { "pipeline_role": "outdoor_canon_photo", "place_id": od["place_id"], } if od.get("prompt_version"): _photo_meta["direct_prompt_version"] = od["prompt_version"] if od.get("master_asset_id"): _photo_meta["asset_id"] = od["master_asset_id"] ref_role_metadata.insert(0, _photo_meta) logger.info( "Scene %s Shot %s: outdoor direct canon refs injected (%s, place=%s)", scene_index, shot_index, where, od["place_id"], ) def build_scene_attached_refs( *, still: Any, episode_id: str, still_data: Dict[str, Any], visible_entities: List[Dict[str, Any]], ref_image_map: Dict[str, bytes], # per-episode cached (caller 가 1회 build 후 reuse — N+1 회피) cached_style_context: str, cached_entity_text_map: Dict[str, str], scene_paths_by_index_by_id: Dict[str, Path], location_scene_history: Dict[str, Any], background_chain_bg_map: Dict[str, Any], dep_detail_map: Dict[str, Any], staging: Optional[Dict[str, Any]], entity_lookup: Dict[str, Dict], project_id: str, project_config: Dict[str, Any], reference_svc: Any, stills: Optional[List[Any]] = None, var_t2i_override: Optional[str] = None, indoor_pose_ctx: Optional[Dict[str, Any]] = None, outdoor_direct_ctx: Optional[Dict[str, Any]] = None, return_payload: bool = False, ) -> "tuple": """spec §4.1 — 배치 generate_images 의 ref 빌드 블록(coordinator.py:178-326) 동등 보존. chain_bg / prev_shot_ref / state_variant 통합. 5a~5d 분기. v2 (audit IMPORTANT 1): - state_variant_sids 는 helper 내부 detect / best_prev_bytes + bytes_source_kind 는 coordinator 가 resolve (provenance annotator, spec docs/superpowers/specs/2026-05-15-zoom-in-detail-source-provenance-design.md §3). - cached_style_context + cached_entity_text_map 은 caller 가 episode 단위 build 후 전달 (배치 N stills 시 N×rebuild 회귀 방지). D5 §4.3 (2026-05-09): return 을 (full_prompt, labeled_refs, attached_meta) 3-tuple. attached_meta = list[(kind, id)] — chain_bg / prev_shot / character / character_outlook / character_state / prop 모두 source-of-truth 에서 record. invariant 1: len(labeled_refs) == len(attached_meta). Area #5 W3 §4.3 (2026-05-18): return 을 4-tuple 로 확장. `reference_phrase_kinds` 는 scene_detail v26 producer (W1) 가 each t2i_variations entry 에 emit 한 sidecar enum list — phantom guard 가 sidecar exact compare 로 step 6 검사 (ref_contract_validator.py:295-334). single path 는 helper 의 t2i_variations[0] sidecar 를 그대로 unpack 받아 validate_attached_refs 에 전달. variation path 는 별도 (caller 가 var dict 에서 direct read 후 _generate_variation_in_loop 에 전달). Returns (default, return_payload=False): 4-tuple (full_prompt, labeled_refs, attached_meta, reference_phrase_kinds). P0 (2026-07-01, Codex NARROW_1): return_payload=True 면 5-tuple (..., _payload[LabeledRefPayload, ref_role_metadata 에 asset_id stamp 포함]). 단건 경로(_build_single_scene_prompt_and_refs)만 True — legacy caller/test 는 기본 4-tuple 유지(byte-identical). """ # 1. var_t2i — load_shot_t2i_variations t2i_variations = load_shot_t2i_variations( settings.projects_dir, project_id, episode_id, camera_json=still.camera_json, scene_index=still.scene_index, still_index=still.still_index, shot_index=still.shot_index, ) # FINDING 11 (e2e-bughunt-v1): consume-side 'character' over-declaration # normalization. required_refs SOT 를 보장하려고 helper 호출 전에 # render_prompt_card 를 resolve — single regen 경로는 still_data 에 RPC 가 # 아직 없을 수 있다. inject 후 후단 lookup 은 read-through 가 된다. 배치 # path _generate_scene_in_loop 의 rpc resolution 패턴과 동등. _rpc_f11 = still_data.get("render_prompt_card") if _rpc_f11 is None: _rpc_f11 = lookup_render_prompt_card( project_id=project_id, episode_id=episode_id, scene_index=still.scene_index, shot_index=still.shot_index, ) still_data["render_prompt_card"] = _rpc_f11 from app.core.steps.detail_steps import ( _strip_overdeclared_character_phrase_kind, ) _strip_overdeclared_character_phrase_kind(t2i_variations, _rpc_f11) var_t2i = ( t2i_variations[0].get("t2i_prompt", still_data["still_frame_prompt"]) if t2i_variations else still_data["still_frame_prompt"] ) # W21B-W7 W-C2 (2026-06-12): zoom_continuity_anchor 의 source_wide # image-phase prompt override (Codex guard: scene_detail cp 불변 — # 실사용 프롬프트는 prompt_used 로, hash provenance 는 anchor cp 로 추적). # revised 는 W-C1 토큰 audit 통과본만 — 새 ref 요구 없어 아래 # reference_phrase_kinds sidecar / required_refs 검증이 그대로 유효하다. if var_t2i_override is not None: var_t2i = var_t2i_override # W22 직행 (2026-07-10): 야외 직행 샷 — scene_detail t2i 대신 s33 9절 # 결정론 조립 프롬프트로 교체 (설계 §7-4 사용자 확정). 명시 override # (custom/zoom)가 있으면 그쪽 우선. ctx 부재/flag OFF = no-op byte-identical. _od = (outdoor_direct_ctx or {}).get( f"{still.scene_index}_{still.shot_index}" ) if _od and _od.get("prompt") and var_t2i_override is None: var_t2i = _od["prompt"] # Area #5 W3 §4.3 (Codex iter 2 BLOCKING fix): sidecar `reference_phrase_kinds` # 를 var_t2i 와 동일 source (t2i_variations[0]) 에서 direct read. producer # (scene_detail v26 W1) emit 의무. No Silent Fallback — stale v25 cp 가 # sidecar 없이 들어오면 RefContractError fail-fast (spec §4.3 + plan Task 3.1). # `.get(..., [])` silent fallback 절대 금지. if t2i_variations: _first_var = t2i_variations[0] if "reference_phrase_kinds" not in _first_var: raise RefContractError( "scene_detail t2i_variations[0] missing 'reference_phrase_kinds' " "sidecar — producer (scene_detail v26 W1) emit required. " "stale v25 checkpoint? Run alembic upgrade + scene_detail re-run " "to bump producer. No Silent Fallback (Area #5 W3 §4.3)" ) _phrase_kinds_raw = _first_var["reference_phrase_kinds"] if _phrase_kinds_raw is None: raise RefContractError( "scene_detail t2i_variations[0].reference_phrase_kinds is None — " "producer emit malformed. No Silent Fallback (Area #5 W3 §4.3)" ) # Codex iter 3 IMPORTANT fix: non-list malformed sidecar 가 list(...) 변환 # 으로 validator 의 "must be list" fail-fast 우회 방지. spec §4.3 isinstance # gate 명시. if not isinstance(_phrase_kinds_raw, list): raise RefContractError( f"scene_detail t2i_variations[0].reference_phrase_kinds " f"malformed: must be list, got " f"{type(_phrase_kinds_raw).__name__}={_phrase_kinds_raw!r}. " f"No Silent Fallback (Area #5 W3 §4.3)" ) reference_phrase_kinds: list = list(_phrase_kinds_raw) else: # t2i_variations 자체가 비어 있으면 still_frame_prompt 단독 fallback path — # 이 경우 phantom guard 진입점 없음 (labeled_refs/meta 도 빈 list). reference_phrase_kinds = [] # Fix A (2026-05-10): 단건 regen path 도 target_variations union 으로 ref # build — first variation 만 보고 attach 하던 결함 fix. _max_t2i 슬라이스 # 는 batch path 와 동등하게 settings.scene_variation_count 적용. _max_t2i = settings.scene_variation_count _target_raw = [v for v in (t2i_variations or []) if v.get("t2i_prompt")][-_max_t2i:] if _target_raw: _ref_prompts = [v.get("t2i_prompt", "") for v in _target_raw] else: # fallback — t2i_variations 없거나 모두 빈 prompt 면 still_frame_prompt 단독 _fallback = still_data.get("still_frame_prompt", "") _ref_prompts = [_fallback] if _fallback else [] # W22 직행 (Codex W4a NARROW_1): 직행 샷은 resolver 도 9절 기준 — # 실제 생성 프롬프트(var_t2i=9절)와 ref resolve 입력 불일치 차단. _ref_prompts = _direct_resolver_prompts(_od, _ref_prompts) # 단건 regen path 의 still_data 가 scene_index/shot_index 를 누락할 때 # still ORM 값으로 helper-local shape 를 정규화한다. chain_bg key뿐 아니라 # prev_shot/dependency lookup과 log도 같은 indices 를 보게 하기 위함. _scene_idx = still_data.get("scene_index") if _scene_idx is None: _scene_idx = getattr(still, "scene_index", 0) or 0 still_data["scene_index"] = _scene_idx _shot_idx = still_data.get("shot_index") if _shot_idx is None: _shot_idx = getattr(still, "shot_index", 0) or 0 still_data["shot_index"] = _shot_idx if "dependent_scene_id" not in still_data: still_data["dependent_scene_id"] = getattr(still, "dependent_scene_id", None) # 2. scene_ref_image_map (location 자동 제외) # P0 (2026-07-01): scene_ref_asset_id_map = scene_ref_image_map 과 동일 key → # 실제 ImageAsset UUID. resolve_refs 가 attach 시점에 asset_id stamp → 최종 # scene asset input_image_ids lineage SOT. scene_ref_asset_id_map: Dict[str, str] = {} scene_ref_image_map = reference_svc.build_scene_ref_image_map( ref_image_map, entity_lookup, out_asset_id_map=scene_ref_asset_id_map, ) # 3. (v2) state_variant detect — helper 내부 (batch line 245-251 동등) # identity-variant aware (2026-07-02): staging=base 표기 vs VE=variant 표기 # 어긋남을 entity_relation family 로 브리지 (관계 없으면 빈 dict=기존 동작). state_variant_sids = reference_svc.detect_state_variant_sids( visible_entities, entity_lookup, scene_ref_image_map, staging, identity_family_by_sid=reference_svc.load_identity_family_by_sid(episode_id), ) if state_variant_sids: logger.info( "Scene %d Shot %d: state_variant=%s", still_data.get("scene_index", 0), still_data.get("shot_index", 0), list(state_variant_sids.keys()), ) # 4. resolve_refs_for_prompt_set (entity-only) — Fix A (2026-05-10): # target_variations union 으로 ref build (legacy single-prompt 회귀 차단). # Patch A — RPC.asset_requirements.required_refs 추출 후 resolver 에 전달 # → Tier 2 forced attach (required prop 강제 binding). _patch_a_required_refs = ( ((still_data.get("render_prompt_card") or {}).get("asset_requirements") or {}) .get("required_refs") or [] ) # W21B-W8 option C (2026-06-15): composition continuity guide context 를 1회 # 로드 — (a) force_character_names(이 shot 의 staged 강제 attach 대상, step 이 # this-shot ∩ anchor_source staged 교집합으로 precompute)를 resolver 에 thread, # (b) 아래 attach 단계의 mode/anchor_source/source bytes 해석에 재사용. 두 flag # OFF/guide 부재 시 빈 dict → 빈 set = default 경로 byte-identical. from app.core.steps.outdoor_site_layout_step import ( load_composition_guide_context, ) _cg_ctx = load_composition_guide_context(project_id, episode_id) _cg_entry = (_cg_ctx or {}).get( (still_data.get("scene_index"), still_data.get("shot_index"))) or {} _force_char_names = set(_cg_entry.get("forced_character_names") or []) # Area #11 v1 W2: 2-tuple → LabeledRefPayload return (spec §3.3). _payload = reference_svc.resolve_refs_for_prompt_set( t2i_prompts=_ref_prompts, visible_entities=visible_entities, scene_ref_image_map=scene_ref_image_map, entity_lookup=entity_lookup, state_variant_sids=state_variant_sids, required_refs=_patch_a_required_refs, # option C: forced staged character attach 는 staged-character backstop # (resolve_refs_for_prompt 의 `if staging:` 블록)에서 발화한다 → guide 샷 # (force_character_names 존재)에 한해 staging 을 함께 넘긴다. 비-guide 샷은 # staging=None 으로 backstop 휴면 = default 경로 byte-identical. staging=staging if _force_char_names else None, force_character_names=_force_char_names or None, scene_ref_asset_id_map=scene_ref_asset_id_map, ) labeled_refs = list(_payload.labeled_refs) attached_meta = list(_payload.attached_meta) ref_roles = list(_payload.ref_roles) ref_role_metadata = list(_payload.ref_role_metadata) # B-run S15 실측 fix (2026-06-12): staging back_to_camera enum 인물의 # character/outfit ref 에 view_angle_constraint stamp (좁은 helper — # 휴면 W2 backstop 미활성, staging 부재 시 no-op). apply_back_to_camera_constraints( attached_meta, ref_roles, ref_role_metadata, staging=staging, visible_entities=visible_entities, entity_lookup=entity_lookup, ) # 5. best_prev_bytes + bytes_source_kind resolve (coordinator = provenance annotator). # spec docs/superpowers/specs/2026-05-15-zoom-in-detail-source-provenance-design.md §3 §6. current_location_ids = [ e["id"] for e in visible_entities if e.get("entity_type") == "location" ] # B (2026-07-02): outdoor prev-frame 의무첨부 컨텍스트 — flag OFF → {} no-op. # VE 에 location 이 없는 outdoor 샷의 history fallback + required 진단에 쓴다. from app.core.steps.outdoor_site_layout_step import ( load_outdoor_prev_frame_context, ) from app.modules.pipeline.outdoor_site_layout_plan import ( build_outdoor_prev_frame_diag, primary_location_uuid_by_scene, resolve_outdoor_history_fallback_uuid, ) _opf_ctx = load_outdoor_prev_frame_context(project_id, episode_id) _opf_uuid_by_scene = primary_location_uuid_by_scene( _opf_ctx.get("primary_location_by_scene") or {}, _opf_ctx.get("outdoor_loc_sids") or set(), entity_lookup, ) if _opf_ctx else {} _opf_hist_still: Optional[Dict[str, Any]] = None best_prev_bytes = None bytes_source_kind: Literal["dep_scene", "location_history", "none"] = "none" dep_scene_id = still_data.get("dependent_scene_id") # W3 (2026-06-11, Codex 합의): dep_detail(shot_dependency_t2i)이 명시한 타깃 # still 을 bytes 공급원으로 우선 — dependent_scene_id(shot_dependency 산출)와 # 불일치 시 'SAME FRAME' 지시가 엉뚱한 프레임에 적용된다 (S29 sh11 실측). # 타깃 still PNG 부재 시 기존 dependent_scene_id 그대로 (backward-compat). _w3_dep_key = ( f"{still_data.get('scene_index', 0)}_{still_data.get('shot_index', 0)}" ) _w3_dep_info = (dep_detail_map or {}).get(_w3_dep_key) or {} _w3_target_id = resolve_dep_target_still_id( stills, _w3_dep_info.get("dep_scene_index"), _w3_dep_info.get("dep_shot_index"), ) if _w3_target_id and _w3_target_id in scene_paths_by_index_by_id \ and scene_paths_by_index_by_id[_w3_target_id].exists(): if dep_scene_id and dep_scene_id != _w3_target_id: logger.info( "Scene %s Shot %s: dep bytes target realigned to dep_detail " "(%s → %s) — W3 source coherence", still_data.get("scene_index"), still_data.get("shot_index"), dep_scene_id, _w3_target_id, ) dep_scene_id = _w3_target_id if dep_scene_id and dep_scene_id in scene_paths_by_index_by_id: dep_path = scene_paths_by_index_by_id[dep_scene_id] if dep_path.exists(): best_prev_bytes = dep_path.read_bytes() bytes_source_kind = "dep_scene" # W20F10 O policy: ref_usage == "zoom_in_detail" 인 shot 은 dep_scene only. # dep_scene PNG 부재 시 location_history 로 silent fallback 하면 callee # (scene_reference_service.py:994 Layer 3 invariant) 에서 RefContractError 가 # cap exceeded 본질을 가린다. ref_usage 는 shot_dependency_t2i 의 dep_detail_map # 에서 scene_index/shot_index 키로 조회 (문자열 의미 추론 X). 다른 ref_usage 는 # 기존 fallback 그대로. _dep_key_for_policy = ( f"{still_data.get('scene_index', 0)}_{still_data.get('shot_index', 0)}" ) _ref_usage_for_policy = ( (dep_detail_map or {}).get(_dep_key_for_policy, {}) or {} ).get("ref_usage", "") if not best_prev_bytes and _ref_usage_for_policy != "zoom_in_detail": for loc_id in current_location_ids: if loc_id in location_scene_history: best_prev_bytes, _opf_hist_still = location_scene_history[loc_id] bytes_source_kind = "location_history" break # B (2026-07-02): VE 에 location 이 없어 위 lookup 을 놓친 outdoor 샷 — # scene_director.primary_location 구조 fallback (short_id→UUID 역매핑). # hit 시 current_location_ids 에 보강해 Layer 2.6 provenance 가드와 정합. if not best_prev_bytes and _opf_uuid_by_scene: _opf_fb = resolve_outdoor_history_fallback_uuid( still_data.get("scene_index"), primary_loc_uuid_by_scene=_opf_uuid_by_scene, location_scene_history=location_scene_history, ) if _opf_fb: best_prev_bytes, _opf_hist_still = location_scene_history[_opf_fb] bytes_source_kind = "location_history" if _opf_fb not in current_location_ids: current_location_ids.append(_opf_fb) logger.info( "outdoor_prev_frame: S%ssh%s history fallback — primary_" "location UUID 로 prev 프레임 확보 (VE location 부재 보정)", still_data.get("scene_index"), still_data.get("shot_index")) elif not best_prev_bytes and _ref_usage_for_policy == "zoom_in_detail": logger.warning( "scene_generation_coordinator: %s zoom_in_detail dep_scene bytes " "missing — location_history fallback skipped (W20F10 O policy)", _dep_key_for_policy, ) # 6. _is_close_framing 검사 (shot_staging.framing_scale enum SOT) _framing_scale = get_framing_scale_or_raise( staging, where="scene_generation_coordinator.build_scene_attached_refs", ) _is_close_framing = _framing_scale == FRAMING_CLOSE # 7. background fallback chain — spec §4.1 5a~5d 분기 # W2 (2026-06-11 fresh full E2E 육안 피드백, Codex 합의): dep_scene # continuity 가 있는 shot 은 bg map(빈 establishing plate/chain bg)이 # prev_shot continuity ref 를 대체하지 않는다 — 시신 자세·핏자국·직전 동작 # 같은 dynamic state 는 prev_shot 에만 있다 (S12 sh7↔sh12 자세 불일치 실측). # 우선순위: dep_scene continuity > bg map(space plate/chain) > location_history. # W2-B 정련 (재생성 육안 반복): dep 우선은 dynamic-state usage 에만 — # zoom_in_detail(같은 프레임 줌)·exact_background(prev 배경 그대로). mood 만 # 제공하는 atmosphere_reference 가 환경 정체성 bg 를 밀어내면 모델이 환경을 # 발명한다 (S10 sh5 실측: 떠 있는 유리판). _w2_ref_usage = ( (dep_detail_map or {}).get(f"{_scene_idx}_{_shot_idx}", {}) or {} ).get("ref_usage", "") _has_dep_continuity = ( bytes_source_kind == "dep_scene" and _w2_ref_usage in ("zoom_in_detail", "exact_background") ) _bc_key = f"{_scene_idx}_{_shot_idx}" _bc_bg = background_chain_bg_map.get(_bc_key) # W22 직행 consistency fix (2026-07-10, 2회차 E2E 육안 — 같은 장소 샷 간 # 일관성 붕괴 실측): 직행도 아래 else 블록의 prev_shot 연속성 기계를 그대로 # 탄다(직행 최초 구현은 이 분기를 통째로 대체해 같은 장소 직전 스틸 앵커가 # 소실). 캐논 2장은 분기 뒤에 최상단 insert — 최종 순서=[실사, 맵, prev, ...]. if (not _od) and ( _bc_bg and _bc_bg.get("image_bytes") and not _is_close_framing and not _has_dep_continuity ): # 5a: chain_bg + NOT close → insert (batch line 285 동등) # Area #11 v1 W2 (Codex iter 2 Important 3 fix): fail-fast strict on missing bg_id # (No Silent Fallback gate, payload length-parity invariant 일관). _bg_id = _bc_bg.get("bg_id") if not _bg_id: raise RefRoleError( "chain_bg missing bg_id (single path) — No Silent Fallback gate (Area #11 v1)" ) labeled_refs.insert(0, (_bc_bg["label"], _bc_bg["image_bytes"])) # D5 §4.2 P1: bg_id 는 T1 의 entry 확장에서 SOT read (라벨 파싱 X). attached_meta.insert(0, ("background", _bg_id)) # Area #11 v1 W2: parallel sidecar insert ref_roles.insert(0, "background_chain_ref") # P0 (2026-07-01, Codex BLOCKING fix): bg_id(예 L04B06)는 UUID 아님 → # chain_bg ImageAsset UUID 를 구조키로 resolve 해 asset_id 로 stamp # (input_image_ids lineage SOT). 미해결이면 asset_id 생략 → unresolved. _bg_meta = {"bg_id": _bg_id, "pipeline_role": "background_render"} _bg_asset_uuid = reference_svc.resolve_chain_bg_asset_id(episode_id, _bg_id) if _bg_asset_uuid: _bg_meta["asset_id"] = _bg_asset_uuid ref_role_metadata.insert(0, _bg_meta) logger.info( "Scene %d Shot %d: background_chain ref injected (bg_id=%s)", still_data.get("scene_index", 0), still_data.get("shot_index", 0), _bg_id, ) else: # 5b: chain_bg + close → chain skip + prev_shot try # 5c: chain 부재 → prev_shot try # 5e(W2): chain_bg + dep_scene continuity → chain skip + prev_shot(dep) 우선 if _bc_bg and _bc_bg.get("image_bytes") and _has_dep_continuity and not _is_close_framing: logger.info( "Scene %d Shot %d: bg map ref SKIPPED — dep_scene continuity " "priority (W2)", still_data.get("scene_index", 0), still_data.get("shot_index", 0), ) # W2-A: skip 결정의 결과로 required background 가 미충족될 수 있다 # (특히 chain bg 미렌더 L## — lineage lookup 불가). policy 결정의 # runtime reconciliation 신호 — validate 직전 waiver 로 소비. still_data["_dep_continuity_bg_waiver"] = True elif _bc_bg and _bc_bg.get("image_bytes") and _is_close_framing: logger.info( "Scene %d Shot %d: chain_bg ref SKIPPED — close framing " "(framing_scale=%r)", still_data.get("scene_index", 0), still_data.get("shot_index", 0), _framing_scale, ) try: _prev_shot_ref = reference_svc.build_prev_shot_background_ref( best_prev_bytes=best_prev_bytes, bytes_source_kind=bytes_source_kind, still_data=still_data, visible_entities=visible_entities, current_location_ids=current_location_ids, dep_scene_id=dep_scene_id, stills=stills or [], location_scene_history=location_scene_history, dep_detail_map=dep_detail_map, staging=staging, state_variant_sids=state_variant_sids, entity_lookup=entity_lookup, ) except RefContractError: raise except TypeError: # spec §6 — signature 위반 (bytes_source_kind 누락 등) fail-fast (No Silent Fallback Gate) raise except Exception as exc: logger.warning( "Scene %d Shot %d: prev_shot_ref build failed (%s) — " "entity-only fallback", still_data.get("scene_index", 0), still_data.get("shot_index", 0), exc, ) _prev_shot_ref = None # B (2026-07-02): prev 프레임 source still 구조키 — dep=dep_scene_id / # history=hit entry 의 still id (진단 + lineage post-hoc resolve 용). # ★composition continuity_anchor(option C) 샷은 제외 — 그 lane 이 prev-shot # ref bytes 를 그룹 anchor 프레임으로 교체하므로 여기 구조키가 남으면 # 다른 still 로 오해석된 lineage 가 생긴다 (diag 도 option C lane 소유). _b_src_sid = ( dep_scene_id if bytes_source_kind == "dep_scene" else ((_opf_hist_still or {}).get("id") if bytes_source_kind == "location_history" else None) ) _b_cg_anchor = ( ((_cg_ctx or {}).get( (still_data.get("scene_index"), still_data.get("shot_index"))) or {}).get("mode") == "continuity_anchor" ) if _prev_shot_ref: # Area #11 v1 W2: helper 가 5-tuple — ref_role + ref_role_metadata 동시 unpack. _label, _bytes, _loc_id, _role, _metadata = _prev_shot_ref # B: flag ON + source still 확정 시 lineage 구조키 stamp (persistence # post-hoc resolve → input_image_ids 엣지, A5 와 동일 패턴). OFF = 불변. if _opf_ctx and _b_src_sid and not _b_cg_anchor: _metadata = {**_metadata, "pipeline_role": "scene_prev_frame", "source_still_id": str(_b_src_sid)} if _od: # W22 직행 place-continuity (Codex BLOCKING_1): consumer 가 전용 # 지시 분기(고정 디테일 연속성만, 조명 복사 금지)를 타도록 마킹. _metadata = {**_metadata, "outdoor_direct_place_continuity": True} labeled_refs.insert(0, (_label, _bytes)) attached_meta.insert(0, ("background_prev_shot", _loc_id)) # Area #11 v1 W2: parallel sidecar insert (4 list 모두 length parity 유지) ref_roles.insert(0, _role) ref_role_metadata.insert(0, _metadata) # 5d: prev_shot 도 None → entity-only (이미 labeled_refs 에 character/prop 만) # B (2026-07-02): outdoor prev-frame 의무첨부 진단 — outdoor+plate無+신호 # 샷에서 미해결 시 previous_frame_required_missing (WARNING+metadata 영속, # hard fail 아님 — Codex 합의). 대상 아니면 diag None = 불변. # option C continuity_anchor 샷은 그 lane 이 prev 프레임을 소유 — 제외. if _opf_ctx and not _b_cg_anchor: _b_ve_loc_sids = { _sid for _lid in current_location_ids if isinstance( (_sid := (entity_lookup.get(_lid) or {}).get("short_id")), str) and _sid } _b_dep_key = ( f"{still_data.get('scene_index', 0)}_{still_data.get('shot_index', 0)}" ) _b_si, _b_shi = still_data.get("scene_index"), still_data.get("shot_index") _b_diag = build_outdoor_prev_frame_diag( scene_index=_b_si, shot_index=_b_shi, outdoor_loc_sids=_opf_ctx.get("outdoor_loc_sids") or set(), primary_location_by_scene=( _opf_ctx.get("primary_location_by_scene") or {}), ve_location_sids=_b_ve_loc_sids, chain_bg_attached=False, ref_usage=_w2_ref_usage, has_dep=bool(dep_scene_id or (dep_detail_map or {}).get(_b_dep_key)), has_same_scene_prior=any( s_.get("scene_index") == _b_si and isinstance((_p_shi := s_.get("shot_index")), int) and isinstance(_b_shi, int) and _p_shi < _b_shi for s_ in (stills or []) ), resolved=bool(_prev_shot_ref), bytes_source_kind=bytes_source_kind, source_still_id=str(_b_src_sid) if _b_src_sid else None, close_framing=_is_close_framing, ) if _b_diag: still_data["_prev_frame_chain_diag"] = _b_diag if not _b_diag["resolved"]: logger.warning( "outdoor_prev_frame: S%ssh%s previous_frame_required_missing " "— outdoor+plate無+연속성 신호 샷인데 prev 프레임 미해결 " "(dep=%s, same_scene_prior=%s)", _b_si, _b_shi, _b_diag["required_signals"]["dep"], _b_diag["required_signals"]["same_scene_prior"]) if _od: # W22 직행: 캐논 2장 최상단 삽입 — prev 연속성 ref 뒤에 실행되므로 # 최종 순서=[실사(룩 SOT), 맵(배치), prev(연속성), ...] (s33 계약 유지). _insert_outdoor_canon_refs( labeled_refs, attached_meta, ref_roles, ref_role_metadata, _od, scene_index=still_data.get("scene_index", 0), shot_index=still_data.get("shot_index", 0), where="single", ) # 8. image index + var_t2i rewrite # D5 §4.3: attached_meta passthrough (label rewrite, meta 변경 0 — P1). # Area #11 v1 W2: ref_roles + ref_role_metadata 도 parallel passthrough (6-tuple). labeled_refs, _sid_to_img, _sid_info, ref_roles, ref_role_metadata, attached_meta = _build_image_index_helper( labeled_refs, entity_lookup, ref_roles=ref_roles, ref_role_metadata=ref_role_metadata, attached_meta=attached_meta, ) # W21B-W7 W-B (2026-06-12): printed_prop anchor keep_real_scale — 해당 # still 의 prop_ref 라벨에 scale_contract 1문장. build_image_index 가 prop # 라벨을 canon 기반으로 재작성하므로 반드시 그 **이후** 적용 (실측 2026-06-12: # 이전 위치에선 재작성에 지워짐). batch path 와 같은 helper (post-pass 금지). from app.core.steps.visual_continuity_anchor_step import ( apply_keep_real_scale_to_prop_labels, load_printed_prop_anchor_context, ) _vca_ctx = load_printed_prop_anchor_context(project_id, episode_id) if _vca_ctx: labeled_refs = apply_keep_real_scale_to_prop_labels( labeled_refs, ref_roles, ref_role_metadata, scene_index=still_data.get("scene_index"), shot_index=still_data.get("shot_index"), anchor_ctx=_vca_ctx, ) # W21B-W8 composition guide (option C, 2026-06-15): mode='sketch'(그룹 첫 샷) # → 마네킹 구도 스케치 ref append + same-room bg identity-only 완화. mode= # 'continuity_anchor'(이후 샷) → anchor_source 샷의 현재-run 완성 프레임 bytes 를 # 명시 참조(source_bytes_resolver)해 prev-shot ref 교체 (batch path 와 같은 helper). # 두 flag OFF/guide 부재 시 no-op byte-identical. _cg_ctx 는 위 force_char 단계서 로드. if _cg_ctx: from app.core.steps.outdoor_site_layout_step import ( attach_composition_guide_ref, ) attach_composition_guide_ref( labeled_refs, ref_roles, ref_role_metadata, attached_meta, scene_index=still_data.get("scene_index"), shot_index=still_data.get("shot_index"), guide_ctx=_cg_ctx, source_bytes_resolver=_make_osl_source_bytes_resolver( stills, scene_paths_by_index_by_id), framing=_framing_scale, ) # P8 I-1 (2026-06-28): registered (bg-aware) immobilized pose guide — image-time # 생성/부착(VCA step 21.66 엔 bg plate 부재, background_render=24.72). build_image_index # 이후 위치(라벨 재작성 안 받음, composition_guide 동일 패턴). flag OFF / 멤버 아님 / # plate 부재 → no-op byte-identical (white-bg fallback 없음). batch path 와 같은 helper. from app.core.steps.visual_continuity_anchor_step import ( attach_registered_pose_guide_ref, load_immobilized_subject_anchor_context, ) from app.services.registered_pose_guide_service import GUIDE_SUBDIR _imm_ctx = load_immobilized_subject_anchor_context(project_id, episode_id) if _imm_ctx: # Phase C: guide(gpt-edit) + underlay(PIL) 중간물을 capture scope 안에서 영속화. # flag OFF → _imm_ctx falsy 라 scope 미개방(byte-identical). _imm_ctx 가 있어도 # 이 샷이 멤버가 아니면 attach 가 내부에서 no-op → 빈 queue → flush 0(무해). from app.services.image_capture.context import generation_context with generation_context( project_id, episode_id, stage="registered_pose_guide", still_id=still_data.get("id"), scene_index=still_data.get("scene_index"), shot_index=still_data.get("shot_index"), ): attach_registered_pose_guide_ref( labeled_refs, ref_roles, ref_role_metadata, attached_meta, scene_index=still_data.get("scene_index"), shot_index=still_data.get("shot_index"), anchor_ctx=_imm_ctx, background_chain_bg_map=background_chain_bg_map, cache_dir=Path(settings.projects_dir) / project_id / "episodes" / episode_id / "images" / GUIDE_SUBDIR, ) # Wave5 (2026-06-30): 실내 shared-model pose 가이드 attach (worker lookup-only). # 단건 path 는 indoor_pose_ctx=None(미precompute) → no-op byte-identical. precompute # 는 batch generate_images main thread 1회만(단건 regen 은 v1 미지원, 문서화). from app.core.steps.indoor_shared_pose_guide_context import ( attach_indoor_pose_guide_ref, ) attach_indoor_pose_guide_ref( labeled_refs, ref_roles, ref_role_metadata, attached_meta, scene_index=still_data.get("scene_index"), shot_index=still_data.get("shot_index"), indoor_pose_ctx=indoor_pose_ctx, framing=_framing_scale, ) # A5 (2026-07-02): immobilized 그룹 후속 멤버 샷에 선행 environment 멤버의 # 완성 프레임을 연속성 ref 로 부착 (star-to-env, stale fallback 금지 — # composition_continuity 와 동일 resolver 계약). flag OFF / 비멤버 / anchor # 프레임 부재 → no-op(+진단은 still_data 로 persistence 전달). batch path 와 # 같은 helper. _imm_ctx 는 위 registered guide 단계서 로드. if _imm_ctx: from app.core.steps.visual_continuity_anchor_step import ( attach_immobilized_prev_frame_ref, ) _ipf_key_to_sid: Dict[Tuple[int, int], str] = {} for _ipf_s in (stills or []): _ipf_si, _ipf_shi = _ipf_s.get("scene_index"), _ipf_s.get("shot_index") _ipf_id = _ipf_s.get("id") if isinstance(_ipf_si, int) and isinstance(_ipf_shi, int) and _ipf_id: _ipf_key_to_sid[(_ipf_si, _ipf_shi)] = str(_ipf_id) _ipf_diag: Dict[str, Any] = {} attach_immobilized_prev_frame_ref( labeled_refs, ref_roles, ref_role_metadata, attached_meta, scene_index=still_data.get("scene_index"), shot_index=still_data.get("shot_index"), anchor_ctx=_imm_ctx, source_bytes_resolver=_make_osl_source_bytes_resolver( stills, scene_paths_by_index_by_id), source_still_id_by_key=_ipf_key_to_sid, diag_out=_ipf_diag, ) if _ipf_diag: still_data["_prev_frame_chain_diag"] = _ipf_diag var_t2i = _rewrite_t2i_helper(var_t2i, _sid_to_img, _sid_info) # GROUNDING-V2 D (2026-08-31) — `location_part` 고증 참조 묶음 # (맥락 + 상세)을 **덧붙인다**. # # ★★**RPC sidecar 가 있을 때만** 붙는다. 없으면 `attach_from_rpc` 가 # 0 을 돌려주고 네 평행 목록을 **한 글자도 안 건드린다** — 기존 CP 는 # 그래서 비회귀다(`test_bundle_end_to_end_inert` 의 등가 시험이 None · # {} · 글자 · 숫자 · 다른 키에서 목록이 `==` 로 같음을 잠근다). # # ★★이 함수는 요구를 **안 쓴다** — 만드는 쪽(`write_sidecar`)이 적은 것을 # **읽고 대조만** 한다. 쓰면 SOT 가 아니게 되고, 조립이 이 호출을 # 빠뜨리면 쓰기도 같이 빠져 **거짓 통과**한다(Codex 재현 2026-08-31). from app.modules.pipeline.grounding_reference_bundle import ( attach_from_rpc as _grounding_attach, ) def _grounding_load_bytes(coord: Dict[str, Any]): """좌표로 사진을 읽는다. ★**갈래를 갈라** 읽고 **읽은 자리**를 돌려준다. ★★★`asset_id or path` 로 모호하게 두면 **읽지도 않은 자산 UUID 가 계보로 남는다** (Codex 2026-08-31). 그리고 `ref_image_map` 의 열쇠는 **참조 map key** 지 ImageAsset UUID 가 **아니다** — 그것을 asset_id 라고 부르면 안 된다. Returns: `(bytes, resolved)`. `resolved` 는 **실제로 읽은 자리**다. """ from app.modules.pipeline.grounding_reference_bundle import ( SOURCE_ASSET, SOURCE_FILE, ) src = str((coord or {}).get("source") or "") if src == SOURCE_FILE: from pathlib import Path as _P p = _P(str(coord.get("path") or "")) if not p.exists(): return b"", {} return p.read_bytes(), {"source": SOURCE_FILE, "path": str(p)} if src == SOURCE_ASSET: # ★★아직 **배선 안 됐다**. 조용히 다른 데서 읽어 UUID 를 계보로 # 남기지 않는다 — D cutover 에서 project-scoped asset loader 를 # 잇는다. 그전까지 이 갈래를 쓰는 CP 는 **없다**. raise NotImplementedError( "`source=asset` loader 가 아직 안 이어졌다 — D cutover 에서 " "project-scoped ImageAsset loader 를 잇는다. 그때까지 " "sidecar 는 `source=file` 좌표만 쓴다") return b"", {} _gb_added = _grounding_attach( labeled_refs, attached_meta, ref_roles, ref_role_metadata, _rpc_f11, load_bytes=_grounding_load_bytes, where=f"scene{still.scene_index}_shot{still.shot_index}", ) if _gb_added: logger.info( "Scene %s Shot %s: grounding bundle refs attached (%d)", still.scene_index, still.shot_index, _gb_added, ) # 9. (v2) _build_final_scene_prompt — caller 가 cached_* 전달 # Area #11 v1 W2: payload arg (4 parallel list → LabeledRefPayload). _payload = make_labeled_ref_payload( labeled_refs=labeled_refs, ref_roles=ref_roles, ref_role_metadata=ref_role_metadata, attached_meta=attached_meta, ) # ★★조사가 고른 것이 **실제로 붙었나** — 기존 validator 의 클로즈업·plate # 면제와 **별개**다. 붙이는 쪽이 넘긴 목록이 아니라 **RPC 에 적힌 것**을 # 읽는다(Codex: `assert_from_rpc` caller 가 0 이었다). from app.modules.pipeline.grounding_reference_bundle import ( assert_from_rpc as _grounding_assert, ) _grounding_assert(_rpc_f11, attached_meta) try: _full_prompt = _build_final_scene_prompt( var_t2i, _payload, cached_style_context, project_config=project_config, entity_text_map=cached_entity_text_map, ) except Exception as exc: logger.warning("Prompt build failed: %s", exc) _full_prompt = var_t2i # P0 (2026-07-01, Codex NARROW_1): 기본은 기존 4-tuple(legacy caller/test 호환, # byte-identical). return_payload=True(단건 경로만)면 _payload(ref_role_metadata # 에 asset_id stamp 포함)를 5번째로 반환 → scene_result 에 실제 첨부 UUID lineage # 실어 영속화. if return_payload: return _full_prompt, labeled_refs, attached_meta, reference_phrase_kinds, _payload return _full_prompt, labeled_refs, attached_meta, reference_phrase_kinds class SceneGenerationCoordinator: """Scene 이미지 생성 전용 코디네이터 (W5 F22 Phase B.25). SceneImageService의 bound method들을 분리해 facade의 독립성을 확보. 상태는 facade와 동일한 의존성을 주입받음 (db/project/actor/logger + 5 svc). 모든 메서드는 facade에서 literal lift (동작 보존). """ def __init__( self, db: OrmSession, project_id: str, actor_id: str, activity_logger: ActivityLogger, persistence_svc: ScenePersistenceService, reference_svc: SceneReferenceService, validation_svc: SceneValidationService, variation_svc: SceneVariationService, provenance_svc: SceneProvenanceService, ) -> None: self._db = db self._project_id = project_id self._actor_id = actor_id self._logger = activity_logger self._persistence_svc = persistence_svc self._reference_svc = reference_svc self._validation_svc = validation_svc self._variation_svc = variation_svc self._provenance_svc = provenance_svc def _get_style_context(self, episode_id: str = None) -> str: """프로젝트 세계관을 T2I 프롬프트 앞에 붙일 컨텍스트로 변환 (HEAD 시그니처 보존).""" ps = self._db.query(ProjectSettings).filter( ProjectSettings.project_id == self._project_id).first() if ps and ps.style_rules_json: sr = json.loads(ps.style_rules_json) return ( f"Photorealistic cinematic still. " f"Setting: {sr.get('era','')}, {sr.get('region','')}. " f"Avoid: {sr.get('must_avoid','')}." ) if episode_id: cp = ( Path(settings.projects_dir) / self._project_id / "checkpoints" / "episodes" / episode_id / "visual_world_rules" / "manifest.json" ) if cp.exists(): vwr = json.loads(cp.read_text(encoding="utf-8")).get("data", {}) era = vwr.get("era", "") region = vwr.get("region", "") if era or region: return f"Photorealistic cinematic still. Setting: {era}, {region}." return "Photorealistic cinematic still." def _init_scene_gen_clients( self, ) -> "tuple[GeminiImageClient, OpenAIClient, PromptSanitizer, GenerationTracker]": """Scene 생성에 공통으로 필요한 4개 클라이언트 초기화. W5 F22 Phase B.23.2 (2026-04-23): generate_images + generate_single_scene_image 양쪽에서 동일하게 쓰이던 4-client 초기화 블록을 공통화. Returns: (gemini_client, openai_client, sanitizer, tracker) PromptSanitizer(openai_client)는 API 호환성을 위해 인자를 받지만 현재 내부적으로 사용하지 않음 (prompt_sanitizer.py 확인). 순서만 보존. """ gemini_client = GeminiImageClient(model=settings.gemini_image_model) openai_client = OpenAIClient() sanitizer = PromptSanitizer(openai_client) tracker = GenerationTracker(self._db, self._project_id) return gemini_client, openai_client, sanitizer, tracker def _generate_scene_in_loop( self, *, si: int, stills: List[Dict[str, Any]], entity_lookup: Dict[str, Dict[str, Any]], scene_paths_by_index_by_id: Dict[str, Path], location_scene_history: Dict[str, tuple], staging_map: Dict[str, Any], scene_primary_asset_id_by_still_id: Optional[Dict[str, str]] = None, scene_ref_image_map: Dict[str, bytes], scene_ref_asset_id_map: Optional[Dict[str, str]] = None, background_chain_bg_map: Dict[str, Any], dep_detail_map: Dict[str, Any], gemini_client: Any, sanitizer: Any, validator: Optional[ImageValidator], scene_dir: Path, cached_style_context: str, cached_entity_text_map: Dict[str, str], world_guide: Dict[str, Any], episode_id: str, zoom_ctx: Optional[Dict[str, Any]] = None, indoor_pose_ctx: Optional[Dict[str, Any]] = None, outdoor_direct_ctx: Optional[Dict[str, Any]] = None, ) -> tuple: """Generate N variation images for one still (runs in thread pool). W5 F22 Phase B.22.4 (2026-04-22): generate_images nested function _generate_one_scene을 bound method로 lift. closure 16개 변수를 explicit keyword-only param으로 승격. Returns (scene_index, list_of_results, visible_entities, current_location_ids). Each result dict (from _generate_variation_in_loop) or None. """ still_data = stills[si] try: visible_ids = json.loads(still_data["visible_entities_json"]) except json.JSONDecodeError: visible_ids = [] visible_entities = [] for v in visible_ids: if isinstance(v, dict): eid = v.get("id") or v.get("entity_id", "") if eid and eid in entity_lookup: visible_entities.append(entity_lookup[eid]) elif isinstance(v, str): if v in entity_lookup: visible_entities.append(entity_lookup[v]) # 시각적 연관 씬 참조 이미지 + bytes_source_kind (provenance annotator, spec §6). # spec docs/superpowers/specs/2026-05-15-zoom-in-detail-source-provenance-design.md. best_prev_bytes = None bytes_source_kind: Literal["dep_scene", "location_history", "none"] = "none" current_location_ids = [ e["id"] for e in visible_entities if e.get("entity_type") == "location" ] # B (2026-07-02): outdoor prev-frame 의무첨부 컨텍스트 — 단건 path 와 동일. # flag OFF → {} no-op. worker thread 안전(파일 read + dict 조인만). from app.core.steps.outdoor_site_layout_step import ( load_outdoor_prev_frame_context, ) from app.modules.pipeline.outdoor_site_layout_plan import ( build_outdoor_prev_frame_diag, primary_location_uuid_by_scene, resolve_outdoor_history_fallback_uuid, ) _opf_ctx = load_outdoor_prev_frame_context(self._project_id, episode_id) _opf_uuid_by_scene = primary_location_uuid_by_scene( _opf_ctx.get("primary_location_by_scene") or {}, _opf_ctx.get("outdoor_loc_sids") or set(), entity_lookup, ) if _opf_ctx else {} _opf_hist_still: Optional[Dict[str, Any]] = None # W21B-W7 W-C2b (2026-06-12): zoom 멤버 still — 독립 T2I 대신 source_wide # primary 의 crop/i2i-fill (단건 path 와 같은 helper). source 는 dep graph # 의 zoom→source edge 로 이전 batch 에서 이미 영속화됨 (batch 경계 = # barrier — race guard). 실패는 비차단 fallback(기존 variation 루프) + # artifacts 의 fallback reason (W-C2b acceptance 에선 실패 기록). _zc_key = (still_data.get("scene_index"), still_data.get("shot_index")) _zc_target = ((zoom_ctx or {}).get("zoom_targets") or {}).get(_zc_key) if _zc_target: from app.core.steps.visual_continuity_anchor_step import ( build_zoom_fill_prop_refs, ) from app.services.zoom_continuity_render_service import ( ZoomContinuityRenderError, generate_continuity_crop_png, ) # S12sh12 fix (2026-06-12): printed_prop anchor 계약을 fill 의 object # ref 라벨에 적용 (단건 path 와 같은 helper — 파일 read 만, worker 안전). # content-SOT 절은 anchor 계약 ref 가 실제 있을 때만 (Codex narrow). _zc_prop_refs, _zc_prop_anchor_applied = build_zoom_fill_prop_refs( self._project_id, episode_id, visible_entities, scene_ref_image_map, scene_index=still_data.get("scene_index"), shot_index=still_data.get("shot_index"), ) # ★worker thread 에선 db 세션 사용 금지 (S23sh4 실측: SQLAlchemy # concurrent operations 충돌) — source bytes 는 # scene_paths_by_index_by_id (worker 읽기 전용) 에서 직접 읽는다. _zc_src_key = tuple(_zc_target.get("source") or ()) _zc_src_id = next( (s_.get("id") for s_ in stills if (s_.get("scene_index"), s_.get("shot_index")) == _zc_src_key), None, ) _zc_src_bytes = None _zc_src_path = scene_paths_by_index_by_id.get(_zc_src_id) if _zc_src_id else None if _zc_src_path is not None and Path(_zc_src_path).exists(): _zc_src_bytes = Path(_zc_src_path).read_bytes() # Phase C: source 의 primary asset UUID 를 메인스레드 사전 맵에서 조회 # (worker DB 금지). 직전 batch 의 _finalize_and_track_primary 가 채움 — # 직전 run resume 으로 done 인 source 는 부재 → None(부분 lineage 허용). _zc_src_asset_id = ( (scene_primary_asset_id_by_still_id or {}).get(_zc_src_id) if _zc_src_id else None ) try: if _zc_src_bytes is None: raise ZoomContinuityRenderError( f"source_unavailable: S{_zc_src_key} primary path not in " "scene_paths map (dep edge 가 source 를 선행 batch 로 " "보장하므로 정상 경로에선 발생하지 않아야 함)") _zc_png, _zc_fill_prompt, _zc_diag = generate_continuity_crop_png( project_id=self._project_id, episode_id=episode_id, scene_index=still_data.get("scene_index"), shot_index=still_data.get("shot_index"), zoom_target=_zc_target, gemini_client=gemini_client, prop_refs=_zc_prop_refs, prop_refs_have_anchor_contract=_zc_prop_anchor_applied, source_bytes=_zc_src_bytes, # P0c (2026-06-18): batch scene_image_gen 로그에 still_id 포함 # (operation_type 은 함수 기본값 scene_image_gen). trace_meta={"still_id": still_data.get("id")}, # Phase C: zoom crop 중간물 영속화 — still_id + source UUID lineage. still_id=still_data.get("id"), source_asset_id=_zc_src_asset_id, ) _zc_out = scene_dir / f"{_new_id()}.png" _zc_out.parent.mkdir(parents=True, exist_ok=True) _zc_out.write_bytes(_zc_png) _zc_result = { "id": _new_id(), "asset_type": "scene", "entity_id": None, "still_id": still_data.get("id"), "episode_id": episode_id, "file_path": to_relative_image_path(str(_zc_out)), "prompt_used": _zc_fill_prompt, "generation_model": settings.gemini_image_model, "width": None, "height": None, "status": "generated", "review_notes": json.dumps( {"zoom_continuity": _zc_diag}, ensure_ascii=False), "variant_type": "zoom_continuity_crop", "theme_label": "zoom continuity crop", "created_at": _now(), } logger.info( "zoom_continuity(batch): S%ssh%s crop/i2i-fill 성공 (group=%s)", still_data.get("scene_index"), still_data.get("shot_index"), _zc_target.get("group_id"), ) return (si, [_zc_result], visible_entities, current_location_ids) except ZoomContinuityRenderError as _zc_exc: logger.warning( "zoom_continuity(batch): S%ssh%s crop fallback → 기존 T2I (%s)", still_data.get("scene_index"), still_data.get("shot_index"), _zc_exc, ) # 1순위: dependent_scene_id (씬 분석에서 결정된 시각적 연관 씬) dep_scene_id = still_data.get("dependent_scene_id") # W3 (2026-06-11): dep_detail 명시 타깃 still 우선 — single path 와 동일 # (S29 sh11 실측: bytes=sh4 ↔ 지시=sh7 불일치 → 프레임 복제). 상세 주석은 # resolve_dep_target_still_id docstring. _w3_dep_key = ( f"{still_data.get('scene_index', 0)}_{still_data.get('shot_index', 0)}" ) _w3_dep_info = (dep_detail_map or {}).get(_w3_dep_key) or {} _w3_target_id = resolve_dep_target_still_id( stills, _w3_dep_info.get("dep_scene_index"), _w3_dep_info.get("dep_shot_index"), ) if _w3_target_id and _w3_target_id in scene_paths_by_index_by_id \ and scene_paths_by_index_by_id[_w3_target_id].exists(): if dep_scene_id and dep_scene_id != _w3_target_id: logger.info( "Scene %s Shot %s: dep bytes target realigned to dep_detail " "(%s → %s) — W3 source coherence", still_data.get("scene_index"), still_data.get("shot_index"), dep_scene_id, _w3_target_id, ) dep_scene_id = _w3_target_id if dep_scene_id and dep_scene_id in scene_paths_by_index_by_id: dep_path = scene_paths_by_index_by_id[dep_scene_id] if dep_path.exists(): best_prev_bytes = dep_path.read_bytes() bytes_source_kind = "dep_scene" # 2순위: 같은 장소 이전 씬 (location_scene_history) # W20F10 O policy: build_scene_attached_refs 와 동일 — ref_usage="zoom_in_detail" # 시 location_history silent fallback 금지. dep_detail_map 의 scene_index/ # shot_index 키 lookup (문자열 의미 추론 X). _dep_key_for_policy = ( f"{still_data.get('scene_index', 0)}_{still_data.get('shot_index', 0)}" ) _ref_usage_for_policy = ( (dep_detail_map or {}).get(_dep_key_for_policy, {}) or {} ).get("ref_usage", "") if not best_prev_bytes and _ref_usage_for_policy != "zoom_in_detail": for loc_id in current_location_ids: if loc_id in location_scene_history: best_prev_bytes, _opf_hist_still = location_scene_history[loc_id] bytes_source_kind = "location_history" break # B (2026-07-02): VE location 부재 outdoor 샷 — primary_location 구조 # fallback (단건 path 와 동일, Layer 2.6 정합 위해 loc UUID 보강). if not best_prev_bytes and _opf_uuid_by_scene: _opf_fb = resolve_outdoor_history_fallback_uuid( still_data.get("scene_index"), primary_loc_uuid_by_scene=_opf_uuid_by_scene, location_scene_history=location_scene_history, ) if _opf_fb: best_prev_bytes, _opf_hist_still = location_scene_history[_opf_fb] bytes_source_kind = "location_history" if _opf_fb not in current_location_ids: current_location_ids.append(_opf_fb) logger.info( "outdoor_prev_frame(batch): S%ssh%s history fallback — " "primary_location UUID 로 prev 프레임 확보", still_data.get("scene_index"), still_data.get("shot_index")) elif not best_prev_bytes and _ref_usage_for_policy == "zoom_in_detail": logger.warning( "scene_generation_coordinator: %s zoom_in_detail dep_scene bytes " "missing — location_history fallback skipped (W20F10 O policy)", _dep_key_for_policy, ) # v5: N개 변형 로드 — t2i_variations가 있으면 사용, 없으면 기본 1개 variations = still_data.get("t2i_variations", []) if not variations: # fallback: 기본 프롬프트 1개 base_t2i = still_data.get("t2i_prompt_cinematic") or still_data.get("still_frame_prompt", "") variations = [{"theme": "base", "theme_label": "base", "t2i_prompt": base_t2i}] # W21B-W7 W-C2b: image-phase prompt override (단건 path 와 동일 계약 — # scene_detail cp 불변, 실사용 프롬프트는 prompt_used 로, provenance 는 # anchor cp/로그). variation index 매칭. W21B-W8: source_overrides 는 # caller 가 zoom/site 를 merge 한 맵일 수 있다 (custom > zoom > site > # original) — entry 의 prompt_source 가 출처. _zc_override = ((zoom_ctx or {}).get("source_overrides") or {}).get(_zc_key) if _zc_override: variations = [ ({**var, "t2i_prompt": _zc_override["revised"][vi]} if vi in _zc_override["revised"] else var) for vi, var in enumerate(variations) ] logger.info( "scene prompt override 적용(batch): S%ssh%s " "(prompt_source=%s, group=%s)", still_data.get("scene_index"), still_data.get("shot_index"), _zc_override.get("prompt_source", "zoom_continuity_anchor"), _zc_override.get("group_id"), ) # Fix A (2026-05-10): target_variations[*].t2i_prompt union 으로 ref build — # 옛 first_t2i 만 보던 결함이 24 shot deterministic ref-contract 실패의 # 직접 root cause. 한 variation 의 outlook 만 ref attach → 다른 variation # 의 outlook 누락 → validator RefContractError. # 실제 executor submit 될 variation set (`valid_variations[-max_t2i:]`) # 만 union — 버려지는 variation 은 attach 안 함 (over-attach 회피). _max_t2i = settings.scene_variation_count _target_raw = [v for v in variations if v.get("t2i_prompt")][-_max_t2i:] _target_raw_prompts = [v.get("t2i_prompt", "") for v in _target_raw] # W22 직행 (Codex W4a NARROW_1): _od 는 resolver 입력 산출 **이전**에 # 계산 — 직행 샷은 실제 생성 프롬프트(9절)가 resolver/attach/validator # 의 공통 SOT 여야 단건 path 와 정합. zoom crop 조기 return 은 위에서 # 이미 처리(별도 생성 경로) — zoom/site override 와 직행이 동시 존재해도 # resolver·생성 SOT 는 직행 9절. _od_key = f"{still_data.get('scene_index', 0)}_{still_data.get('shot_index', 0)}" _od = (outdoor_direct_ctx or {}).get(_od_key) _od_prompt = (_od or {}).get("prompt") or "" _target_raw_prompts = _direct_resolver_prompts(_od, _target_raw_prompts) # state_variant: staging subject_state immobilized (dead/unconscious/severely_injured) # 인물 감지 → variant ref 제공. Area #2 W5 (2026-05-17): legacy mixed gaze field literal 폐기 → # is_immobilized_state SOT 단일 path (detail은 detect_state_variant_sids). # W5 F22 Phase B.22.1: scene_reference_service.detect_state_variant_sids로 이관 _staging_key = f"{still_data.get('scene_index', 0)}_{still_data.get('shot_index', 0)}" _stg = staging_map.get(_staging_key) if staging_map else None _state_variant_sids = self._reference_svc.detect_state_variant_sids( visible_entities, entity_lookup, scene_ref_image_map, _stg, ) if _state_variant_sids: logger.info("Scene %d Shot %d: state_variant=%s", still_data.get("scene_index", 0), still_data.get("shot_index", 0), list(_state_variant_sids.keys())) # D5 T4 §4.6 / FINDING 9 Cat3: 씬 단위 RPC 1회 build — variation 마다 # 재계산 회피. RPC 는 caller-injected (still_data) 우선, 없으면 manifest # lookup. lookup 실패 시 RefContractError propagate (silent skip 금지, # G1 정합) — 이 변형들은 모두 차단 (씬 단위 fail-fast). resolver·validator # 가 동일 still_data["render_prompt_card"] 를 소비하도록 resolve 호출 이전 배치. _scene_index = still_data.get("scene_index") _shot_index = still_data.get("shot_index") or still_data.get("_shot_index") rpc = still_data.get("render_prompt_card") if rpc is None: rpc = lookup_render_prompt_card( project_id=self._project_id, episode_id=episode_id, scene_index=_scene_index, shot_index=_shot_index, ) still_data["render_prompt_card"] = rpc # W1-A (2026-06-11): space_set_bg no-plate 진단 shot 의 required # background runtime reconciliation (사본, persisted RPC 불변). rpc = waive_required_background_if_no_plate( rpc, background_chain_bg_map, still_data, ) still_data["render_prompt_card"] = rpc # FINDING 11 (e2e-bughunt-v1): consume-side — scene_detail sidecar # reference_phrase_kinds 의 'character' over-declaration 을 required_refs # SOT 와 정합. 기존 checkpoint 도 image-gen 직전 정규화 (scene_detail # 재실행 불요). W4 'prop' normalization 의 'character' 변종. from app.core.steps.detail_steps import ( _strip_overdeclared_character_phrase_kind, ) _strip_overdeclared_character_phrase_kind(variations, rpc) # D5 T4 §4.6: resolver return — (labeled_refs, attached_meta) 평행 list. # T2 의 임시 discard 폐기 — 배치 path 도 단건과 동등 contract enforcement. # Fix A (2026-05-10): resolve_refs_for_prompt_set 호출 — 모든 target # variation prompt 의 ID union 으로 ref build. # Patch A — RPC.asset_requirements.required_refs 추출 후 resolver 에 전달. _patch_a_required_refs_batch = ( ((still_data.get("render_prompt_card") or {}).get("asset_requirements") or {}) .get("required_refs") or [] ) # W21B-W8 option C (2026-06-15): composition continuity guide context 1회 # 로드 — force_character_names(step precompute: this-shot ∩ anchor_source # staged 교집합)를 resolver 에 thread + 아래 attach 에 재사용. 두 flag OFF/ # guide 부재 시 빈 set = byte-identical. from app.core.steps.outdoor_site_layout_step import ( load_composition_guide_context, ) _cg_ctx = load_composition_guide_context(self._project_id, episode_id) _cg_entry = (_cg_ctx or {}).get( (still_data.get("scene_index"), still_data.get("shot_index"))) or {} _force_char_names = set(_cg_entry.get("forced_character_names") or []) # Area #11 v1 W2: 2-tuple → LabeledRefPayload return (batch path). _batch_payload = self._reference_svc.resolve_refs_for_prompt_set( t2i_prompts=_target_raw_prompts, visible_entities=visible_entities, scene_ref_image_map=scene_ref_image_map, entity_lookup=entity_lookup, state_variant_sids=_state_variant_sids, required_refs=_patch_a_required_refs_batch, # option C: guide 샷(force 존재)에만 staging 전달해 backstop 활성화 — # 비-guide 샷은 None 으로 휴면(byte-identical). 단건 path 와 동일. staging=_stg if _force_char_names else None, force_character_names=_force_char_names or None, scene_ref_asset_id_map=scene_ref_asset_id_map, ) labeled_refs = list(_batch_payload.labeled_refs) attached_meta = list(_batch_payload.attached_meta) ref_roles = list(_batch_payload.ref_roles) ref_role_metadata = list(_batch_payload.ref_role_metadata) # B-run S15 실측 fix (2026-06-12): back_to_camera stamp — single path 동일. apply_back_to_camera_constraints( attached_meta, ref_roles, ref_role_metadata, staging=_stg, visible_entities=visible_entities, entity_lookup=entity_lookup, ) # ── 배경 참조 fallback chain (labeled_refs 첫 번째에 주입) ── # 우선순위: # 1) background_chain_render 결과 PNG (있으면 + framing_scale 이 # close 가 아닐 때만 — Phase 9.1) # 2) 이전 같은-위치 샷 이미지 (chain 없을 때 fallback — 야외/저빈도 location) # 둘 다 없으면 entity ref만 사용. 같은 슬롯(첫 ref)을 둘 중 하나만 채운다. # PR #5(2026-04-27): set_design 자리를 background_chain으로 교체. # 본 fix(2026-04-27 추가): chain skipped location은 prev_shot_ref로 fallback. # Phase 9.1: shot framing_scale 이 close 이면 chain_bg ref 를 의도적으로 # skip — wide bg PNG + close 인물 합성 scale 충돌 방지. _bc_key = f"{still_data.get('scene_index', 0)}_{still_data.get('shot_index', 0)}" _bc_bg = background_chain_bg_map.get(_bc_key) # W22 직행: _od 는 위(resolver 입력 산출 전)에서 계산됨 (NARROW_1). _framing_scale = get_framing_scale_or_raise( _stg, where="scene_generation_coordinator.generate_scene_in_loop", ) _is_close_framing = _framing_scale == FRAMING_CLOSE # W2 (2026-06-11, Codex 합의): dep_scene continuity > bg map — single path # 와 동일 (S12 sh7↔sh12 dynamic state 불일치 실측). 주석 상세는 single path. # W2-B 정련: dynamic-state usage(zoom_in_detail/exact_background)에만 적용. _w2_ref_usage = ( (dep_detail_map or {}).get(_bc_key, {}) or {} ).get("ref_usage", "") _has_dep_continuity = ( bytes_source_kind == "dep_scene" and _w2_ref_usage in ("zoom_in_detail", "exact_background") ) # W22 직행 consistency fix (2026-07-10): 단건 path 와 동일 — 직행도 # else 블록의 prev_shot 연속성 기계를 타고, 캐논 2장은 분기 뒤 최상단 # insert (최종 순서=[실사, 맵, prev, ...]). if (not _od) and ( _bc_bg and _bc_bg.get("image_bytes") and not _is_close_framing and not _has_dep_continuity ): # Area #11 v1 W2 (Codex iter 2 Important 3 fix): fail-fast strict on missing # bg_id (No Silent Fallback gate). payload length-parity invariant 일관. _bg_id = _bc_bg.get("bg_id") if not _bg_id: raise RefRoleError( "chain_bg missing bg_id (batch path) — No Silent Fallback gate (Area #11 v1)" ) labeled_refs.insert(0, (_bc_bg["label"], _bc_bg["image_bytes"])) # D5 T4 §4.6: bg_id 는 T1 의 bg_map entry 확장에서 직접 read (label 파싱 X). attached_meta.insert(0, ("background", _bg_id)) # Area #11 v1 W2: parallel sidecar insert (4 list length parity 강제) ref_roles.insert(0, "background_chain_ref") # P0 (2026-07-01, Codex BLOCKING fix): bg_id(L04B06)는 UUID 아님 → # chain_bg ImageAsset UUID 를 구조키로 resolve 해 asset_id 로 stamp. # 미해결이면 asset_id 생략 → unresolved(input_image_ids 미포함). _bg_meta = {"bg_id": _bg_id, "pipeline_role": "background_render"} _bg_asset_uuid = self._reference_svc.resolve_chain_bg_asset_id(episode_id, _bg_id) if _bg_asset_uuid: _bg_meta["asset_id"] = _bg_asset_uuid ref_role_metadata.insert(0, _bg_meta) logger.info( "Scene %d Shot %d: background_chain ref injected", still_data.get("scene_index", 0), still_data.get("shot_index", 0), ) else: if _bc_bg and _bc_bg.get("image_bytes") and _has_dep_continuity and not _is_close_framing: # W2: 의도적 skip — 빈 establishing plate 가 dep_scene 의 dynamic # continuity(자세·핏자국·직전 동작)를 대체하지 않게. 이후 # prev_shot_ref(dep_scene bytes) 분기로 떨어진다. logger.info( "Scene %d Shot %d: bg map ref SKIPPED — dep_scene continuity " "priority (W2)", still_data.get("scene_index", 0), still_data.get("shot_index", 0), ) # W2-A: validate 직전 waiver 로 소비 (single path 와 동일). still_data["_dep_continuity_bg_waiver"] = True elif _bc_bg and _bc_bg.get("image_bytes") and _is_close_framing: # Phase 9.1: 의도적 skip — wide bg와 close 인물 scale mismatch 방지. # 이후 prev_shot_ref fallback 분기로 떨어진다. logger.info( "Scene %d Shot %d: chain_bg ref SKIPPED — close framing " "detected (framing_scale=%r)", still_data.get("scene_index", 0), still_data.get("shot_index", 0), _framing_scale, ) # prev_shot_ref 호출 실패는 still 자체를 실패시키지 않는다 — 빈 chain 슬롯으로 # 진행하면 entity refs만 사용한 정상 fallback으로 떨어진다 (Codex review C3). try: _prev_shot_ref = self._reference_svc.build_prev_shot_background_ref( best_prev_bytes=best_prev_bytes, bytes_source_kind=bytes_source_kind, still_data=still_data, visible_entities=visible_entities, current_location_ids=current_location_ids, dep_scene_id=dep_scene_id, stills=stills, location_scene_history=location_scene_history, dep_detail_map=dep_detail_map, staging=_stg, state_variant_sids=_state_variant_sids, entity_lookup=entity_lookup, ) except RefContractError: raise except TypeError: # spec §6 — signature 위반 fail-fast (No Silent Fallback Gate) raise except Exception as exc: logger.warning( "Scene %d Shot %d: prev_shot_ref build failed (%s) — falling back to entity-only refs", still_data.get("scene_index", 0), still_data.get("shot_index", 0), exc, ) _prev_shot_ref = None # B (2026-07-02): prev 프레임 source still 구조키 (단건 path 와 동일). # ★composition continuity_anchor(option C) 샷 제외 — bytes 교체 lane 소유. _b_src_sid = ( dep_scene_id if bytes_source_kind == "dep_scene" else ((_opf_hist_still or {}).get("id") if bytes_source_kind == "location_history" else None) ) _b_cg_anchor = ( ((_cg_ctx or {}).get( (still_data.get("scene_index"), still_data.get("shot_index"))) or {}).get("mode") == "continuity_anchor" ) if _prev_shot_ref: # Area #11 v1 W2: helper 가 5-tuple — ref_role + ref_role_metadata 동시 unpack. _label, _bytes, _loc_id, _role, _metadata = _prev_shot_ref # B: flag ON + source still 확정 시 lineage 구조키 stamp (persistence # post-hoc resolve → input_image_ids 엣지). OFF = 불변. if _opf_ctx and _b_src_sid and not _b_cg_anchor: _metadata = {**_metadata, "pipeline_role": "scene_prev_frame", "source_still_id": str(_b_src_sid)} if _od: # W22 직행 place-continuity (Codex BLOCKING_1) — 단건 path 와 동일. _metadata = {**_metadata, "outdoor_direct_place_continuity": True} labeled_refs.insert(0, (_label, _bytes)) # length parity 강제 (4 parallel list 모두 same length insert, loc_id None 도 빈 string) attached_meta.insert(0, ("background_prev_shot", _loc_id or "")) # Area #11 v1 W2: parallel sidecar insert ref_roles.insert(0, _role) ref_role_metadata.insert(0, _metadata) logger.info( "Scene %d Shot %d: prev_shot_ref injected (no chain bg for this shot, role=%s)", still_data.get("scene_index", 0), still_data.get("shot_index", 0), _role, ) # B (2026-07-02): outdoor prev-frame 의무첨부 진단 (단건 path 와 동일 — # 대상 아니면 None = 불변, 미해결 시 WARNING + metadata 영속). # option C continuity_anchor 샷은 그 lane 이 prev 프레임 소유 — 제외. if _opf_ctx and not _b_cg_anchor: _b_ve_loc_sids = { _sid for _lid in current_location_ids if isinstance( (_sid := (entity_lookup.get(_lid) or {}).get("short_id")), str) and _sid } _b_dep_key = ( f"{still_data.get('scene_index', 0)}_{still_data.get('shot_index', 0)}" ) _b_si, _b_shi = still_data.get("scene_index"), still_data.get("shot_index") _b_diag = build_outdoor_prev_frame_diag( scene_index=_b_si, shot_index=_b_shi, outdoor_loc_sids=_opf_ctx.get("outdoor_loc_sids") or set(), primary_location_by_scene=( _opf_ctx.get("primary_location_by_scene") or {}), ve_location_sids=_b_ve_loc_sids, chain_bg_attached=False, ref_usage=_w2_ref_usage, has_dep=bool(dep_scene_id or (dep_detail_map or {}).get(_b_dep_key)), has_same_scene_prior=any( s_.get("scene_index") == _b_si and isinstance((_p_shi := s_.get("shot_index")), int) and isinstance(_b_shi, int) and _p_shi < _b_shi for s_ in (stills or []) ), resolved=bool(_prev_shot_ref), bytes_source_kind=bytes_source_kind, source_still_id=str(_b_src_sid) if _b_src_sid else None, close_framing=_is_close_framing, ) if _b_diag: still_data["_prev_frame_chain_diag"] = _b_diag if not _b_diag["resolved"]: logger.warning( "outdoor_prev_frame(batch): S%ssh%s previous_frame_" "required_missing — outdoor+plate無+연속성 신호 샷인데 " "prev 프레임 미해결 (dep=%s, same_scene_prior=%s)", _b_si, _b_shi, _b_diag["required_signals"]["dep"], _b_diag["required_signals"]["same_scene_prior"]) if _od: # W22 직행: 캐논 2장 최상단 삽입 — prev 연속성 ref 뒤에 실행되므로 # 최종 순서=[실사, 맵, prev, ...] (단건 path 와 대칭). _insert_outdoor_canon_refs( labeled_refs, attached_meta, ref_roles, ref_role_metadata, _od, scene_index=still_data.get("scene_index", 0), shot_index=still_data.get("shot_index", 0), where="batch", ) # ── Image N 인덱싱: labeled_refs에 번호 부여 + T2I의 C##O##를 "the character from Image N"으로 치환 ── # D5 T4 §4.3: helper 가 attached_meta 평행 통과 (label 만 rewrite, meta 는 변형 X — P1 강화). # Area #11 v1 W2: 6-tuple — ref_roles + ref_role_metadata parallel passthrough. labeled_refs, _sid_to_img, _sid_info, ref_roles, ref_role_metadata, attached_meta = _build_image_index_helper( labeled_refs, entity_lookup, ref_roles=ref_roles, ref_role_metadata=ref_role_metadata, attached_meta=attached_meta, ) # W21B-W7 W-B (2026-06-12): printed_prop anchor keep_real_scale — # build_image_index 의 prop 라벨 재작성 이후 적용 (단건 path 와 같은 # helper, post-pass 금지). from app.core.steps.visual_continuity_anchor_step import ( apply_keep_real_scale_to_prop_labels, load_printed_prop_anchor_context, ) _vca_ctx = load_printed_prop_anchor_context(self._project_id, episode_id) if _vca_ctx: labeled_refs = apply_keep_real_scale_to_prop_labels( labeled_refs, ref_roles, ref_role_metadata, scene_index=still_data.get("scene_index"), shot_index=still_data.get("shot_index"), anchor_ctx=_vca_ctx, ) # W21B-W8 composition guide (option C, 2026-06-15): 단건 path 와 같은 helper — # mode='sketch'→마네킹 구도 스케치 ref + same-room bg 완화, mode='continuity_anchor' # →anchor_source 의 현재-run 완성 프레임 bytes 명시 참조(source_bytes_resolver). # _cg_ctx 는 위 force_char 단계서 로드. 두 flag OFF/guide 부재 시 no-op. if _cg_ctx: from app.core.steps.outdoor_site_layout_step import ( attach_composition_guide_ref, ) attach_composition_guide_ref( labeled_refs, ref_roles, ref_role_metadata, attached_meta, scene_index=still_data.get("scene_index"), shot_index=still_data.get("shot_index"), guide_ctx=_cg_ctx, source_bytes_resolver=_make_osl_source_bytes_resolver( stills, scene_paths_by_index_by_id), framing=_framing_scale, ) # P8 I-1 (2026-06-28): registered (bg-aware) immobilized pose guide — 단건 # path 와 같은 helper. image-time 생성/부착(white-bg fallback 없음). flag OFF / # 멤버 아님 / plate 부재 → no-op byte-identical. from app.core.steps.visual_continuity_anchor_step import ( attach_registered_pose_guide_ref, load_immobilized_subject_anchor_context, ) from app.services.registered_pose_guide_service import GUIDE_SUBDIR _imm_ctx = load_immobilized_subject_anchor_context(self._project_id, episode_id) if _imm_ctx: # Phase C: batch worker 본문 내 scope — 같은 thread set/reset(bind_context # 불필요), flush 는 독립 SessionLocal(thread-safe, Codex #6). flag OFF → # scope 미개방(byte-identical). _imm_ctx 있어도 비멤버 샷은 attach no-op → # 빈 queue → flush 0(무해). from app.services.image_capture.context import generation_context with generation_context( self._project_id, episode_id, stage="registered_pose_guide", still_id=still_data.get("id"), scene_index=still_data.get("scene_index"), shot_index=still_data.get("shot_index"), ): attach_registered_pose_guide_ref( labeled_refs, ref_roles, ref_role_metadata, attached_meta, scene_index=still_data.get("scene_index"), shot_index=still_data.get("shot_index"), anchor_ctx=_imm_ctx, background_chain_bg_map=background_chain_bg_map, cache_dir=Path(settings.projects_dir) / self._project_id / "episodes" / episode_id / "images" / GUIDE_SUBDIR, ) # Wave5 (2026-06-30): 실내 shared-model pose 가이드 attach (worker lookup-only). # precompute(main thread)에서 생성+capture 끝난 guide bytes 를 ctx 에서 lookup 만. # flag OFF / 비멤버 샷 / 미admit → no-op byte-identical. DB/VLM/생성 0. from app.core.steps.indoor_shared_pose_guide_context import ( attach_indoor_pose_guide_ref, ) attach_indoor_pose_guide_ref( labeled_refs, ref_roles, ref_role_metadata, attached_meta, scene_index=still_data.get("scene_index"), shot_index=still_data.get("shot_index"), indoor_pose_ctx=indoor_pose_ctx, framing=_framing_scale, ) # A5 (2026-07-02): immobilized prev 완성프레임 chaining — 단건 path 와 같은 # helper. dep edge(generate_images)가 env anchor 선생성을 보장, resolver 는 # 현재-run path map 만(stale fallback 금지). flag OFF / 비멤버 / anchor 부재 # → no-op(+진단 still_data 전달, var_result 로 영속). if _imm_ctx: from app.core.steps.visual_continuity_anchor_step import ( attach_immobilized_prev_frame_ref, ) _ipf_key_to_sid: Dict[Tuple[int, int], str] = {} for _ipf_s in (stills or []): _ipf_si, _ipf_shi = _ipf_s.get("scene_index"), _ipf_s.get("shot_index") _ipf_id = _ipf_s.get("id") if isinstance(_ipf_si, int) and isinstance(_ipf_shi, int) and _ipf_id: _ipf_key_to_sid[(_ipf_si, _ipf_shi)] = str(_ipf_id) _ipf_diag: Dict[str, Any] = {} attach_immobilized_prev_frame_ref( labeled_refs, ref_roles, ref_role_metadata, attached_meta, scene_index=still_data.get("scene_index"), shot_index=still_data.get("shot_index"), anchor_ctx=_imm_ctx, source_bytes_resolver=_make_osl_source_bytes_resolver( stills, scene_paths_by_index_by_id), source_still_id_by_key=_ipf_key_to_sid, diag_out=_ipf_diag, ) if _ipf_diag: still_data["_prev_frame_chain_diag"] = _ipf_diag # 항상 마지막 N-1개 프롬프트로 T2I 생성 + 앵글 1개 = N개 # scene_variation_count = T2I 개수, fal.ai 앵글은 항상 +1 max_t2i = settings.scene_variation_count # 기본 2 → T2I 2개 + 앵글 1개 = 3개 valid_variations = [] # _od_prompt 는 위(resolver 입력 산출 전)에서 계산됨 (NARROW_1) for vi, var in enumerate(variations): var_theme = var.get("variant_label", var.get("theme", f"var_{vi}")) var_theme_label = var.get("camera_effect", var.get("theme_label", var_theme)) if _od_prompt: # W22 직행: 전 variation 동일 9절 (FREE_CAMERA 재량이 자연 # variation — 설계 §7 확정). [Camera:] prepend/sid 치환 생략 — # 구도 지시 0 계약 + 9절엔 sid 토큰 없음. valid_variations.append( (vi, var, _od_prompt, var_theme, var_theme_label) ) continue var_t2i = var.get("t2i_prompt", "") # Image N 치환 (모든 variation에 동일 매핑 적용) var_t2i = _rewrite_t2i_helper(var_t2i, _sid_to_img, _sid_info) # v4: 촬영 기법 이름을 카메라 directive로 prepend shot_name = var.get("shot_name", "") if shot_name and shot_name.lower() not in var_t2i.lower(): var_t2i = f"[Camera: {shot_name}] {var_t2i}" if var_t2i: valid_variations.append((vi, var, var_t2i, var_theme, var_theme_label)) # 마지막 N-1개만 선택 (프롬프트 수와 무관) target_variations = valid_variations[-max_t2i:] logger.info("Scene %d: %d prompts total, generating last %d T2I + 1 angle", si, len(valid_variations), len(target_variations)) # D5 T4 §4.6: chain_bg_lookup 1회 build — variation 마다 재계산 회피. # (씬 단위 RPC resolution 은 resolver 호출 이전으로 이동 — FINDING 9 Cat3.) chain_bg_lookup = build_chain_bg_lookup(background_chain_bg_map) # Patch B-min — moderation sanitize 시점의 polarity 보존 contract. # variation 별로 invariant (shot 단위) → 한 번만 build 해서 변종 loop 에 # 전달. None 일 때 (Rule 1+2 모두 미발화) sanitize 기존 path. _semantic_contract = build_semantic_contract( shot_staging=_stg, render_prompt_card=still_data.get("render_prompt_card"), visible_entities=visible_entities, ) _semantic_constraints = _semantic_contract.sanitizer_constraints # Area #5 W3 §4.3 (Codex iter 2 BLOCKING fix): per-variation sidecar # missing 시 pre-submit typed failure accumulator. executor 진입 전 # detect 하여 worker thread 낭비 회피 + cp.failed 직렬화 의도 explicit. _pre_submit_failures: List[Dict[str, Any]] = [] with ThreadPoolExecutor(max_workers=min(len(target_variations) or 1, 3)) as var_executor: # W20E5 Codex B2 — propagate parent-thread image-call budget # into variation pool workers (each variation may call # GeminiImageClient.generate_image via generate_and_validate_scene). _submit_variation = bind_current_budget(self._generate_variation_in_loop) var_futures = {} for vi, var, var_t2i, var_theme, var_theme_label in target_variations: # Area #5 W3 §4.3: per-variation sidecar direct read — producer # (scene_detail v26 W1) 이 each variation 에 emit. helper # build_scene_attached_refs 의 single-path read 와 보완 — variation # path 는 각 variation 의 sidecar 가 phantom guard 의 SOT. # No Silent Fallback — missing 시 typed failure dict 반환 (Codex # iter 2 BLOCKING fix). `.get(..., [])` silent fallback 절대 금지. if "reference_phrase_kinds" not in var: _pre_submit_failures.append({ "_failure_reason": "REF_CONTRACT_VIOLATION", "_failure_detail": ( f"variation vi={vi} missing 'reference_phrase_kinds' " f"sidecar — producer (scene_detail v26 W1) emit " f"required; stale v25 cp? rerun scene_detail to bump" )[:500], }) continue _var_phrase_kinds_raw = var["reference_phrase_kinds"] if _var_phrase_kinds_raw is None: _pre_submit_failures.append({ "_failure_reason": "REF_CONTRACT_VIOLATION", "_failure_detail": ( f"variation vi={vi} reference_phrase_kinds is None — " f"producer emit malformed" )[:500], }) continue # Codex iter 3 IMPORTANT fix: non-list malformed sidecar 가 list(...) # 변환으로 validator 의 "must be list" fail-fast 우회 방지. if not isinstance(_var_phrase_kinds_raw, list): _pre_submit_failures.append({ "_failure_reason": "REF_CONTRACT_VIOLATION", "_failure_detail": ( f"variation vi={vi} reference_phrase_kinds malformed: " f"must be list, got " f"{type(_var_phrase_kinds_raw).__name__}=" f"{_var_phrase_kinds_raw!r}" )[:500], }) continue _var_phrase_kinds = list(_var_phrase_kinds_raw) # Area #11 v1 W2: variation submit 시 payload (LabeledRefPayload) 전달 — # labeled_refs / attached_meta / ref_roles / ref_role_metadata 4 list parallel # 을 1 container 로 packing (spec §3.3). _var_payload = make_labeled_ref_payload( labeled_refs=labeled_refs, ref_roles=ref_roles, ref_role_metadata=ref_role_metadata, attached_meta=attached_meta, ) # W5 F22 Phase B.22.3: nested → bound method # W20E5 Codex B2: _submit_variation wraps the bound method # with bind_current_budget so child threads see the budget. var_futures[var_executor.submit( _submit_variation, still_data=still_data, var_t2i=var_t2i, var_theme=var_theme, var_theme_label=var_theme_label, visible_entities=visible_entities, payload=_var_payload, rpc=rpc, is_close_framing=_is_close_framing, chain_bg_lookup=chain_bg_lookup, reference_phrase_kinds=_var_phrase_kinds, gemini_client=gemini_client, sanitizer=sanitizer, validator=validator, scene_dir=scene_dir, cached_style_context=cached_style_context, cached_entity_text_map=cached_entity_text_map, world_guide=world_guide, episode_id=episode_id, semantic_constraints=_semantic_constraints, )] = vi var_results = self._await_variation_results(var_futures) # Area #5 W3 §4.3: pre-submit typed failures 를 var_results 에 병합 (cp.failed # 직렬화 — Fix C parity). if _pre_submit_failures: var_results.extend(_pre_submit_failures) return (si, var_results, visible_entities, current_location_ids) def _await_variation_results(self, var_futures: Dict[Any, int]) -> List[Dict[str, Any]]: """variation future loop — completes all + propagates StaleUpstreamError. D6 T9-fix BLOCKING: 이전 ``except Exception`` 만으로는 StaleUpstreamError (validator fallback / preflight bypass path) 가 generic logging 으로 swallow 되어 ``var_results`` 가 빈 list 로 반환 → 상위가 "all variations failed" 로 scene_cp 기록 → STALE_UPSTREAM 의 운영 신호 손실. fix: StaleUpstreamError 는 별도 catch 후 즉시 re-raise (scene-wide stale 의미 — single variation 실패 가 아니라 catalog/render manifest stale, fail-fast 가 정합). Fix C (2026-05-10): generic Exception 도 silent swallow 하지 않고 typed failure dict (`_failure_reason`/`_failure_detail`/`_failure_traceback`) 로 변환해 var_results 에 누적. scene_image_service 가 cp.failed 메시지 직렬화 시 사용 — 24 shot deterministic 실패의 root cause 가시화. """ import traceback as _traceback from app.core.errors import StaleUpstreamError var_results: List[Dict[str, Any]] = [] for future in as_completed(var_futures): try: result = future.result() if result: var_results.append(result) except StaleUpstreamError: raise except Exception as exc: # ★정지·락 상실은 **이 variation 하나의 실패가 아니다.** 여기서 # 실패 dict 로 바꿔 담으면 「그 한 장만 실패」로 기록되고 나머지 # variation·씬은 계속 돈다 — 멈추라는 말이 안 들리는 것과 같다. # 위로 올려 씬 배치와 스텝을 차례로 세운다. # # ★별도 except 절로 빼지 않는다. 그러면 정지와 무관한 오류까지 # 가로채 「한 장이 실패해도 나머지는 간다」는 원래 동작이 사라진다. from app.core.run_control import is_abort # ★코드 목록은 **한 곳**(`ABORT_CODES`)이다 — # 여기 다시 적으면 한쪽만 고쳐진다 if is_abort(exc): logger.warning( "variation 중단 — %s", getattr(exc, "message", str(exc)), ) raise tb = _traceback.format_exc() logger.error( "Variation generation thread error: %r\n%s", exc, tb, ) var_results.append({ "_failure_reason": "GENERIC_EXCEPTION", "_failure_detail": f"{type(exc).__name__}: {exc}"[:500], "_failure_traceback": tb[:2000], }) return var_results def _run_fal_angle_pipeline( self, *, var_results_list: List[Dict[str, Any]], saved_asset_ids: List[str], still_data: Dict[str, Any], stills: List[Dict[str, Any]], scene_dir: Path, episode_id: str, si: int, ) -> bool: """Run fal.ai angle pipeline for a scene (N+1 image). W5 F22 Phase B.22.6 (2026-04-22): scene_image_service.generate_images의 fal.ai 앵글 블록(~95 LOC)을 bound method로 lift. 처리 (literal lift): 1. settings.fal_ai_enabled + settings.fal_key + var_results_list 비어있지 않으면 진입 2. 각 variation file_path에서 bytes 수집 3. dependent_scene_id 있으면 _prev_title/_prev_desc 로드 4. _select_and_recommend_angle → angle_selection (None이면 skip) 5. _apply_fal_angle → fal_bytes (실패 시 skip) 6. fal_bytes 있으면 파일 저장 + PNG 메타데이터 embed + DB asset 저장 7. var_results_list와 saved_asset_ids에 append (in-place) 8. 전체 try/except — 실패 시 warning log + fal_generated=False Returns fal_generated bool. var_results_list / saved_asset_ids mutate 발생. """ fal_generated = False if not (settings.fal_ai_enabled and settings.fal_key and len(var_results_list) > 0): return fal_generated try: # Collect image bytes from all N variations angle_image_bytes = [] for vr in var_results_list: fp = Path(vr["file_path"]) if vr.get("file_path") else None if fp and fp.exists(): angle_image_bytes.append(fp.read_bytes()) if not angle_image_bytes: return fal_generated # 앞쪽 씬 텍스트 맥락 (시각적 연속성 참고) _prev_title = "" _prev_desc = "" dep_id = still_data.get("dependent_scene_id") if dep_id: for _ss in stills: if _ss.get("id") == dep_id: _prev_title = _ss.get("beat_title", "") _prev_desc = _ss.get("still_frame_prompt", "") break angle_selection = _select_and_recommend_angle( angle_image_bytes, still_data.get("beat_title", ""), still_data.get("t2i_prompt_cinematic", ""), prev_scene_title=_prev_title, prev_scene_description=_prev_desc, ) if not angle_selection: return fal_generated sel_idx = angle_selection["best_for_angle"] logger.info( "Scene %d: angle selection — image %d/%d, H=%.0f V=%.0f Z=%.0f — %s", si, sel_idx + 1, len(var_results_list), angle_selection["horizontal_angle"], angle_selection["vertical_angle"], angle_selection["zoom"], angle_selection.get("reason", ""), ) # Apply fal.ai angle to selected image sel_result = var_results_list[sel_idx] if sel_idx < len(var_results_list) else var_results_list[0] sel_path = Path(sel_result["file_path"]) if sel_result.get("file_path") else None sel_asset_id = saved_asset_ids[sel_idx] if sel_idx < len(saved_asset_ids) else saved_asset_ids[0] if not (sel_path and sel_path.exists()): return fal_generated fal_bytes, fal_elapsed = _apply_fal_angle( sel_path.read_bytes(), angle_selection["horizontal_angle"], angle_selection["vertical_angle"], angle_selection["zoom"], ) if not fal_bytes: return fal_generated fal_id = _new_id() fal_path = scene_dir / f"{fal_id}.png" fal_path.write_bytes(fal_bytes) # PNG 메타데이터 삽입 self._provenance_svc.embed_fal_angle_metadata( file_path=fal_path, source_image_name=sel_path.name, horizontal_angle=angle_selection["horizontal_angle"], vertical_angle=angle_selection["vertical_angle"], zoom=angle_selection["zoom"], created_at=_now(), ) self._persistence_svc.save_fal_angle_asset( asset_id=fal_id, file_path=str(fal_path), still_id=still_data["id"], episode_id=episode_id, prompt_used=( f"[fal.ai angle] H={angle_selection['horizontal_angle']}" f" V={angle_selection['vertical_angle']}" f" Z={angle_selection['zoom']}" ), source_image_id=sel_asset_id, created_at=_now(), ) # Append fal result to tracking lists (in-place mutation) saved_asset_ids.append(fal_id) var_results_list.append({ "id": fal_id, "file_path": str(fal_path), "asset_type": "scene", "still_id": still_data["id"], "episode_id": episode_id, "variant_type": "angle_fal", }) fal_generated = True logger.info( "Scene %d: fal.ai angle image saved (%d ms) → %s", si, fal_elapsed, fal_id, ) except Exception as exc: logger.warning("fal.ai angle pipeline failed for scene %d: %s", si, exc) return fal_generated def _finalize_and_track_primary( self, *, best_idx: int, saved_asset_ids: List[str], var_results_list: List[Dict[str, Any]], still_data: Dict[str, Any], si: int, scene_cp: Any, scene_results_by_index: Dict[int, Dict[str, Any]], scene_paths_by_index: Dict[int, Path], scene_paths_by_index_by_id: Dict[str, Path], scene_primary_asset_id_by_still_id: Optional[Dict[str, str]] = None, ) -> Path: """Primary asset 설정 + 3개 tracking map 갱신 + 체크포인트 기록. W5 F22 Phase B.22.9 (2026-04-23): scene_image_service.generate_images의 primary 마킹/추적/체크포인트 블록(~22 LOC)을 bound method로 lift. 2026-04-27: sd_shot_path_map 인자 제거 (downstream 미사용 dead variable). 처리: 1. selected_id = saved_asset_ids[best_idx] fallback [0] 2. persistence_svc.set_primary_asset(selected_id) 3. primary_result = var_results_list[best_idx] fallback [0] 4. scene_results_by_index[si] = primary_result (무조건 기록) 5. primary_path가 exists()면 path map 2개 업데이트 6. scene_cp.mark_completed(still_id, {asset_ids, primary_id, primary_path}) Returns: primary_path (caller가 location_scene_history 업데이트에 사용) """ selected_id = ( saved_asset_ids[best_idx] if best_idx < len(saved_asset_ids) else saved_asset_ids[0] ) self._persistence_svc.set_primary_asset(selected_id) primary_result = ( var_results_list[best_idx] if best_idx < len(var_results_list) else var_results_list[0] ) scene_results_by_index[si] = primary_result primary_path = Path(primary_result["file_path"]) if primary_path.exists(): scene_paths_by_index[si] = primary_path scene_paths_by_index_by_id[still_data.get("id", "")] = primary_path # Phase C: zoom crop source lineage 용 평행 맵 — still_id→primary asset UUID. # 다음 batch 의 zoom worker(DB 금지)가 source_asset_id 로 읽는다. if scene_primary_asset_id_by_still_id is not None: scene_primary_asset_id_by_still_id[still_data.get("id", "")] = selected_id scene_cp.mark_completed(still_data.get("id", ""), { "asset_ids": saved_asset_ids, "primary_id": selected_id, "primary_path": str(primary_path), }) return primary_path def _finalize_scene_with_variations( self, *, still: Any, still_id: str, original_result: Dict[str, Any], original_id: str, variant_a_result: Optional[Dict[str, Any]], variant_b_result: Optional[Dict[str, Any]], ip: Optional[str], ) -> Dict[str, Any]: """generate_scene_with_variations 최종화: recommended primary + commit + log + dict. W5 F22 Phase B.24.2 (2026-04-23): A/B variation 생성 후 recommended 설정 + db.commit() + activity log + return 블록(~30 LOC)을 bound method로 이관. 처리 (literal 보존): 1. `recommended = still.recommended_variant or 'original'` 기본값 2. recommended == 'A' + variant_a_result 있으면 A primary elif 'B' + variant_b_result 있으면 B primary (그 외: 아무것도 안 함 — original이 이미 primary) 3. `self._db.commit()` 4. activity log: action='still.generate_with_variations', resource_type='still', resource_id=still_id, detail={original_id, variant_a_id, variant_b_id, recommended} 5. 반환 dict: {original, variant_a, variant_b, recommended} _set_variant_primary_helper는 module-level (sib=0, target=1 update, commit은 caller). """ recommended = still.recommended_variant or "original" if recommended == "A" and variant_a_result: _set_variant_primary_helper(self._db, self._project_id, variant_a_result["id"], still_id) elif recommended == "B" and variant_b_result: _set_variant_primary_helper(self._db, self._project_id, variant_b_result["id"], still_id) self._db.commit() self._logger.log( actor_id=self._actor_id, action="still.generate_with_variations", resource_type="still", resource_id=still_id, project_id=self._project_id, detail={ "original_id": original_id, "variant_a_id": variant_a_result["id"] if variant_a_result else None, "variant_b_id": variant_b_result["id"] if variant_b_result else None, "recommended": recommended, }, ip_address=ip, ) return { "original": original_result, "variant_a": variant_a_result, "variant_b": variant_b_result, "recommended": recommended, } def _maybe_generate_variation_slot( self, *, slot: str, still: Any, i2i_editor: Any, original_bytes: bytes, original_id: str, scene_dir: Path, episode_id: str, lineage: Dict[str, Any], ) -> Optional[Dict[str, Any]]: """generate_scene_with_variations의 variation_{slot}_* 슬롯 기반 i2i 생성. W5 F22 Phase B.24.1 (2026-04-23): A/B variation 호출 중복 블록(각 ~16 LOC) 공통화. slot='a' 또는 'b'를 받아 getattr로 still의 동적 필드 접근. 처리 (literal 보존): 1. still.variation_{slot}_type이 truthy이고 'none'이 아니면 i2i 진입 2. variation_svc._generate_variation_i2i 호출: - variant_label=f'variant_{slot}' - var_type, angle_json, color_prompt: getattr로 still의 variation_{slot}_{type,angle,color} 3. type이 없거나 'none'이면 None 반환 """ var_type = getattr(still, f"variation_{slot}_type", None) if not var_type or var_type == "none": return None return self._variation_svc._generate_variation_i2i( i2i_editor=i2i_editor, original_bytes=original_bytes, original_id=original_id, still=still, variant_label=f"variant_{slot}", var_type=var_type, angle_json=getattr(still, f"variation_{slot}_angle", None), color_prompt=getattr(still, f"variation_{slot}_color", None), scene_dir=scene_dir, episode_id=episode_id, lineage=lineage, ) def build_single_still_context( self, *, episode_id: str, still: Any, ) -> Dict[str, Any]: """Task 2 of single-vs-batch reference contract fix. 단건 generate_single_scene_image 의 helper input prep — batch generate_images 의 사전 build 코드와 **동일 source 호출** 로 9 fields build (DRY 강제). spec: docs/superpowers/specs/2026-05-08-single-batch-reference-contract-design.md §4.1 plan v2 Task 2: build_single_still_context — batch 동일 source. Source mapping (vs batch generate_images): - entity_lookup = ref_svc.load_episode_entity_lookup(episode_id) - ref_image_map = ref_svc.load_entity_reference_images(...) - background_chain_bg_map = load_background_chain_bg_map(...) - dep_detail_map = load_shot_dependency_map(...) - staging_map → staging = load_shot_staging_map(...)[scene_shot_key] - cached_style_context = self._get_style_context() - cached_entity_text_map = ref_svc.build_entity_text_map(entities) - scene_paths_by_index_by_id, location_scene_history = persistence_svc.build_resume_state(...) - stills = persistence_svc.fetch_episode_stills(...) """ from app.services.scene_checkpoint_loaders import ( load_background_chain_bg_map, load_outdoor_direct_context, load_shot_dependency_map, load_shot_staging_map, ) pid = self._project_id # 1. entity_lookup entity_lookup = self._reference_svc.load_episode_entity_lookup(episode_id) # 2. ref_image_map ref_image_map = self._reference_svc.load_entity_reference_images( list(entity_lookup.values()) ) # 3. background_chain_bg_map background_chain_bg_map = load_background_chain_bg_map( settings.projects_dir, pid, episode_id, ) # 3b. W22 야외 직행 컨텍스트 — flag OFF/cp 부재 = {} (no-op) outdoor_direct_ctx = load_outdoor_direct_context( settings.projects_dir, pid, episode_id, ) # 4. dep_detail_map dep_detail_map = load_shot_dependency_map( settings.projects_dir, pid, episode_id, ) # 5. staging (per-shot dict — staging_map[key] lookup) staging_map = load_shot_staging_map(settings.projects_dir, pid, episode_id) _staging_key = f"{still.scene_index}_{still.shot_index}" staging = staging_map.get(_staging_key) if staging_map else None # 6. cached_style_context (episode 단위 — 단건은 1회 build) cached_style_context = self._get_style_context(episode_id) # 7. cached_entity_text_map (episode 단위) # ★화를 넘긴다 — 안 넘기면 이 화의 배정을 못 찾아 C##O## 합성 키가 # 통째로 안 만들어진다 (2026-09-04). cached_entity_text_map = self._reference_svc.build_entity_text_map( list(entity_lookup.values()), episode_id=episode_id ) # 8. stills + 9. scene_paths_by_index_by_id + location_scene_history # — persistence_svc.build_resume_state 가 batch 와 동일 source. # Fix-B (2026-06-11 fresh full E2E 발견#8): 기존 `set()` 전달은 "전체 # build" 의도였으나 build_resume_state 는 빈 set 을 "복원 대상 없음" 으로 # early-return — 단건 재생성 경로의 scene_paths/dep bytes/location_history # 가 항상 빈 상태였다 (silent; batch run 은 in-memory 누적이라 가려짐 — # S12 sh12 재생성에서 dep_scene bytes missing 실측). 전체 still id 집합을 # 명시 전달해 의도("전체 build")와 동작을 일치시킨다. stills, _stills_orm = self._persistence_svc.load_episode_still_dicts(episode_id) _all_still_ids = {s.get("id") for s in stills if s.get("id")} # B (2026-07-02): VE 에 location 이 없는 outdoor still 의 history 복원용 # primary_location fallback 맵 — flag OFF → {} = 기존과 byte-identical. from app.core.steps.outdoor_site_layout_step import ( load_outdoor_prev_frame_context, ) from app.modules.pipeline.outdoor_site_layout_plan import ( primary_location_uuid_by_scene, ) _opf_ctx_s = load_outdoor_prev_frame_context(pid, episode_id) _opf_uuid_by_scene_s = primary_location_uuid_by_scene( _opf_ctx_s.get("primary_location_by_scene") or {}, _opf_ctx_s.get("outdoor_loc_sids") or set(), entity_lookup, ) if _opf_ctx_s else {} _rs = self._persistence_svc.build_resume_state( stills, _all_still_ids, entity_lookup, fallback_location_uuid_by_scene=_opf_uuid_by_scene_s or None, ) return { "entity_lookup": entity_lookup, "ref_image_map": ref_image_map, "scene_paths_by_index_by_id": _rs.get("scene_paths_by_index_by_id", {}), "location_scene_history": _rs.get("location_scene_history", {}), "background_chain_bg_map": background_chain_bg_map, "outdoor_direct_ctx": outdoor_direct_ctx, "dep_detail_map": dep_detail_map, "staging": staging, "cached_style_context": cached_style_context, "cached_entity_text_map": cached_entity_text_map, "stills": stills, } def _build_single_scene_prompt_and_refs( self, *, still: Any, episode_id: str, still_data: Dict[str, Any], visible_entities: List[Dict[str, Any]], ref_image_map: Dict[str, bytes], var_t2i_override: Optional[str] = None, ) -> "tuple[str, list, list, object]": """generate_single_scene_image의 non-custom_prompt 경로 prompt+refs 구성. W5 F22 Phase B.23.5 (2026-04-23): _build_final_scene_prompt 경로 블록 (~48 LOC)을 bound method로 lift. 처리 (literal 보존): 1. `load_shot_t2i_variations` — 체크포인트에서 shot 변형 프롬프트 로드 2. var_t2i = variations[0].t2i_prompt or still_frame_prompt fallback 3. `load_episode_entity_lookup(episode_id)` 로드 4. `build_entity_text_map(entity_lookup.values())` 빌드 5. `build_scene_ref_image_map(ref_image_map, entity_lookup)` 빌드 6. `resolve_refs_for_prompt(var_t2i, visible, scene_ref, entity_lookup)` 7. `_get_style_context(episode_id)` 스타일 맥락 8. `_build_final_scene_prompt(...)` try/except — 예외 시 warning + var_t2i fallback D5 §4.3 (2026-05-09): return 을 (_full_prompt, labeled_refs, attached_meta) 3-tuple. T2 의 plumbing — T3 에서 validator 가 attached_meta 활용 예정. Returns: (_full_prompt, labeled_refs, attached_meta) """ # Task 2 (single-vs-batch reference contract): build_scene_attached_refs # helper 호출 — 단건 경로가 chain_bg / prev_shot_ref / state_variant / # close framing skip 정책을 batch 와 동등하게 적용. # spec: docs/superpowers/specs/2026-05-08-single-batch-reference-contract-design.md §4.1 ctx = self.build_single_still_context(episode_id=episode_id, still=still) # caller 가 받은 ref_image_map 우선 사용 (test/mock 호환), 없으면 ctx 사용 _ref_image_map = ref_image_map if ref_image_map else ctx["ref_image_map"] full_prompt, labeled_refs, attached_meta, reference_phrase_kinds, _attached_payload = build_scene_attached_refs( still=still, episode_id=episode_id, still_data=still_data, visible_entities=visible_entities, ref_image_map=_ref_image_map, var_t2i_override=var_t2i_override, cached_style_context=ctx["cached_style_context"], cached_entity_text_map=ctx["cached_entity_text_map"], scene_paths_by_index_by_id=ctx["scene_paths_by_index_by_id"], location_scene_history=ctx["location_scene_history"], background_chain_bg_map=ctx["background_chain_bg_map"], dep_detail_map=ctx["dep_detail_map"], staging=ctx["staging"], entity_lookup=ctx["entity_lookup"], project_id=self._project_id, project_config=load_project_llm_config(self._db, self._project_id), reference_svc=self._reference_svc, stills=ctx["stills"], outdoor_direct_ctx=ctx.get("outdoor_direct_ctx"), # W22 직행 return_payload=True, # P0: 5-tuple(payload 포함) — 단건 lineage 영속화. ) # Task 3 (single-vs-batch reference contract §4.2): validate_attached_refs # fail-fast 한 지점 — helper 호출 직후. rpc + is_close_framing 정보로 # required_refs 비교 + classifier-based "from the reference" guard. # 위반 시 RefContractError raise (HTTP 422), generate 차단. # # D1 patch (단건 RPC inject): scene_detail/manifest.json 에서 RPC lookup. # 단건 경로는 SceneStill ORM 만 보므로 manifest 의 render_prompt_card 가 # still_data 에 inject 안 되어 required_refs 검사 skip 되던 결함 보정. # lookup 실패 시 RefContractError propagate (silent skip 금지). from app.core.ref_contract_validator import validate_attached_refs _staging = ctx.get("staging") _framing_scale = get_framing_scale_or_raise( _staging, where="scene_generation_coordinator.single_scene_attached_ref_validator", ) _is_close_framing = _framing_scale == FRAMING_CLOSE # caller 가 이미 inject 했으면 우선, 없으면 manifest lookup _rpc = still_data.get("render_prompt_card") if _rpc is None: _rpc = lookup_render_prompt_card( project_id=self._project_id, episode_id=episode_id, scene_index=still.scene_index, shot_index=still.shot_index, ) still_data["render_prompt_card"] = _rpc # W1-A (2026-06-11): space_set_bg no-plate 진단 shot 의 required # background runtime reconciliation (사본, persisted RPC 불변). _rpc = waive_required_background_if_no_plate( _rpc, ctx["background_chain_bg_map"], still_data, ) still_data["render_prompt_card"] = _rpc # D5 T3 (사용자 binding): validator 가 attached_meta + chain_bg_lookup # 받음. substring fallback 폐기 — (kind, id) exact match. # G3: chain_bg_lookup(bg_id)=None 이면 prev_shot lineage 통과 X. _chain_bg_lookup = build_chain_bg_lookup(ctx["background_chain_bg_map"]) # FINDING C W2 (Category A): background over-declaration consumer # normalization — attached_meta 확정 후, validator 호출 직전. reference_phrase_kinds = _normalize_background_phrase_kind( reference_phrase_kinds, attached_meta, _rpc, ) validate_attached_refs( _rpc, labeled_refs, attached_meta, full_prompt, is_close_framing=_is_close_framing, chain_bg_lookup=_chain_bg_lookup, reference_phrase_kinds=reference_phrase_kinds, ) # P0: _attached_payload 도 반환(4-tuple) — 단건 scene_result lineage 영속화. return full_prompt, labeled_refs, attached_meta, _attached_payload def _finalize_single_scene( self, *, scene_result: Dict[str, Any], visible_entities: List[Dict[str, Any]], still_id: str, ip: Optional[str], ) -> Dict[str, Any]: """Single scene 생성 후처리: lineage + asset 저장 + activity log + image dict. W5 F22 Phase B.23.4 (2026-04-23): generate_single_scene_image의 후처리 블록(~21 LOC)을 bound method로 lift. 처리: 1. `ref_entity_ids` = visible_entities의 id 목록 2. `build_lineage_fields` 호출 (prompt_type='original') 3. `save_single_scene_asset(scene_result, lineage)` 위임 4. activity log: action='image.generate_single_scene', detail={'still_id'} 5. `image_to_dict(asset)` 반환 literal 보존: builder/save/log 순서 + 페이로드 필드 + prompt_type='original'. """ ref_entity_ids = [e["id"] for e in visible_entities] lineage = build_lineage_fields( self._db, self._project_id, "scene_image_generator", ref_entity_ids=ref_entity_ids, prompt_type="original", ) asset = self._persistence_svc.save_single_scene_asset(scene_result, lineage) self._logger.log( actor_id=self._actor_id, action="image.generate_single_scene", resource_type="image", resource_id=scene_result["id"], project_id=self._project_id, detail={"still_id": still_id}, ip_address=ip, ) return image_to_dict(asset) def _generate_variation_in_loop( self, *, still_data: Dict[str, Any], var_t2i: str, var_theme: str, var_theme_label: str, visible_entities: list, payload: LabeledRefPayload, rpc: Optional[Dict[str, Any]], is_close_framing: bool, chain_bg_lookup: Optional[Callable[[str], Optional[str]]], reference_phrase_kinds: List[str], gemini_client: Any, sanitizer: Any, validator: Optional[ImageValidator], scene_dir: Path, cached_style_context: str, cached_entity_text_map: Dict[str, str], world_guide: Dict[str, Any], episode_id: str, semantic_constraints: Optional[Dict[str, Any]] = None, ) -> Optional[Dict[str, Any]]: """Generate a single variation image for a scene (runs in thread pool). W5 F22 Phase B.22.3 (2026-04-22): generate_images nested function _generate_one_variation을 bound method로 lift. closure 변수 8개 (gemini_client, sanitizer, validator, scene_dir, cached_*, world_guide, episode_id)를 explicit keyword param으로 승격. D5 T4 §4.6 (B2 architectural fix): variation 별 _full_prompt 가 다르므로 phantom guard 도 variation-specific. validator 는 _build_final_scene_prompt 직후 + generate_and_validate_scene 직전에 호출. RefContractError 시 그 variation 만 skip (다른 variation 계속). Returns dict with result data or None on failure. Retry 3회 (원본 1회 + sanitized 2회), ModerationError 및 재시도 가능 일반 예외 처리 보존. """ from app.modules.pipeline.scene_image_pipeline import generate_and_validate_scene from app.core.ref_contract_validator import validate_attached_refs, RefContractError try: # Area #11 v1 W2: payload arg cascade (was: labeled_refs). _full_prompt = _build_final_scene_prompt( var_t2i, payload, cached_style_context, entity_text_map=cached_entity_text_map, ) except Exception as _prompt_exc: logger.warning("Prompt build failed for variation %s: %s", var_theme, _prompt_exc) _full_prompt = f"{cached_style_context}\n\n{var_t2i}" if cached_style_context else var_t2i # D5 T4 §4.6 (B2): full_prompt build 후, generate_and_validate_scene 전. # variation 단위 fail-fast — RefContractError 시 그 variation 만 skip, # 다른 variation/still 은 계속 (기존 fail-fast 정책 보존). try: # W2-A (2026-06-11): dep continuity skip 의 waiver 는 주입 단계에서 # still_data flag 로 기록된다 — rpc 는 주입 전에 resolve 되므로 # validate 직전 재적용 (sentinel 경로는 resolve 시점에 이미 반영, # helper 는 idempotent). bg_map 은 flag 경로에선 불필요라 빈 dict. rpc = waive_required_background_if_no_plate(rpc, {}, still_data) # FINDING C W2 (Category A): background over-declaration consumer # normalization — payload.attached_meta 확정 후, validator 호출 직전. reference_phrase_kinds = _normalize_background_phrase_kind( reference_phrase_kinds, payload.attached_meta, rpc, ) # Area #11 v1 W2: payload.labeled_refs / payload.attached_meta 사용. validate_attached_refs( rpc, payload.labeled_refs, payload.attached_meta, _full_prompt, is_close_framing=is_close_framing, chain_bg_lookup=chain_bg_lookup, reference_phrase_kinds=reference_phrase_kinds, ) except RefContractError as _ref_exc: logger.error( "Scene %d Shot %s variation=%s: ref contract violation — skip variation: %s", still_data.get("scene_index", 0), still_data.get("shot_index") or still_data.get("_shot_index"), var_theme, _ref_exc, ) # Fix C (2026-05-10): None 반환 대신 typed failure dict 반환 → # scene_image_service 가 cp.failed[still_id] 에 reason 직렬화 가능. # 기존 None 시그니처는 cp 가 generic "all variations failed" 만 남겨 # 24 shot deterministic 실패 root cause 가시성 차단했음. return { "_failure_reason": "REF_CONTRACT_VIOLATION", "_failure_detail": str(_ref_exc)[:500], } # 순화 retry: 원본 1회 + sanitized 2회 _current_prompt = _full_prompt _sanitization_info = None for _attempt in range(3): try: pipe_result = generate_and_validate_scene( gemini_client=gemini_client, t2i_prompt=_current_prompt, beat_title=still_data.get("beat_title", ""), output_dir=scene_dir, reference_images=payload.labeled_refs if payload.labeled_refs else None, previous_scene_bytes=None, # W3 observability — llm_call_log PID/EID NULL 차단. trace_meta={ "project_id": self._project_id, "episode_id": episode_id, "operation_type": "scene_image_gen", "scene_index": still_data.get("scene_index"), "shot_index": still_data.get("shot_index") or still_data.get("_shot_index"), "still_id": still_data.get("id"), }, semantic_constraints=semantic_constraints, ) validation_score = None validation_result_str = None v_status = "generated" v_review_notes = json.dumps(pipe_result.get("validation", {}), ensure_ascii=False) if validator: entity_names = [e.get("name", "") for e in visible_entities] scene_info = { "scene_heading": still_data.get("screenplay_scene_heading", ""), "beat_title": still_data.get("beat_title", ""), "still_frame_prompt": still_data.get("still_frame_prompt", ""), "world_context": world_guide.get("world_setting_summary", ""), } validation_score, validation_result_str, v_status, v_review_notes = ( self._validation_svc.validate_scene( validator, pipe_result["file_path"], scene_info, entity_names, v_status, v_review_notes, ) ) return { "id": _new_id(), "asset_type": "scene", "entity_id": None, "still_id": still_data.get("id"), "episode_id": episode_id, "file_path": pipe_result["file_path"], "prompt_used": _current_prompt, "generation_model": pipe_result["generation_model"], "width": None, "height": None, "status": v_status, "review_notes": v_review_notes, "validation_score": validation_score, "validation_result": validation_result_str, "variant_type": var_theme, "theme_label": var_theme_label, "sanitization_strategy": _sanitization_info.get("strategy") if _sanitization_info else None, "sanitization_note": _sanitization_info.get("note") if _sanitization_info else None, # goal#3 (Codex 설계 C): 이 still 에 실제 attach 된 pose/구도 가이드 # ImageAsset UUID 만 수집(구조 lineage SOT, 라벨파싱 금지). 최종 scene # asset input_image_ids 로 복원 → pipeline_graph generated_input guide→ # scene 엣지 자동 생성. char/prop reference 는 별도(reference_image_ids). # indoor_pose_guide(실내) + composition_guide(실외 camera_sketch) 둘 다. "pose_guide_asset_ids": [ m.get("asset_id") for m in payload.ref_role_metadata if isinstance(m, dict) and m.get("pipeline_role") in ( "indoor_pose_guide", "composition_guide") and m.get("asset_id") ], # P0 (2026-07-01): 실제 첨부한 모든 image asset UUID + 상세. # save_scene_variations 가 input_image_ids + actual_attached_refs # 로 영속화 → 모달/캔버스 generated_input 엣지 진실화. **_attached_lineage_fields(payload), # A5 (2026-07-02): prev-frame chaining 진단(부착/미부착 사유) — # persistence 가 pipeline_metadata 로 영속 (Codex 합의 필드). **({"prev_frame_chain": still_data.get("_prev_frame_chain_diag")} if still_data.get("_prev_frame_chain_diag") else {}), "created_at": _now(), } except ModerationError as exc: if _attempt >= 2: logger.warning("Shot %s variation %s blocked after %d sanitize retries: %s", still_data.get("id", "")[:8], var_theme, _attempt, exc.block_reason) return None try: _san = sanitizer.sanitize( _current_prompt, exc.block_reason, exc.block_categories, attempt=_attempt + 1, semantic_constraints=semantic_constraints, ) _current_prompt = _san.get("sanitized_prompt", _current_prompt) _sanitization_info = _san logger.info("Shot %s sanitized (attempt %d, strategy: %s)", still_data.get("id", "")[:8], _attempt+1, _san.get("strategy", "")) except Exception: logger.warning("Sanitization failed for shot %s", still_data.get("id", "")[:8]) return None except AppError: # Phase 4 iter 7 B1 — LVM contract / 기타 deterministic AppError # 는 retry 대상이 아님. retry 후 return None 으로 변질되면 step # level 의 fail-fast catch 가 silent skip 으로 무력화 (W1 의도 # 위반). 즉시 re-raise 하여 batch except 가 mark_failed 처리. raise except Exception as exc: # 재시도 가능한 에러 (MALFORMED_FUNCTION_CALL, timeout 등) if _attempt < 2: logger.warning("Shot %s variation %s failed (attempt %d/3, retrying): %s", still_data.get("id", "")[:8], var_theme, _attempt + 1, exc) import time; time.sleep(3) continue logger.error("Shot %s variation %s failed after 3 attempts: %s", still_data.get("id", "")[:8], var_theme, exc) return None return None # for 루프 정상 종료 시 (도달 불가)