"""SceneContextLoader — 20+ 체크포인트 로드를 단일 진입점으로. Phase 3.6. 기존 `detail_steps._execute`의 170+ 줄 로드 로직을 이 Loader로 추출. 목적: - `_execute`는 "로드 → 프롬프트 구성 → 병렬 실행"의 3단계로 축소. - 테스트에서 fixture로 `SceneAnalysisContext`만 조립하면 _analyze_one 단위 테스트 가능. """ from __future__ import annotations import logging import re from typing import Any, Dict, List, Optional, Tuple from sqlalchemy import text as sql_text from app.core.dto.scene_analysis import SceneAnalysisContext from app.core.steps._evidence_helpers import _normalize_scene_consistency_result logger = logging.getLogger(__name__) class SceneContextLoader: """SceneDetailStep 전용 체크포인트 로더. 입력: step runner 인스턴스 (self.project_id, episode_id, db, _load_prev_checkpoint). 출력: SceneAnalysisContext dataclass 인스턴스. """ def __init__(self, runner: Any): self.runner = runner def load_all(self) -> SceneAnalysisContext: ctx = SceneAnalysisContext( project_id=self.runner.project_id, episode_id=self.runner.episode_id, ) ctx.segments = self._load_segments() ctx.director_scenes = self._load_director_scenes() ctx.scene_visible = self._build_scene_visible(ctx.director_scenes) ctx.dependencies = self._load_dependencies() ctx.outlook_data = self._load_outlook_data() ctx.shot_type_rows = self._load_shot_type_rows() ctx.shot_types_block = self._build_shot_types_block(ctx.shot_type_rows) ctx.staging_map = self._load_staging_map() ctx.fixed_elements_by_scene = self._load_fixed_elements() # v0.5.10: location_consistency 주입 롤백 — shot_dependency_t2i의 이전 샷 참조 + ignore/keep이 더 효과적. # step 자체는 보존하되 scene_detail 주입만 비활성. ctx.location_visuals_by_id = {} ctx.scene_shots_map = self._load_legacy_shot_cinematography(ctx.shot_type_rows) ctx.entities = self._load_entities() ctx.shot_director_ve, ctx.shot_director_vr = self._load_shot_director() ctx.selected_map = self._load_selected_map() ctx.shot_scenes_map = self._load_shot_scenes_map(ctx.selected_map) ctx.beats_by_scene = self._load_beats_by_scene() ctx.summaries = self._load_summaries() ctx.world_rules = self._load_world_rules() ctx.planning_context = self._load_planning_context() ctx.chain_bg_guide_by_shot = self._load_chain_bg_guide_by_shot() ctx.chain_bg_camera_meta_by_shot = self._load_chain_bg_camera_meta_by_shot() ctx.chain_bg_owned_by_shot = self._load_chain_bg_owned_by_shot() ctx.chain_bg_id_by_shot = self._load_chain_bg_id_by_shot() ctx.essence_by_shot = self._load_essence_by_shot() # G4.6 Phase 4 Codex iter 1 B1 — main thread prebuild. # ThreadPool worker 안에서 SQLAlchemy session race 회피. # _build_entity_traits_block / validate_visible_entities_contract 가 # db 대신 본 map 사용. project_id + entity_type='character' scoped. ctx.name_by_short_id, ctx.traits_by_short_id = ( self._load_entity_canon_character_maps() ) # Task 14 Wave 2 — main thread prebuild for the prop_term_map used by # `reconcile_owned_prop_namespace_overlap` (visible-prop ↔ background- # owned namespace overlap reclass). Same threading rationale as the # character map above — must run on main thread before ThreadPool # workers dispatch. ctx.prop_term_map = self._load_entity_canon_prop_term_map() ctx.episode_reference_policy = self._load_episode_reference_policy() # shot visible ↔ staging camera_direction dual SOT drift fail-fast. # shot_director 가 staging 결과를 보지 못한 채 visible 결정 → staging # camera_direction NL 에 명시된 off-camera 인물이 director.visible 에 # 남아 있을 수 있다. consumer-side auto-fix 는 silent SOT 변조이므로 # fail-fast (운영자가 어느 step 을 force 할지 명시적으로 결정). self._assert_no_visible_staging_drift(ctx) return ctx # ── 개별 로드 헬퍼 ───────────────────────────────────── def _load_segments(self) -> List[Dict[str, Any]]: cp = self.runner._load_prev_checkpoint("scene_save") return cp.get("data", {}).get("segments", []) if cp else [] def _load_director_scenes(self) -> List[Dict[str, Any]]: cp = self.runner._load_prev_checkpoint("scene_director") return cp.get("data", {}).get("scenes", []) if cp else [] @staticmethod def _build_scene_visible(director_scenes: List[Dict[str, Any]]) -> Dict[int, List[str]]: scene_visible: Dict[int, List[str]] = {} for ds in director_scenes: si = ds.get("scene_index") scene_visible[si] = ds.get("present_entity_ids", []) return scene_visible def _load_dependencies(self) -> List[Dict[str, Any]]: """shot_dependency_t2i (v5 — zoom_in_detail 등 refined ref_usage 포함) 우선. 없으면 shot_dependency (1차) → scene_dependency (legacy) fallback. v0.5.18 drift 감지: shot_dependency_t2i의 DB step_run이 `stale`이면 — scene_detail force 이후 shot_dependency_t2i가 재실행되지 않은 상태 — warning 로그로 운영자에게 2-pass drift 가능성 통지. (파일은 scene_detail.consumes_downstream 정책으로 보존되어 있지만, 내용은 이전 scene_detail 기반이라 현재 t2i_prompt와 정합 안 됨 가능.) Area D-next-min (2026-05-15) + C7 (2026-05-20): shot_dependency_t2i cp 가 v8 schema 인지 (schema_version=4) + keep_elements shape 가 List[{label, kind∈{environment, static_prop}, subject_kind}] 인지 L3 fail-fast 검증. legacy v5 cp (schema_version=1 또는 missing, keep_elements List[str]) / legacy v6 cp (schema_version=2, kind=immobilized_character 포함) / legacy v7 cp (schema_version=3, subject_kind 부재) 가 본 loader 를 우회해 downstream (detail_steps forward_zoom_targets) 으로 silently 유입되는 것을 차단. scene_checkpoint_loaders.py 의 동일 contract 와 정합. character state 보존은 별도 layer (scene_consistency / character_state_variant / semantic_contract_router) 책임. """ from app.core.errors import AppError from app.core.keep_elements import validate_keep_elements t2i_cp = self.runner._load_prev_checkpoint("shot_dependency_t2i") if t2i_cp and t2i_cp.get("data", {}).get("dependencies"): # Area D-next-min L3 fail-fast — schema_version + shape 검증. schema_v = t2i_cp.get("schema_version") if schema_v != 4: raise AppError( code="step.scene_context_loader.legacy_keep_elements_cp", message=( f"shot_dependency_t2i checkpoint schema_version=" f"{schema_v!r}, expected 4 (C7 bump — keep_elements[] " f"subject_kind required field). Force re-run " f"shot_dependency_t2i 권장 — legacy v5 (List[str]) / v6 " f"(immobilized_character) / v7 (schema_version 3, " f"subject_kind 부재) cp 가 downstream (detail_steps " f"forward_zoom_targets) 에 silently 유입되는 것을 " f"차단합니다." ), status_code=409, ) for dep in t2i_cp["data"]["dependencies"]: si = dep.get("scene_index") shi = dep.get("shot_index") for idx, ref in enumerate(dep.get("location_refs", []) or []): if "keep_elements" not in ref: raise AppError( code="step.scene_context_loader.keep_elements_missing_key", message=( f"shot_dependency_t2i cp S{si}_Shot{shi} " f"location_refs[{idx}] missing required " f"'keep_elements' key — schema v8 required " f"field. value={ref!r}" ), status_code=400, ) validate_keep_elements( ref["keep_elements"], label_source=f"scene_context_loader S{si}_Shot{shi} loc_ref[{idx}]", error_code_entry="step.scene_context_loader.keep_elements_entry_invalid", error_code_legacy_str="step.scene_context_loader.keep_elements_legacy_str", ) t2i_run = self.runner._get_step_run("shot_dependency_t2i") if t2i_run and t2i_run["status"] == "stale": logger.warning( "scene_detail: shot_dependency_t2i checkpoint is STALE — " "이전 scene_detail 기반 ref_usage를 사용 중. drift 방지를 위해 " "scene_detail 완료 후 shot_dependency_t2i도 force 재실행 권장." ) return t2i_cp["data"]["dependencies"] shot_dep_cp = self.runner._load_prev_checkpoint("shot_dependency") if shot_dep_cp and shot_dep_cp.get("data", {}).get("dependencies"): return shot_dep_cp["data"]["dependencies"] dep_cp = self.runner._load_prev_checkpoint("scene_dependency") return dep_cp.get("data", {}).get("dependencies", []) if dep_cp else [] def _load_outlook_data(self) -> Dict[str, Any]: outlook_cp = self.runner._load_prev_checkpoint("outlook_phase3") if not outlook_cp or not outlook_cp.get("data"): outlook_cp = self.runner._load_prev_checkpoint("outlook_extraction") return outlook_cp.get("data", {}) if outlook_cp else {} def _load_shot_type_rows(self) -> List[Any]: return self.runner.db.execute(sql_text( "SELECT name, category, description, llm_description " "FROM shot_type WHERE is_active = true ORDER BY sort_order" )).fetchall() @staticmethod def _build_shot_types_block(rows: List[Any]) -> str: if not rows: return "" return "\n".join(f"- {r[0]} [{r[1]}]: {r[2]}" for r in rows) def _load_staging_map(self) -> Dict[str, Dict[str, Any]]: cp = self.runner._load_prev_checkpoint("shot_staging") staging_map: Dict[str, Dict[str, Any]] = {} if cp and cp.get("data", {}).get("shots"): for st in cp["data"]["shots"]: key = f"{st.get('scene_index')}_{st.get('shot_index')}" staging_map[key] = st logger.info("scene_detail: %d shot_staging entries loaded", len(staging_map)) return staging_map def _load_fixed_elements(self) -> Dict[int, List[Dict[str, Any]]]: """G2.1/G2.2 (Claude+Codex IMPORTANT defense in depth): partial / blocked / violation scene 의 fixed_elements 는 downstream 에 전달하지 않음. dispatcher 의 ``allow_partial_downstream=False`` 가 1차 차단이고, 본 가드는 force-run / manual override 같은 우회 path 에서도 violation element 가 LLM prompt 에 주입되지 않도록 하는 2차 안전망. 2-tier: 1) status 직접 차단 — failed/blocked/violation 2) status="ok" 또는 status=None 옛 cp — fixed_elements 직접 검증 (downstream-only force-run 시 dep status="completed" 면 dispatcher gate 통과 하지만 옛 cp 가 violation 가질 수 있음 — Codex iter#2 IMPORTANT). """ # iter#4 fix: consumer 가드 single-source helper 사용. circular import 회피 # 위해 function-local. 두 consumer (본 loader + shot_dependency_t2i_step) 가 # 동일 helper 호출 → 새 partial status 추가 시 drift 자동 방지. from app.core.steps.scene_consistency_step import is_scene_result_consumer_safe cp = self.runner._load_prev_checkpoint("scene_consistency") fixed_map: Dict[int, List[Dict[str, Any]]] = {} skipped_unsafe = 0 if cp and cp.get("data", {}).get("scenes"): for sc in cp["data"]["scenes"]: # G3.1: 옛 cp 4-field 누락 lazy backfill (마킹=legacy). _normalize_scene_consistency_result( sc, where="scene_context_loader._load_fixed_elements", ) si = sc.get("scene_index") if not is_scene_result_consumer_safe(sc): skipped_unsafe += 1 continue elements = sc.get("fixed_elements", []) if elements: fixed_map[si] = elements if skipped_unsafe: logger.warning( "scene_detail: %d unsafe scene(s) skipped — fixed_elements 미적용", skipped_unsafe, ) logger.info("scene_detail: %d scenes with fixed elements loaded", len(fixed_map)) elif cp is None: logger.warning("scene_detail: scene_consistency 체크포인트 없음 — 고정 요소 미적용") return fixed_map def _load_location_visuals(self) -> Dict[str, str]: """location_consistency 체크포인트에서 L## → fixed_visual_description 매핑. analysis_summary가 "실패" prefix인 항목은 entity_detail의 한국어·분위기 포함 description을 fallback으로 담고 있으므로 scene_detail 주입에서 제외한다. 외형 고정 목적과 배치되는 내용을 영구 주입하면 안 되므로. """ cp = self.runner._load_prev_checkpoint("location_consistency") visuals: Dict[str, str] = {} skipped = 0 if cp and cp.get("data", {}).get("locations"): for loc in cp["data"]["locations"]: lid = loc.get("location_id", "") desc = loc.get("fixed_visual_description", "") or "" summary = loc.get("analysis_summary", "") or "" if summary.startswith("실패"): skipped += 1 continue if lid and desc: visuals[lid] = desc logger.info( "scene_detail: %d location visuals loaded (skipped %d failed)", len(visuals), skipped, ) elif cp is None: logger.warning("scene_detail: location_consistency 체크포인트 없음 — 외형 고정 미적용") return visuals def _load_legacy_shot_cinematography( self, shot_type_rows: List[Any], ) -> Dict[int, List[Dict[str, Any]]]: cp = self.runner._load_prev_checkpoint("shot_cinematography") scene_shots_map: Dict[int, List[Dict[str, Any]]] = {} if not (cp and cp.get("data", {}).get("shots")): return scene_shots_map shot_desc_map = ( {r[0]: r[3] for r in shot_type_rows} if shot_type_rows else {} ) for sc in cp["data"]["shots"]: si = sc.get("scene_index") if si not in scene_shots_map: scene_shots_map[si] = [] for tkey in ["technique_1", "technique_2"]: t = dict(sc.get(tkey, {})) t["llm_description"] = shot_desc_map.get(t.get("name", ""), "") t["_shot_index"] = sc.get("shot_index", 0) scene_shots_map[si].append(t) return scene_shots_map def _load_entities(self) -> Dict[str, Any]: cp = self.runner._load_prev_checkpoint("entity_t2i") return cp.get("data", {}) if cp else {} def _load_shot_director( self, ) -> Tuple[Dict[Tuple[int, int], List[str]], Dict[Tuple[int, int], Dict[str, str]]]: cp = self.runner._load_prev_checkpoint("shot_director") ve_map: Dict[Tuple[int, int], List[str]] = {} vr_map: Dict[Tuple[int, int], Dict[str, str]] = {} if cp and cp.get("data", {}).get("scenes"): for sc in cp["data"]["scenes"]: si = sc["scene_index"] for sh in sc.get("shots", []): key = (si, sh["shot_index"]) ve_map[key] = sh.get("visible_entity_ids", []) if sh.get("variant_resolved"): vr_map[key] = sh["variant_resolved"] return ve_map, vr_map def _load_selected_map(self) -> Dict[int, set]: cp = self.runner._load_prev_checkpoint("shot_selection") selected: Dict[int, set] = {} if cp and cp.get("data", {}).get("scenes"): for sc in cp["data"]["scenes"]: selected[sc["scene_index"]] = set(sc.get("selected_shot_indices", [])) return selected def _load_shot_scenes_map( self, selected_map: Dict[int, set], ) -> Dict[int, List[Dict[str, Any]]]: cp = self.runner._load_prev_checkpoint("shot_validator") shot_scenes_map: Dict[int, List[Dict[str, Any]]] = {} if not (cp and cp.get("data", {}).get("scenes")): return shot_scenes_map from app.core.steps.shot_validator_step import assert_no_failed_scenes assert_no_failed_scenes( cp, getattr(self.runner, "project_config", None), consumer_step="scene_context_loader", ) for sc in cp["data"]["scenes"]: si = sc["scene_index"] sel_indices = selected_map.get(si) if sel_indices is not None: shot_scenes_map[si] = [ sh for sh in sc.get("shots", []) if sh.get("shot_index") in sel_indices ] else: shot_scenes_map[si] = sc.get("shots", []) return shot_scenes_map def _load_beats_by_scene(self) -> Dict[int, Dict[int, Dict[str, Any]]]: cp = self.runner._load_prev_checkpoint("beat_extract") beats_by_scene: Dict[int, Dict[int, Dict[str, Any]]] = {} if cp and cp.get("data", {}).get("scenes"): for sc in cp["data"]["scenes"]: si = sc.get("scene_index") beats_by_scene[si] = { b["beat_index"]: b for b in sc.get("beats", []) if b.get("beat_index") } return beats_by_scene def _load_summaries(self) -> Dict[int, str]: cp = self.runner._load_prev_checkpoint("scene_summary") if not cp: return {} return { s["scene_index"]: s.get("scene_summary", "") for s in cp.get("data", {}).get("summaries", []) } def _load_entity_canon_character_maps( self, ) -> Tuple[Dict[str, str], Dict[str, List[str]]]: """G4.6 Phase 4 Codex iter 1 B1 — main thread prebuild. ThreadPool worker 안에서 self.db query 가 SQLAlchemy session thread- unsafe race 를 만든다. _build_entity_traits_block 와 validate_visible_entities_contract 가 db 대신 본 map 사용. project_id scoped + entity_type='character' filter — Claude iter 1 I2 가 검출한 defensive consistency carry. C##/L##/P## prefix 가 corrupt 되어도 character 만 통과. Claude iter 2 I1 carry — silent fallback 제거 (`feedback_no_silent_fallback.md`): - 정상 결과 (character 0 row) 인 project 는 빈 map 허용 (downstream `_build_entity_traits_block` 가 visible_entities 안 character base 존재 시 fail-fast 로 잡음). - DB query / load 실패는 AppError fail-fast — `({}, {})` silent degrade 절대 금지. Phase 4 visible_entities contract 가 prebuild map 의존이라 silent empty 는 contract 무력화 (특히 character 미포함 shot 에선 downstream 도 못 잡음). Returns: (name_by_short_id, traits_by_short_id) — short_id → name / stable_traits (parsed list). character row 0 일 시 둘 다 빈 dict. Raises: AppError: DB query / load exception (connection / session / schema 오류 등). 정상 빈 결과는 raise 0. """ from app.core.entity_protection import _parse_traits from app.core.errors import AppError from app.models.project import EntityCanon try: rows = self.runner.db.query( EntityCanon.short_id, EntityCanon.name, EntityCanon.stable_traits, ).filter( EntityCanon.project_id == self.runner.project_id, EntityCanon.entity_type == "character", ).all() except Exception as exc: # Claude iter 2 I1 — DB exception 은 silent fallback 금지. # 정상 0 row 와 load 실패 분리: rows 가 [] 인 정상 case 만 빈 map # 허용 (try block 외 정상 path). raise AppError( code="step.context_loader.entity_canon_load_failed", message=( f"EntityCanon character maps are required for G4.6 " f"visible_entities contract — DB query failed for " f"project_id={self.runner.project_id}: {exc!r}. Fix: " f"ensure DB connection is healthy and entity_canon table " f"is accessible." ), status_code=500, ) from exc name_by_short_id: Dict[str, str] = {} traits_by_short_id: Dict[str, List[str]] = {} for r in rows: sid = getattr(r, "short_id", None) if not isinstance(sid, str) or not sid.startswith("C"): continue name_by_short_id[sid] = r.name or "" traits_by_short_id[sid] = _parse_traits(r.stable_traits) return name_by_short_id, traits_by_short_id def _load_entity_canon_prop_term_map(self) -> Dict[str, frozenset]: """Task 14 Wave 2 — main thread prebuild of prop term map. ``reconcile_owned_prop_namespace_overlap`` 의 Gate C (term-match) 입력. ThreadPool worker 진입 전 (`load_all` 의 본 호출 시점) main thread 에서 project_id + entity_type='prop' + status='active' scoped EntityCanon + EntityAlias rows 한 번 query → short_id → frozenset of normalized term strings. term source (모두 normalize 후 frozenset 으로 합집합): - entity_canon.name - entity_canon.description - entity_canon.t2i_prompt - entity_alias.alias (모든 alias row) 정규화 = ``_normalize_prop_term`` (strip + lowercase + collapse internal whitespace). 빈 / whitespace-only 결과는 제외. helper consumer 가 owned_token 도 같은 normalize 함수로 정규화 — producer / consumer symmetric. Wave 2 fixup (Task 14 — owned-prop namespace overlap fix): producer expands each ASCII source via ``_expand_prop_term_variants`` into full normalized phrase + ASCII content tokens (word-boundary exact, minus a fixed stopword set). Korean / non-ASCII sources yield only the full normalized phrase. This lets the consumer helper match owned single-word tokens (e.g. ``"map"``) against canon prop sources that only contain longer phrases (e.g. ``"a folded printed paper map showing detailed coastlines"``), without forcing the helper to do sub-word splitting. Returns: prop_term_map — ``{short_id: frozenset[str]}``. prop row 가 없으면 빈 dict. helper 가 빈 dict / None 입력에 short-circuit pass-through 이므로 caller 는 별도 분기 불필요. Raises: AppError(step.context_loader.prop_canon_load_failed): DB query failure (connection / session / schema). 정상 0 row 는 raise 아님 — character maps loader 와 동일 패턴. """ from app.core.errors import AppError from app.core.steps._owned_helpers import _expand_prop_term_variants from app.models.project import EntityAlias, EntityCanon try: prop_rows = self.runner.db.query( EntityCanon.id, EntityCanon.short_id, EntityCanon.name, EntityCanon.description, EntityCanon.t2i_prompt, ).filter( EntityCanon.project_id == self.runner.project_id, EntityCanon.entity_type == "prop", EntityCanon.status == "active", ).all() except Exception as exc: raise AppError( code="step.context_loader.prop_canon_load_failed", message=( f"EntityCanon prop maps are required for Task 14 owned-" f"prop namespace reconciliation — DB query failed for " f"project_id={self.runner.project_id}: {exc!r}. Fix: " f"ensure DB connection is healthy and entity_canon table " f"is accessible." ), status_code=500, ) from exc # Build canon_id → list of aliases (single query, in-memory join). canon_ids = [ r.id for r in prop_rows if isinstance(getattr(r, "id", None), str) ] aliases_by_canon: Dict[str, List[str]] = {} if canon_ids: try: alias_rows = self.runner.db.query( EntityAlias.canon_id, EntityAlias.alias, ).filter(EntityAlias.canon_id.in_(canon_ids)).all() except Exception as exc: raise AppError( code="step.context_loader.prop_canon_load_failed", message=( f"EntityAlias prop alias query failed for " f"project_id={self.runner.project_id}: {exc!r}." ), status_code=500, ) from exc for ar in alias_rows: cid = getattr(ar, "canon_id", None) alias = getattr(ar, "alias", None) if not isinstance(cid, str) or not isinstance(alias, str): continue aliases_by_canon.setdefault(cid, []).append(alias) prop_term_map: Dict[str, frozenset] = {} for r in prop_rows: sid = getattr(r, "short_id", None) if not isinstance(sid, str) or not sid.startswith("P"): # Defensive consistency — entity_type='prop' 가 P 접두여야 함. continue terms: set = set() for src in ( getattr(r, "name", None), getattr(r, "description", None), getattr(r, "t2i_prompt", None), ): if isinstance(src, str) and src.strip(): # Wave 2 fixup (Task 14) — producer expands each source # into full normalized phrase + ASCII content tokens # (minus stopwords). Korean / non-ASCII phrases yield # only the full phrase via the same util. terms |= _expand_prop_term_variants(src) for alias in aliases_by_canon.get(getattr(r, "id", ""), []): if isinstance(alias, str) and alias.strip(): terms |= _expand_prop_term_variants(alias) if terms: prop_term_map[sid] = frozenset(terms) if prop_term_map: logger.info( "scene_detail: %d prop term maps loaded for Task 14 " "namespace reconciliation", len(prop_term_map), ) return prop_term_map def _assert_no_visible_staging_drift(self, ctx: SceneAnalysisContext) -> None: """shot_director.visible vs shot_staging.camera_direction dual SOT drift 검사 — Path 1 structured = blocking, Path 2 proximity = diagnostic. Area #3 W3: function split. Path 1 (character_angles structured) 만 blocking (VisibleStagingDriftError). Path 2 (proximity NL fallback) 은 logger.warning only (no raise). mode= param 금지 (silent coupling 회피). scope: shot_director_ve 와 staging_map 양쪽이 모두 있는 shot 만. 한쪽이라도 없으면 skip (legacy / 부분 cp 호환). character entity 한정. """ from app.core.errors import VisibleStagingDriftError from app.modules.pipeline.shot_visibility import ( detect_offscreen_drift_structured, detect_offscreen_drift_proximity_diagnostic, ) if not ctx.shot_director_ve or not ctx.staging_map: return if not ctx.name_by_short_id: return for (si, shi), visible_ids in ctx.shot_director_ve.items(): staging = ctx.staging_map.get(f"{si}_{shi}") or {} cam = staging.get("camera_direction") or "" if not cam: continue char_visible = [sid for sid in visible_ids if sid.startswith("C")] if not char_visible: continue character_angles = staging.get("character_angles") or [] # Path 1 — structured + OFFSCREEN_RE hybrid gate → blocking drift_structured = detect_offscreen_drift_structured( char_visible, camera_direction=cam, character_angles=character_angles, id_to_name=ctx.name_by_short_id, ) if drift_structured: raise VisibleStagingDriftError( shot_label=f"S{si}_Shot{shi}", visible=list(visible_ids), camera_direction=cam, drift_entities=drift_structured, ) # Path 2 — proximity NL fallback → diagnostic only (no raise) drift_proximity = detect_offscreen_drift_proximity_diagnostic( char_visible, camera_direction=cam, id_to_name=ctx.name_by_short_id, character_angles=character_angles, ) if drift_proximity: logger.warning( "S%d_Shot%d Path 2 proximity NL drift candidate (diagnostic only): %s", si, shi, drift_proximity, ) def _load_episode_reference_policy(self) -> Optional[Dict[str, Any]]: """episode_reference_policy checkpoint → manifest dict (or None). 부재 시 None — build_render_prompt_card 가 None 을 no-op 처리. """ cp = self.runner._load_prev_checkpoint("episode_reference_policy") if not cp: return None return cp.get("data") def _load_world_rules(self): cp = self.runner._load_prev_checkpoint("visual_world_rules") return cp.get("data") if cp else None def _load_planning_context(self): from app.core.planning_doc_context import get_planning_context return get_planning_context( self.runner.project_id, self.runner.episode_id, self.runner.db, ) def _load_chain_bg_guide_by_shot(self) -> Dict[Tuple[int, int], str]: """background_render(Phase 7) / background_chain_render(Phase 5) manifest의 shot_guides를 (scene_index, shot_index) → guide str 매핑으로 펼침. 지원 shape (우선순위 순): - Phase 7: background_render — data.groups[bg_id].{status, shot_ids[], shot_guides[]} - Phase 5: background_chain_render — data.groups[group_id].{status, shot_ids[], shot_guides[]} - Phase 4 LEGACY: data.locations[loc_id].nodes[].{shot_ids[], shot_guides[]} shot_guides 항목 필드명: `guide` (Phase 5) 또는 `guide_text` (Phase 7) — 양쪽 정규화. Phase 7 우선, 비어있으면 Phase 5 fallback. 두 shape이 모두 있으면 Phase 7 우선 (중복 shot 시 Phase 7 보존). - manifest 미존재 → {} - shot_guides 필드 없는 legacy 노드/group → skip + warning - input shot_ids에 없는 extra shot_id → skip + warning - 중복 shot_id → 첫 번째만 - 'S{int}_Shot{int}' 형식 미준수 shot_id → skip + warning """ # Phase 7 (background_render) 먼저 시도. 미존재 또는 데이터 비어있으면 Phase 5 fallback. cp_p7 = self.runner._load_prev_checkpoint("background_render") cp_p5 = self.runner._load_prev_checkpoint("background_chain_render") result: Dict[Tuple[int, int], str] = {} data_p7 = (cp_p7 or {}).get("data") or {} data_p5 = (cp_p5 or {}).get("data") or {} if not data_p7 and not data_p5: return result sid_re = re.compile(r"^S(\d+)_Shot(\d+)$") warnings = 0 legacy_nodes_missing = 0 # Phase 4 v1 prompt 노드 (shot_guides 미생성) phase5_groups_missing = 0 # Phase 5 group이지만 shot_guides 필드 없음 (예외적 — render 직접 호출 등) def _absorb(container_id: str, container_kind: str, shot_ids: List[str], guides: Any) -> None: nonlocal warnings if not isinstance(guides, list): # 잘못된 shape (예: dict, str, None이 아닌 다른 타입) — 조용히 drop 대신 warning. if guides is not None: logger.warning( "_load_chain_bg_guide_by_shot: %s %s shot_guides is %s, expected list — skip", container_kind, container_id, type(guides).__name__, ) warnings += 1 return input_shot_ids = set(shot_ids or []) for sg in guides: # 비-dict 항목 (str, int, None 등) — get() 호출 시 AttributeError 방지. if not isinstance(sg, dict): warnings += 1 continue sid = sg.get("shot_id") or "" # Phase 7 schema는 'guide_text', Phase 5 schema는 'guide' — 양쪽 허용. guide = sg.get("guide") or sg.get("guide_text") or "" if not sid or not guide: warnings += 1 continue if sid not in input_shot_ids: logger.warning( "_load_chain_bg_guide_by_shot: extra shot_id %r not in %s %s shot_ids — skip", sid, container_kind, container_id, ) warnings += 1 continue m = sid_re.match(sid) if not m: logger.warning( "_load_chain_bg_guide_by_shot: invalid shot_id format %r — skip", sid, ) warnings += 1 continue key = (int(m.group(1)), int(m.group(2))) if key in result: # 중복 → 첫 번째만 보존 continue result[key] = guide # Phase 7 (background_render) 우선: data.groups[bg_id]. status='ok'만 처리. # 그 다음 Phase 5 (background_chain_render): 동일 shape data.groups, status='ok'. # 마지막 Phase 4 LEGACY (background_chain_render만): data.locations[loc_id].nodes[]. # _absorb는 중복 key 첫 번째만 보존하므로 자연스럽게 Phase 7 우선이 된다. for source_label, source_data in (("phase7", data_p7), ("phase5", data_p5)): groups = source_data.get("groups") or {} if isinstance(groups, dict): for group_id, gres in groups.items(): if not isinstance(gres, dict): continue if gres.get("status") != "ok": continue guides = gres.get("shot_guides") if guides is None: phase5_groups_missing += 1 continue _absorb( group_id, f"group({source_label})", gres.get("shot_ids") or [], guides, ) # Phase 4 LEGACY는 background_chain_render checkpoint(=data_p5)만 가질 수 있다. locations = data_p5.get("locations") or {} if isinstance(locations, dict): for loc_id, loc_data in locations.items(): if not isinstance(loc_data, dict): continue for node in loc_data.get("nodes", []) or []: if not isinstance(node, dict): continue guides = node.get("shot_guides") if guides is None: legacy_nodes_missing += 1 continue _absorb(node.get("id", loc_id), "node", node.get("shot_ids") or [], guides) if legacy_nodes_missing: logger.warning( "_load_chain_bg_guide_by_shot: %d Phase 4 LEGACY nodes had no shot_guides " "field (v1 prompt). chain_bg_guide prepend will be 0 for those.", legacy_nodes_missing, ) if phase5_groups_missing: logger.warning( "_load_chain_bg_guide_by_shot: %d Phase 5 groups had no shot_guides field " "(unexpected — Phase 5 render always emits shot_guides). chain_bg_guide " "prepend will be 0 for those.", phase5_groups_missing, ) if result: logger.info( "_load_chain_bg_guide_by_shot: %d shot guides loaded (%d skipped warnings)", len(result), warnings, ) return result def _load_chain_bg_owned_by_shot(self) -> Dict[Tuple[int, int], List[str]]: """G3.2: background_prompt cp 의 objects_owned_by_background 를 (scene_index, shot_index) → list[str] 매핑으로 펼친다. source = background_prompt cp (Spec 4.1 / round 3 #1 결정). shot 매핑 source (round 3 minor): - 우선: ``backgrounds[bid].spec.applies_to_shots`` (LLM 입력 그대로). - fallback: ``backgrounds[bid].shot_guides[].shot_id`` (LLM 출력). fail-fast (Spec 8.7 / round 4 BLOCKING 1 / round 5 IMPORTANT 2): - bg-on + cp 부재 → AppError fail-fast (silent {} 차단). - bg-on + cp schema<2 → AppError("contract_violation"). - bg-on + ok background entry 의 owned 부재 → AppError. 허용 path: - bg-off → {} (caller 진행). - bg-on + cp ok + 모든 ok bg 에 owned 1+ entries → 정상 매핑. Multi-bg 가 같은 shot 에 적용되면 owned 합집합, normalize 통과 후 sorted. """ from app.core.config import settings from app.core.steps._owned_helpers import ( assert_background_prompt_owned_contract, normalize_owned_list, ) bg_on = settings.background_mode in {"on", "floor_plan_anchored"} bp_cp = self.runner._load_prev_checkpoint("background_prompt") # fail-fast (옛 v4 / cp 부재 / partial v5 cp 차단). assert_background_prompt_owned_contract( bp_cp, background_mode_on=bg_on, where="scene_context_loader._load_chain_bg_owned_by_shot", ) if not bg_on or bp_cp is None: return {} sid_re = re.compile(r"^S(\d+)_Shot(\d+)$") result: Dict[Tuple[int, int], List[str]] = {} backgrounds = (bp_cp.get("data", {}) or {}).get("backgrounds", {}) or {} for bid, entry in backgrounds.items(): if not isinstance(entry, dict): continue if entry.get("status") != "ok": continue owned = entry.get("objects_owned_by_background") or [] if not owned: continue # shot 매핑: spec.applies_to_shots 우선, shot_guides[].shot_id fallback. shot_ids: List[str] = [] spec = entry.get("spec") or {} if isinstance(spec, dict): shot_ids = list(spec.get("applies_to_shots") or []) if not shot_ids: guides = entry.get("shot_guides") or [] shot_ids = [ sg.get("shot_id", "") for sg in guides if isinstance(sg, dict) ] for sid in shot_ids: m = sid_re.match(sid or "") if not m: logger.warning( "_load_chain_bg_owned_by_shot: invalid shot_id %r in bg %s — skip", sid, bid, ) continue key = (int(m.group(1)), int(m.group(2))) merged = normalize_owned_list(result.get(key, []) + owned) result[key] = merged if result: logger.info( "_load_chain_bg_owned_by_shot: %d shots mapped from %d backgrounds", len(result), len(backgrounds), ) return result def _load_chain_bg_id_by_shot(self) -> Dict[Tuple[int, int], str]: """G4.1 Wave 4 R4 B3: background_prompt cp 의 bg_id (bid) 를 (scene_index, shot_index) → bg_id 매핑으로 펼친다. card 의 background_binding.bg_id source. 매핑 source 는 _load_chain_bg_owned_by_shot 와 동일: - 우선: ``backgrounds[bid].spec.applies_to_shots`` (LLM 입력 그대로). - fallback: ``backgrounds[bid].shot_guides[].shot_id`` (LLM 출력). bg-off 는 {} 반환. bg-on + cp 부재 시 owned loader 가 이미 fail-fast raise 하므로 본 loader 는 bg_on + bp_cp None 케이스에서 안전한 {} 반환 (caller 가 owned 와 동시 호출 — owned 가 raise 한 후에는 본 loader 도달 불가). Multi-bg 가 같은 shot 에 적용되면 첫 번째 ok bid (alpha-sorted — deterministic) 만 매핑. 다중 bg 는 G4.x 후속 lift 에서 list 로 확장 가능. """ from app.core.config import settings from app.core.steps._owned_helpers import ( assert_background_prompt_owned_contract, ) bg_on = settings.background_mode in {"on", "floor_plan_anchored"} bp_cp = self.runner._load_prev_checkpoint("background_prompt") # owned loader 가 fail-fast 검증 담당 — 본 loader 는 동일 contract 검사 # (defense in depth) 후 bg-off / cp 부재 시 {}. assert_background_prompt_owned_contract( bp_cp, background_mode_on=bg_on, where="scene_context_loader._load_chain_bg_id_by_shot", ) if not bg_on or bp_cp is None: return {} sid_re = re.compile(r"^S(\d+)_Shot(\d+)$") result: Dict[Tuple[int, int], str] = {} backgrounds = (bp_cp.get("data", {}) or {}).get("backgrounds", {}) or {} # alpha-sorted bid iteration → multi-bg 충돌 시 deterministic 첫 번째 선택. for bid in sorted(backgrounds.keys()): entry = backgrounds[bid] if not isinstance(entry, dict): continue if entry.get("status") != "ok": continue shot_ids: List[str] = [] spec = entry.get("spec") or {} if isinstance(spec, dict): shot_ids = list(spec.get("applies_to_shots") or []) if not shot_ids: guides = entry.get("shot_guides") or [] shot_ids = [ sg.get("shot_id", "") for sg in guides if isinstance(sg, dict) ] for sid in shot_ids: m = sid_re.match(sid or "") if not m: continue # owned loader 가 이미 warning logged. key = (int(m.group(1)), int(m.group(2))) # 첫 매핑만 — alpha-sorted bid 보장 deterministic. if key not in result: result[key] = bid if result: logger.info( "_load_chain_bg_id_by_shot: %d shots mapped to bg_id from %d backgrounds", len(result), len(backgrounds), ) return result def _load_chain_bg_camera_meta_by_shot(self) -> Dict[Tuple[int, int], Dict[str, str]]: """background_render(Phase 7) / background_chain_render(Phase 5) manifest의 ``data.groups[bg_id].camera_recommendations`` (Phase 9.1에서 추가)를 ``shot_ids[]`` 와 join하여 (scene_index, shot_index) → meta dict 매핑. camera_recommendations dict는 {camera_position, camera_height, lens_hint, framing_notes} 4 필드를 포함하며, 빈 dict이면 매핑하지 않는다 (consumer에서 prepend 0). 지원 shape: - Phase 7: background_render — data.groups[bg_id]. status='ok'만. - Phase 5: background_chain_render — 동일 shape, fallback. Phase 4 LEGACY (data.locations[loc_id].nodes[])는 camera_recommendations 를 가질 수 없으므로 지원하지 않는다. 매핑 규칙은 ``_load_chain_bg_guide_by_shot`` 와 동일: - shot_id 'S{int}_Shot{int}' 정규식 미준수 → skip + warning - input shot_ids에 없는 extra → skip + warning - 중복 shot_id → 첫 번째만 - status != 'ok' group → skip """ cp_p7 = self.runner._load_prev_checkpoint("background_render") cp_p5 = self.runner._load_prev_checkpoint("background_chain_render") result: Dict[Tuple[int, int], Dict[str, str]] = {} data_p7 = (cp_p7 or {}).get("data") or {} data_p5 = (cp_p5 or {}).get("data") or {} if not data_p7 and not data_p5: return result sid_re = re.compile(r"^S(\d+)_Shot(\d+)$") warnings = 0 def _absorb(container_id: str, container_kind: str, shot_ids: List[str], meta: Dict[str, Any]) -> None: nonlocal warnings if not isinstance(meta, dict): return # 모두 빈 값이면 매핑 의미 없음 → skip if not any((meta.get(k) or "") for k in ( "camera_position", "camera_height", "lens_hint", "framing_notes" )): return normalized = { "camera_position": meta.get("camera_position", "") or "", "camera_height": meta.get("camera_height", "") or "", "lens_hint": meta.get("lens_hint", "") or "", "framing_notes": meta.get("framing_notes", "") or "", } for sid in shot_ids or []: if not isinstance(sid, str): warnings += 1 continue m = sid_re.match(sid) if not m: logger.warning( "_load_chain_bg_camera_meta_by_shot: invalid shot_id " "format %r in %s %s — skip", sid, container_kind, container_id, ) warnings += 1 continue key = (int(m.group(1)), int(m.group(2))) if key in result: # 중복 → 첫 번째만 (Phase 7 우선 자연스러움) continue result[key] = normalized for source_label, source_data in (("phase7", data_p7), ("phase5", data_p5)): groups = source_data.get("groups") or {} if not isinstance(groups, dict): continue for group_id, gres in groups.items(): if not isinstance(gres, dict): continue if gres.get("status") != "ok": continue cr = gres.get("camera_recommendations") if cr is None: continue _absorb( group_id, f"group({source_label})", gres.get("shot_ids") or [], cr, ) if result: logger.info( "_load_chain_bg_camera_meta_by_shot: %d shots with camera meta " "loaded (%d skipped warnings)", len(result), warnings, ) return result def _load_essence_by_shot(self) -> Dict[Tuple[int, int], List[str]]: """shot_essence_extraction (Phase 1b) 체크포인트의 data.shots[]를 (scene_index, shot_index) → essence list 매핑으로 펼침. - status='failed' shot은 빈 essence가 stub으로 들어있음 → skip - 빈 essence shot은 prepend 의미 없으므로 skip """ cp = self.runner._load_prev_checkpoint("shot_essence_extraction") result: Dict[Tuple[int, int], List[str]] = {} if not (cp and cp.get("data", {}).get("shots")): return result for sh in cp["data"]["shots"]: status = sh.get("status", "ok") essence = sh.get("essence") or [] si = sh.get("scene_index") shi = sh.get("shot_index") if status != "ok" or not essence or si is None or shi is None: continue result[(int(si), int(shi))] = list(essence) if result: logger.info( "_load_essence_by_shot: %d shots with essence loaded", len(result), ) return result