"""상세 Phase StepRunner -- scene_detail, scene_verify.""" import json import logging import re import time from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Set, Tuple from sqlalchemy.exc import SQLAlchemyError from app.core.dto import SceneAnalysisContext from app.core.errors import AppError from app.core.perception_mode import get_perception_guide from app.core.frame_spatial_contract import ( phrase_diagnostic as _fsc_phrase_diagnostic, validate_echoes as _fsc_validate_echoes, ) from app.core.framing_scale import ( FRAMING_CLOSE, get_framing_scale_or_raise, ) from app.core.step_runner import StepRunner from app.core.subject_state import is_immobilized_state from app.core.steps._evidence_helpers import ( _normalize_scene_detail_result, assert_fresh_llm_evidence, ) from app.core.steps.scene_context_loader import SceneContextLoader from app.modules.llm.llm_client import call_structured from app.modules.prompt_loader import load_prompt, load_schema logger = logging.getLogger(__name__) def _compute_forward_zoom_targets( dependencies: Optional[List[Dict[str, Any]]], shot_scenes_map: Dict[int, List[Dict[str, Any]]], scene_index: int, shot_index: Any, ) -> List[Dict[str, Any]]: """현재 shot이 후속 shot의 zoom_in_detail source인지 reverse-lookup. 결함 A fix (forward-look continuity). 같은 씬 내 후속 shot이 location_refs[0] 로 (scene_index, shot_index)를 가리키고 ref_usage='zoom_in_detail' 이면, 이 shot 의 frame에 후속 close-up 대상 영역이 visible 위치로 잡혀야 한다 (특히 죽은/ motion-frozen 인물 케이스). schema 키는 location_refs[].scene_index / shot_index (접두사 없음). shot_dependency_t2i 5.202604201700 / shot_dependency_t2i_step 양쪽 일관. 일반화 메커니즘 — 시나리오 의존 0. dependencies/shot_scenes_map 만 입력. Args: dependencies: ctx.dependencies 리스트. None / [] 모두 안전. shot_scenes_map: scene_index → [shot_meta, ...]. scene_index: 현재 shot의 scene_index. shot_index: 현재 shot의 shot_index (int 또는 str). Returns: 후속 shot 메타 dict 리스트: [{"scene_index", "shot_index", "description" (전문, no truncation), "keep_elements": list[{label, kind}]}, ...] keep_elements entry shape: {label: str, kind: enum["environment", "static_prop"]}. Area D-next-min (2026-05-15) — shot_dependency_t2i v7 producer SOT. character / person / body 묘사는 keep_elements 가 다루지 않음 (별도 layer: scene_consistency / character_state_variant / semantic_contract_router). """ out: List[Dict[str, Any]] = [] for d in dependencies or []: for ref in (d.get("location_refs") or [])[:1]: # Defensive: legacy fixtures 가 location_refs 를 int list 로 # 갖는 경우 skip (dict shape 가 contract — 옛 cp 는 무시). if not isinstance(ref, dict): continue if ref.get("ref_usage") != "zoom_in_detail": continue if ref.get("scene_index") != scene_index: continue if ref.get("shot_index") != shot_index: continue followup_si = d.get("scene_index") followup_shi = d.get("shot_index") followup_desc = "" for sh_meta in shot_scenes_map.get(followup_si, []) or []: if sh_meta.get("shot_index") == followup_shi: followup_desc = ( sh_meta.get("shot_description") or sh_meta.get("description") or "" ) break out.append({ "scene_index": followup_si, "shot_index": followup_shi, "description": followup_desc, "keep_elements": ref.get("keep_elements") or [], }) return out # Phase 9.2 + v12 (2026-05-02): scene_detail prompt v12 — Rule E/F/G/H + 신규 # Rule J Primary Subject Framing (한 shot = 한 framing scale). 모델은 gemini-pro. # SCHEMA_VERSION bump → step_runner가 stale cp를 자동 감지 (force 권장). SCENE_DETAIL_SCHEMA_VERSION = 13 # 2026-05-23 (reference-necessity Phase 2): render_prompt_card 가 episode_reference_policy overlay 반영 — text_only subject 의 id_policy/asset_requirements 변동 + card hash semantics 변경 → 구 cp invalidation. 이전: 12 (C2 v1 owned_object_usage). # Phase3 W3 (2026-05-23): generic_descriptor_allowed no-ID rule promoted above Rule X-2; Rule X-2 made policy-conditional (per-subject `id_policy.subject_reference_policy[]` SOT). cp config_hash invalidation via prompt text change only; SCHEMA_VERSION 13 유지 (output shape 동일). SCENE_DETAIL_PROMPT_VERSION = "47.202608250422" # 2026-08-25 — v47: 촬영 용어·실수치 누출의 진짜 출처를 막는다. 실측(Opik span 전수) — R1 적중 12건 중 **9건이 카드 `render_strategy.camera_direction` 에 이미 있던 낱말**이고, R3 는 **2/2 가 카드 `background_binding.camera_reference` 가 준 값**이었다. 즉 「모델이 규칙을 어겼다」가 아니라 **「모델이 다른 규칙을 지켰다」**. 두 규칙이 정면 충돌하고 있었다 — `## 구조화 입력 우선순위` 는 camera_direction 을 「최우선 반영·임의 해석 금지」라 하고, 금지 절은 **enum(framing_scale) 만** 다뤄 camera_direction 이 산문이라는 사실이 어디에도 없었다. 수리 = ①「무엇을 반영하나」와 「어떤 말로 쓰나」를 가른다 — 반영할 것(자리·거리와 잘림·화면 배치·시각 관계 네 갈래)은 **하나도 빠짐없이 남기고**, **프레임 크기를 등급 이름으로 부르는 말만** 보이는 것으로 바꾼다 ②`camera_reference` 수치 축 추가(맞추는 것과 수치를 옮겨 적는 것은 다르다) ③금지어 **나열을 제거**하고 판별법으로 대체 — `wide view` 가 프롬프트 전체에서 그 금지문 한 줄에만 있었고 카드에 없는데도 출력에 샜다(나열이 곧 프라이밍) ④`camera_effect` 는 **배출구가 아님**을 명시(그 값은 `theme_label` 메타데이터로만 가고 이미지 프롬프트에 안 합쳐진다 — `scene_generation_coordinator.py:1982,2863`). 28,674 → 30,988자. ★A/B 실측(ABBA·source-positive 4컷×3회씩): 등급낱말 **8/12 → 1/12**, 네 샷 전부 개선. 안전 축 넷 손실 없음 — 자리 12→12 · 화면 배치 12→12 · 시각 관계 5→5 · **거리·잘림 3→6(증가)**. source-negative 는 양쪽 0/12. ★남은 것 = `detail_steps.py:2828-2843` 이 같은 cam_dir 을 뒤쪽 「반드시 t2i_prompt에 반영하세요」 블록에 다시 넣는다 — 프롬프트로는 못 이기는 자리(별건). 이전: 46.202608250329 # 2026-08-25 — v46: ID 규칙이 다섯 절에 흩어져 **규칙 하나를 완성하려면 380줄을 오가야 하던 것**(비대 원인 ②) + 정책 분기마다 예시를 세 벌씩 쓰던 것(③). `## ID Policy`(71) · `## 엔티티 참조`(453) · `## 인물 복장 묘사`(464) · `## visible_entities 엄격 규칙`(511) · `## 인물 ID 사용`(524) 다섯을 **한 절 6단계**로 모았다 — ①누가 프레임에 있나 ②어떤 형식으로 적나(policy 3행 대응표) ③옷을 어떻게 붙이나 ④한 벌 예시 ⑤카드가 정하지 않는 다섯 ⑥재현 표면. 7,190 → 4,960자, 전체 30,909 → 28,675자(-2,234, -7.2%). ★③은 **금지형이 아니라 대응표로** 풀었다 — 예시 세 벌(11줄)이 문장 골격은 같고 ID 형식만 달랐다. 그래서 policy 3행 대응표를 주고 예시는 한 벌만 두고 「ID 부분만 갈아 끼운다」로 적었다. ★자리는 71 을 지켰다 — 「배열이 비어 있는 것이 정상」 상기는 E2E 실패로 들어온 것(v41)이라 프롬프트 뒤로 밀지 않는다. ★`## ID Policy` 앞머리도 지켰다 — `## Frame Spatial Contract` 가 부른다(다른 넷은 밖에서 부르는 곳이 없음을 확인). ★재료 유실 없음 — 23개 조항 전수 확인(v39 재현표면·v40 놓인 자세·v41 기본값/비인간·의류 아웃룩 한정·씬 내 일관성·validator 분기 등). 카드 중복이던 policy 3종 서술(카드 id_policy.constraints[2] 와 같은 말)은 대응표로 대체. 이전: 45.202608250305 # 2026-08-25 — v45: 카드가 값과 사용법을 **둘 다** 담은 산문 제거 (프롬프트 비대 원인 ①). 36,761 → 30,909자(-5,852자, -15.9%). 걷은 다섯 절 = `Read RenderPromptCard first` 1,963→464 · `ID Policy` 1,320→1,016 · `Continuity` 2,193→447 · `Background Binding` 1,145→445 · `Spatial consistency` 2,052→449. 근거 ①프롬프트 자신이 15줄에 "본문 prose 는 lift 완료 후 단계적으로 제거 예정" 명시 ②`_card_metadata.lift_status` 의 rule_a/c/h·id_policy_*3·continuity_*3·spatial_*3 전부 True ③**나간 payload 실물 대조** — Opik span 의 user 메시지에서 카드 JSON 을 꺼내 절마다 한 줄씩 맞춰 봤다(빌더 코드가 아니라 실제 주입분). ★남긴 기준 = v38→v39 재발 방지: 카드에 값은 있는데 「그럼 뭐라고 쓰나」가 본문에만 있으면 남긴다. 그래서 v41(정책배열 빈 것이 정상 — E2E 실패로 들어온 상기)·v39(재현 표면 내용 묘사)·v40(놓인 자세 / bg_id 산문 노출)·v43(framing_scale 은 읽는 값)은 그대로 뒀다. 걷은 것은 카드 constraints 가 값과 사용법을 함께 담은 것(ref_usage 3종·view_consistency·mode 4갈래·spatial 3규칙)과 모델이 못 읽는 구현 세부(`render_prompt_card_hash`·`_card_metadata` strip 시점·`backend/app/core/perception_mode.py` 파일 경로 — perception_mode 는 카드가 스스로 constraint 를 붙인다). 절 헤딩 32개는 하나도 안 바꿨다 — 다른 절이 `위 ## Continuity` 식으로 가리키고 있어 이름을 바꾸면 참조가 끊긴다(비대 원인 ②는 별건). 이전: 44.202608250248 # 2026-08-25 — v44: 광원 상태를 「원문이 말한 그 정도」로 옮기는 대응표. v43 은 금지형만 줬는데 그대로 안 통했다 — `switched off` 가 사라진 자리에 `goes out`·`blackout`(정전)이 들어왔다. 원문은 "형광등이 한 번 깜빡인다" 뿐이다. 금지만 있고 「그럼 뭐라고 쓰나」 재료가 없으면 모델은 자기가 아는 말로 간다 (v38→v39 에서 이미 겪은 것). 그래서 켜짐/깜빡임/꺼짐/무언급 4행 대응표를 준다. ★같은 판에서 대조가 됐다 — 대응표를 준 framing_scale 은 1→0 으로 잡혔고, 금지만 준 광원은 안 잡혔다. 이전: 43.202608250202 # 2026-08-25 — v43: framing_scale 낱말이 t2i_prompt 로 새던 것. `Spatial consistency`(1,873자)가 `close/medium/wide` 를 판정 기준으로 길게 쓰고, 384줄 뒤의 금지 절(265자)이 그 낱말을 출력에 쓰지 말라 했는데 **둘이 입력 enum 과 출력 문장이라는 구분이 어디에도 없었다**. 실측(최소 검증판 3씬): t2i_prompt 6개 중 1개에 `wide view`. 금지를 더 세게 쓰는 대신 ①enum→문장 대응표를 재료로 주고 ②`camera_effect` 는 촬영 용어를 써도 되는 자리임을 명시하고 ③두 절을 상호 참조로 이었다. 이전: 2026-08-12 (차렷/증명사진 분석 대응) — v42: ①시선·카메라 인지 기본값(staging 침묵 시에만 — 시선 대상을 장면 안 실재로 정해 적기, 렌즈 도열 구도 배제, 명시 입력이 있으면 그것이 권위) ②identity 참조 역할 한정(얼굴·복장 동일성 전용, 포즈·시선·구도 복사 금지 문장 강제). 선정 계약 3축(DIRECTION·BUILT SPACE·ENTITIES)에 카메라 인지 축이 없고 조립에도 candid 절이 없어 증명사진풍 후보가 걸러지지 않던 구멍의 저작측 재료. 이전: 2026-08-04 (E2E 실패 대응) — v41: 정책 배열이 비었을 때의 기본값을 본문이 다시 말한다. 금월도 E2E 에서 visible_entities=['C28'](검은 염소) 단독 샷이 base_id_missing 으로 두 번(primary+retry) 반려돼 255샷 중 1샷 때문에 스텝 전체가 failed 했다. 같은 시나리오의 v37 실행은 통과했고 그때도 shot 노드 237개 중 93%가 정책배열이 비어 있었다 — 즉 미명시는 정상이고 갈린 것은 프롬프트다. v38 이 를 카드 포인터로 바꾸면서 '명시 안 된 subject 는 기본값 id_and_outlook_required' 라는 상기가 사라졌다. 카드에 그 문구가 있기는 하나 긴 constraints 문자열 끝에 묻혀 있고 배열 자체가 비어 있어 '적용할 정책 없음'으로 읽힌다. 사람이 아닌 subject 도 같다는 것을 함께 적었다(동물 단독 샷이 취약점). 이전: 2026-08-04 (육안 판정 대응) — v40: 통합 A/B 육안 판정에서 확인된 결함 3건에 재료를 준다. ①프레임 점유율은 인체 규모 이상 대상에만 — 손에 들리거나 표면에 놓이는 물체는 관계 묘사로(before/after 둘 다 사진을 프레임 1/3 으로 그렸다. 해당 두 섹션은 v37↔v39 바이트 동일이라 판 무관 결함) ②재현 표면이 놓여 있으면 무엇 위에 어떤 자세로 놓였는지 함께 적는다(같은 카드 idx6·9 에서 lies flat/표면 명시가 네 조합 모두 나왔다 — 조항이 자세를 강제하지 않았다) ③bg_id 는 참조 식별자이므로 공간·조명 산문에는 일반 명사로(`fills the L18B01 space` → 이미지 모델이 임의의 큰 공간으로 해석). 셋 다 금지가 아니라 대안을 함께 준다. 이전: 2026-08-04 (prompt diet ③-fix) — v39: v38 에서 재현 표면 처리 조항을 되살렸다(224자). 육안 A/B 에서 reproduction_surface_rule.applies=true 인 두 shot(s28sh3·s29sh5)이 퇴행 — 사진 속 인물이 사라지고(한국식 기와집+두 소녀 → 인물 없는 서양식 "RURAL CAFE") 사진만 프레임 1/3 으로 비대해졌다. 카드에 값은 다 있었지만 표현이 전부 금지형("C##O## 쓰지 마라, 대신 generic descriptor")이라 "무엇을 그릴지" 재료가 없었다. 값과 그 값을 쓰는 법은 다른 것이다. v37 대비 5,915자(-14.9%) 절감은 유지. 이전: 38.202608032308 (prompt diet ③) — v38 prompt: RenderPromptCard 가 결정론적으로 확정하는 사안의 산문 중복 제거. ①`## ID Policy` 3,601자 → 카드 `id_policy.constraints`(policy enum 3종 처리·default·reproduction surface·demographic format 전부 포함) 를 가리키고 perception_mode 절만 남김 ②`## Rule X-2` 4,309자 → 카드가 주지 않는 5개(descriptor 는 ID 대체 불가 / 같은 ID 반복 금지 / forward 면제 조건 / 미등록 background figure / entity_canon.name 동반 표기)와 validator 동작만 남김. 39,812 → 33,673자(-15.4%). 근거 = `_card_metadata.lift_status` 의 id_policy_*_lifted 전부 True + system.md 자체가 "본문 prose 는 lift 완료 후 단계적으로 제거 예정" 명시. 조건부인 `rule_e_lifted` 와 복장·실루엣·Continuity 고유 규칙은 손대지 않음. 이전: 37.202607081535 (발명 경계 wave3) — v37 prompt: ①세계 사실 근거 경계 절대 규칙(연출=재량 / 날씨·표면·재질·광원 색·시간대=씬 원문·staging 입력·제작자 정정 근거시만, 무근거=침묵, 전역 요약=톤 참고, 배경 재질=background 입력만) ②구조화 입력 우선순위에 CREATOR CORRECTIONS 최우선 명문. 이전: 36.202607021930 (육안 6결함 wave) — v36 prompt: ①의류 명사는 배정 아웃룩(O##) 정의 텍스트에 있는 의류 종류만(창작 금지)+같은 씬 같은 C##O## 는 동일 의류 표현(샷별 드리프트 금지 — S1 티셔츠→재킷 류) ②이동체 방향 일관성 규칙(facing/heading/screen-direction 단일화 — 기하 모순이 하이브리드 형태 유발). 이전: 35.202605281000(미bump 부채 — constant≠latest 기존실패의 실체), 34.202605230753 (reference-necessity Phase 3 W3). # 주의: SCENE_DETAIL_PROMPT_VERSION은 cp config_hash 계산용. 실제 prompt는 def _write_grounding_sidecar(runner, card, visible_entities) -> int: """이 컷의 고증 참조 sidecar 를 카드에 적는다. ★없으면 **무동작**. ★좌표로 적고 bytes 는 안 담는다 — CP 는 JSON 이다. 읽는 것은 조립(`attach_from_rpc`)이 하고, 그때 해시를 확인한다. ★중앙 조사 CP 가 없거나 이 컷에 해당 멤버가 없으면 `0` 을 내고 카드를 **한 글자도 안 건드린다** — 그것이 비회귀의 문이다. """ from app.core.grounding_mode import (resolve_grounding_mode, uses_chunk_producer) from app.modules.pipeline import grounding_reference_bundle as rb from app.modules.pipeline import grounding_sidecar_writer as sw # ★★★넓은 `except Exception` 을 걷었다 (Codex BLOCK 2026-09-02). # 켠 판에서는 읽기 실패가 **기록 손실**이다 — 삼키면 정책도 sidecar 도 # 같이 사라져 참조 없이 그림까지 간다. 옛 판에서만 「없으면 무동작」이다. required = uses_chunk_producer( resolve_grounding_mode(getattr(runner, "project_config", None) or {})) # ★★★HITL 0 (사용자 2026-09-03 · 최상위 불변식): production 은 중앙 CP 를 # **날것으로** 읽는다. 사람 판정 표는 canary·A/B·사후 평가 도구의 것이고 # 여기의 의존성·정지 조건·지문이 아니다. 앞 판은 판정을 얹어 읽어서 사람이 # 「맞다」를 안 누르면 아무것도 안 붙었다(검토 75건 뒤 usable 1/18). if required: cp = runner._load_prev_checkpoint("reference_acquisition") sw.assert_central_checkpoint(cp) else: try: cp = runner._load_prev_checkpoint("reference_acquisition") except Exception: # noqa: BLE001 return 0 if not cp: return 0 # ★★해시·좌표는 사람 판정·probe 와 **같은 helper** 로 (실측 2026-09-02: # `chosen.content_sha256` 칸을 읽었는데 CP 에 그 칸이 없어 7샷 전부 실패) _root = sw.default_reference_root() return sw.write_for_shot( card, cp, visible_entities, content_sha_of=lambda r: sw.row_content_sha256(r, root=_root), coordinate_of=lambda r: sw.row_file_coordinate(r, root=_root), ) # prompt_loader가 prompts/_base/scene_detail/ 디렉토리에서 latest version # (numeric reverse — 11 > 10 > 9...)을 자동 picking. 새 prompt 디렉토리 # 추가 시 이 상수도 같이 갱신해야 cp invalidation이 정확히 작동. def _build_phase2_prepend_blocks( si: int, shi: int, essence_by_shot: Dict[Tuple[int, int], List[str]], chain_bg_guide_by_shot: Dict[Tuple[int, int], str], chain_bg_camera_meta_by_shot: Optional[Dict[Tuple[int, int], Dict[str, str]]] = None, chain_bg_owned_by_shot: Optional[Dict[Tuple[int, int], List[str]]] = None, *, shot_essence_enabled: bool, chain_bg_guide_enabled: bool, chain_bg_camera_meta_enabled: bool = False, is_close_framing: bool = False, ) -> str: """Phase 1b essence + Phase 2 chain_bg_guide + Phase 9.1 camera_meta + G3.2 chain_bg owned objects prepend. 각 토글이 off면 해당 블록 없음. 모두 off/data 없음 → 빈 문자열 (회귀 0). 순서: essence → chain_bg_guide → chain_bg_camera_meta → chain_bg_owned. 사이에 빈 줄. Phase 9.1: chain_bg_camera_meta_by_shot이 None이면 빈 dict처럼 동작 (positional arg 호환 유지). Phase 9.2 (framing_scale enum SOT v1, 2026-05-15): is_close_framing bool 인자 — production caller 가 helper (`get_framing_scale_or_raise`) 로 staging. framing_scale enum 을 read 후 `== FRAMING_CLOSE` 결과 전달. close framing 이면 chain_bg.camera_meta block 을 prepend skip — Rule E (close framing shot 은 합성 단계에서 chain_bg ref 가 자동 skip) 과 LLM 입력 일관 유지. G3.2 (round 4 Q3=A): chain_bg_owned_by_shot 인자 추가 — owned objects block 은 close framing 시 skip + non-close 시 owned 1+ entries 면 inject. toggle 없음 (correctness guard 라 항상 적용). 동시에 close framing 시 chain_bg_guide 도 skip — chain_bg_guide / chain_bg_camera_meta / chain_bg_owned 3 종 모두 deterministic skip 일관 (round 2 #3, Codex 일관성 review). """ blocks: List[str] = [] if shot_essence_enabled: essence = essence_by_shot.get((si, shi)) or [] if essence: lines = "\n".join(f"- {e}" for e in essence) blocks.append( "[샷 핵심 시각 요소 — 반드시 t2i_prompt에 포함]\n" + lines ) # G3.2: chain_bg_guide 도 close framing 시 skip (round 2 #3 일관성). if chain_bg_guide_enabled and not is_close_framing: guide = chain_bg_guide_by_shot.get((si, shi)) or "" if guide: blocks.append( "[chain_bg reference에 이미 있음 — 다시 그리지 말 것]\n" + guide ) if chain_bg_camera_meta_enabled and not is_close_framing: meta = (chain_bg_camera_meta_by_shot or {}).get((si, shi)) or {} if meta and any(meta.get(k) for k in ( "camera_position", "camera_height", "lens_hint", "framing_notes" )): meta_lines = [] for k_label, k_field in ( ("position", "camera_position"), ("height", "camera_height"), ("lens", "lens_hint"), ("framing notes", "framing_notes"), ): v = meta.get(k_field) or "" if v: meta_lines.append(f" {k_label}: {v}") if meta_lines: blocks.append( "[chain_bg reference 카메라 정보 — t2i_prompt에서 일치시키거나 명시적 deviation 명시]\n" + "\n".join(meta_lines) ) # G3.2: chain_bg owned objects block — close framing 시 skip + non-close 만 # owned 1+ entries 면 inject. round 4 Q3=A: toggle 없음 (correctness guard). if not is_close_framing: owned = (chain_bg_owned_by_shot or {}).get((si, shi)) or [] if owned: owned_lines = "\n".join(f"- {o}" for o in owned) blocks.append( "[chain_bg에 이미 그려진 객체 — 새로 그리지 말 것 (Rule C contract)]\n" + owned_lines ) if not blocks: return "" return "\n\n".join(blocks) + "\n\n" def _derive_visible_for_shot(ctx, si: int, shi: Optional[int]) -> List[str]: """ctx 에서 shot 단위 visible_entities 도출 — `_analyze_one()` 와 동일 우선순위. shot_director 결과가 있으면 shot 단위 변형 확정 ID 사용 (variant resolved), 없으면 scene 전체 visible 사용. shot 부재 (legacy / scene-level path) 면 scene 전체. """ if shi is not None and (si, shi) in (ctx.shot_director_ve or {}): return list(ctx.shot_director_ve[(si, shi)]) return list(ctx.scene_visible.get(si, []) or []) def _derive_scene_outlooks_for_shot(ctx, si: int) -> Dict[str, list]: """`_analyze_one()` 의 scene_outlooks 빌딩 로직 mirror. cid → [(osid, oname), ...] 매핑 — 같은 씬의 outlook assignments 만 포함 (cross-scene 배정 차단). """ outlook_id_to_info: Dict[str, tuple] = {} if ctx.outlook_data: for ol in (ctx.outlook_data or {}).get("outlooks", []): osid = ol.get("outlook_id") or ol.get("short_id", "") oname = ol.get("name", "") entry = (osid, oname) if osid: outlook_id_to_info[osid] = entry if oname: outlook_id_to_info[oname] = entry scene_outlooks: Dict[str, list] = {} if ctx.outlook_data: for sa in (ctx.outlook_data or {}).get("scene_assignments", []): if sa.get("scene_index") != si: continue for a in sa.get("assignments", []): cid = a.get("character_id", "") osid_key = a.get("outlook_id") or a.get("outlook_name", "") if cid and osid_key and osid_key in outlook_id_to_info: entry = outlook_id_to_info[osid_key] if cid not in scene_outlooks: scene_outlooks[cid] = [] if entry not in scene_outlooks[cid]: scene_outlooks[cid].append(entry) return scene_outlooks # 2026-05-10 — Fix B: 24 shot deterministic ref-contract fail fix. # RPC.asset_requirements.required_refs 를 LLM 응답 후 t2i_variations 기준으로 # narrow 하기 위한 두 helper. _analyze_one 와 verify_completion 양쪽이 같은 # 함수 사용 → narrow된 stored hash 와 recompute hash 정합 보장. # Area #2 W5 (2026-05-17): legacy immobilized-state literal tuple 폐기 — # is_immobilized_state SOT helper 가 immobilized 판정 단일 path (Gate 1). _OUTLOOK_COMPOSITE_RE = re.compile(r'(C\d{2,3})(O\d{2,3})') def _compute_used_outlook_pairs( t2i_variations: Optional[List[Dict[str, Any]]], ) -> Set[Tuple[str, str]]: """variation 들이 실제 사용한 (character_id, outlook_id) set. outfit_assignments + t2i_prompt regex (C##O##) union — variation 별로 선택된 outlook 만 추출. RPC.required_refs 를 이 set 으로 narrow 하면 attached_meta 와 정합 가능. """ result: Set[Tuple[str, str]] = set() for var in (t2i_variations or []): if not isinstance(var, dict): continue for oa in (var.get("outfit_assignments") or []): if not isinstance(oa, dict): continue cid = oa.get("character_id") or "" oid = oa.get("outlook_id") or "" if cid and oid: result.add((cid, oid)) for m in _OUTLOOK_COMPOSITE_RE.finditer(var.get("t2i_prompt") or ""): result.add((m.group(1), m.group(2))) return result def _strip_invalid_sids(text: str, invalid_sids: Iterable[str]) -> str: """무효 closed-world SID 토큰을 text 에서 token-complete 하게 제거. invalid_sids = _check_prompts() pure_remove / bad_in_sfp 가 산출한 VE 위반 SID 집합 (C##/L##/P##/C##O## — _ve_pattern 형식의 closed-world ID). 각 SID 와 그 후행 소유격('s) 를 한 단위로 소거하되, 양쪽이 모두 공백이면 단일 공백으로, 한쪽이라도 비공백이면 공백 없이 잇는다 (제거 위치 국소 정규화만). 무효 SID 가 text 에 없으면 (subn substitution 0건) 원본 text 를 그대로 반환 — blind global trim/collapse 없음. 제거 자리에 semantic subject phrase 재삽입 없음 (글자 삭제 + 국소 공백 정규화만). """ sids = sorted({s for s in invalid_sids if s}, key=len, reverse=True) if not sids or not text: return text pattern = re.compile( r"(?P\s*)\b(?:" + "|".join(re.escape(s) for s in sids) + r")\b(?:'s)?(?P\s*)" ) def _repl(m: "re.Match[str]") -> str: return " " if (m.group("left") and m.group("right")) else "" cleaned, n = pattern.subn(_repl, text) return cleaned if n else text def _strip_overdeclared_prop_phrase_kind( t2i_variations: Optional[List[Dict[str, Any]]], render_prompt_card: Optional[Dict[str, Any]], ) -> None: """FINDING 9 W4 (Cat4) — reference_phrase_kinds 의 'prop' over-declaration 제거. scene_detail LLM 이 per-variation sidecar `reference_phrase_kinds` 에 'prop' 을 declare 했어도, 그 prop 이 numbered image mapping 에 매핑되지 않은 descriptive P## 언급이면 `render_prompt_card.asset_requirements.required_refs` 에 kind="prop" required_ref 가 생기지 않는다. required_refs 가 prop reference 의 SOT 이므로, prop required_ref 부재 시 모든 variation 의 `reference_phrase_kinds` 에서 'prop' 을 제거해 sidecar 를 SOT 와 정합시킨다. 이렇게 안 하면 ref_contract_validator step 6 phantom guard 가 image-gen 직전 fail-fast 한다 (e2e-bughunt-v1 FINDING 9). producer self-consistency normalization 일 뿐 validator step 6 약화가 아니다 — downstream phantom guard 의 fail-fast 는 그대로 보존된다. 'character' / 'background' 는 건드리지 않는다 (Cat2 / id_policy / background contract 와 얽혀 W4 scope 밖). prop required_ref 가 있으면 무변경. """ if not t2i_variations: return required_refs = ( (render_prompt_card or {}).get("asset_requirements") or {} ).get("required_refs") or [] card_has_prop_ref = any( isinstance(r, dict) and r.get("kind") == "prop" for r in required_refs ) if card_has_prop_ref: return for var in t2i_variations: if not isinstance(var, dict): continue rpk = var.get("reference_phrase_kinds") if isinstance(rpk, list) and "prop" in rpk: var["reference_phrase_kinds"] = [k for k in rpk if k != "prop"] # character ID-form 토큰 (C##, C##O##, O00). prompt 에 이 명시적 ID signal 이 # 있으면 'character' 를 over-declaration 으로 단정하지 않는다 — 명시적 ID 는 # silent-normalize 하지 않고 validator step 6 fail-fast 에 맡긴다 (FINDING 11). _CHARACTER_ID_FORM_RE = re.compile(r"\bC\d{2,3}(?:O\d{2,3})?\b|\bO00\b") def _strip_overdeclared_character_phrase_kind( t2i_variations: Optional[List[Dict[str, Any]]], render_prompt_card: Optional[Dict[str, Any]], ) -> None: """FINDING 11 — reference_phrase_kinds 의 'character' over-declaration 제거. FINDING 9 W4 `_strip_overdeclared_prop_phrase_kind` 의 'character' 변종. 인물이 창문 반사·실루엣·무-안면 등으로만 등장하면 id_policy 가 C## ID 대신 generic descriptor 사용을 지시 → render_prompt_card.asset_requirements. required_refs 에 character/character_outlook ref 가 생기지 않는다 (정상). 그런데 scene_detail LLM 이 per-variation sidecar `reference_phrase_kinds` 에 'character' 를 over-declare 하면 ref_contract_validator step 6 phantom guard 가 image-gen 직전 fail-fast 한다 (e2e-bughunt-v1 FINDING 11, still ec601991). strip 조건을 좁게 잡는다 — 'character' 는 Cat2 / id_policy 와 얽혀 있어 무조건 strip 하면 W2 base_id_required / O00 base / state_variant / legacy 같은 legit character meta 경로를 over-strip 할 수 있다. 다음 셋을 모두 만족할 때만 variation 의 `reference_phrase_kinds` 에서 'character' 제거: 1. variation 의 `reference_phrase_kinds` 에 'character' 가 있다. 2. required_refs 에 kind in {character, character_outlook} 가 없다. 3. 해당 variation 의 t2i_prompt 에 character ID-form signal (C##, C##O##, O00) 이 없다. required character ref 가 있거나 prompt 에 명시적 ID-form signal 이 있으면 무변경 — 명시적 ID 는 silent-normalize 하지 않고 validator step 6 fail-fast 에 맡긴다 (W2 base_id_required / O00 base / state_variant / legacy attach over-strip 방지). producer self-consistency normalization 일 뿐 validator step 6 약화가 아니다 — downstream phantom guard 의 fail-fast 는 그대로 보존된다. 'prop' / 'background' 는 건드리지 않는다 ('prop' 은 W4 helper 소관, 'background' 는 background_chain / prev_shot fallback 등 별도 attach 경로와 얽혀 scope 밖). """ if not t2i_variations: return required_refs = ( (render_prompt_card or {}).get("asset_requirements") or {} ).get("required_refs") or [] card_has_character_ref = any( isinstance(r, dict) and r.get("kind") in ("character", "character_outlook") for r in required_refs ) if card_has_character_ref: return for var in t2i_variations: if not isinstance(var, dict): continue rpk = var.get("reference_phrase_kinds") if not (isinstance(rpk, list) and "character" in rpk): continue if _CHARACTER_ID_FORM_RE.search(var.get("t2i_prompt") or ""): continue # 명시적 character ID-form signal — silent-normalize 금지 var["reference_phrase_kinds"] = [k for k in rpk if k != "character"] def _detect_state_variant_chars( staging: Optional[Dict[str, Any]], visible_chars: Set[str], ctx, ) -> Set[str]: """staging.character_angles 의 subject_state 가 immobilized (dead/unconscious/ severely_injured) 인 character 의 short_id (C##) set. scene_reference_service.detect_state_variant_sids 의 scene_detail-side mirror — scene_ref_image_map 없이 staging + visible 만으로 detect. resolver 가 그 character 를 character_state ref 로 attach 하므로 character_outlook required 에서 제외해야 validator 정합 (S12_Shot6 dead-character 결함 fix). Area #2 W5 (2026-05-17): legacy mixed gaze field literal 판정 폐기 — ``is_immobilized_state(ca["subject_state"])`` 단일 SOT helper (Gate 1). """ if not staging: return set() name_to_sid: Dict[str, str] = {} for e in (ctx.entities.get("characters", []) if hasattr(ctx, "entities") else []): sid = e.get("short_id") or "" name = e.get("name") or "" if sid and name: name_to_sid[name] = sid result: Set[str] = set() for ca in (staging.get("character_angles") or []): if not isinstance(ca, dict): continue state = ca["subject_state"] # required by v13 schema, KeyError = schema violation (Gate 4 fail-fast) if not is_immobilized_state(state): continue sid = name_to_sid.get(ca.get("character") or "") if sid and sid in visible_chars: result.add(sid) return result def _derive_outlook_pairs_for_shot( ctx, si: int, shi: Optional[int], ) -> List[Dict[str, str]]: """`_analyze_one()` 의 _g41_outlook_pairs 빌딩 로직 mirror. scene_outlooks (LLM assignments) + fixed_outfits (1개 자동 배정) 통합. 각 (cid, osid) 1회만. order: scene_outlooks 순서 + fixed_outfits 순서. Codex iter 1 B4 carry — shot-specific filter: 본 shot 의 visible_chars 에 속한 character_id 만 통과 (validator Source 3b 와 lockstep). 같은 scene 의 다른 shot 에만 등장하는 character 의 pair 가 본 shot 의 card 에 흘러가면 정상 shot 이 false-fail. mirror path 가 _analyze_one producer 와 일치해야 verify/_user_edited recompute 가 hash-stable. """ visible = _derive_visible_for_shot(ctx, si, shi) visible_chars = [sid for sid in visible if sid.startswith("C")] visible_chars_set = set(visible_chars) scene_outlooks = _derive_scene_outlooks_for_shot(ctx, si) fixed_outfits: Dict[str, str] = {} for cid in visible_chars: outlooks = scene_outlooks.get(cid, []) if not outlooks: # ★이 씬에 배정이 하나도 없으면 O00 — `_analyze_one` 과 같은 규칙 (2026-09-17) fixed_outfits[cid] = "O00" elif len(outlooks) == 1 and outlooks[0][0] == "O00": fixed_outfits[cid] = "O00" elif len(outlooks) == 1: fixed_outfits[cid] = outlooks[0][0] elif len(outlooks) >= 2: non_null = [(o, n) for o, n in outlooks if o != "O00"] if len(non_null) == 1: fixed_outfits[cid] = non_null[0][0] elif not non_null: fixed_outfits[cid] = "O00" pairs: List[Dict[str, str]] = [] seen: set = set() for cid, ols in scene_outlooks.items(): if cid not in visible_chars_set: continue # B4: shot-specific filter for osid, _ in ols: if osid and (cid, osid) not in seen: pairs.append({"character_id": cid, "outlook_id": osid}) seen.add((cid, osid)) for cid, osid in fixed_outfits.items(): if cid not in visible_chars_set: continue # B4: shot-specific filter (defensive — fixed_outfits # already restricted to visible_chars in loop above) if osid and (cid, osid) not in seen: pairs.append({"character_id": cid, "outlook_id": osid}) seen.add((cid, osid)) return pairs def _derive_fixed_elements_for_shot( ctx, si: int, shi: Optional[int], ) -> List[Dict[str, Any]]: """`_analyze_one()` 의 fixed_for_shot 빌딩 로직 mirror. scene_consistency 의 fixed_elements 중 applies_to_shots 가 본 shi 포함하는 것만. shi=None (scene-level) → 빈 list (scene-level 은 fixed_element 무관). """ if shi is None: return [] try: shi_int = int(shi) except (TypeError, ValueError): return [] out: List[Dict[str, Any]] = [] for fe in (ctx.fixed_elements_by_scene or {}).get(si, []) or []: raw = fe.get("applies_to_shots", []) or [] applies: set = set() for v in raw: try: applies.add(int(v)) except (TypeError, ValueError): continue if shi_int in applies: out.append(fe) return out def _derive_previous_shot_refs( ctx, si: int, shi: Optional[int], ) -> List[Dict[str, Any]]: """`_analyze_one()` 의 _g41_prev_refs 빌딩 로직 mirror. ctx.dependencies 에서 본 (si, shi) 의 location_refs[0] + character_refs 를 추출. shi=None 이면 빈 list. """ if shi is None: return [] dep_for_shot = next( (d for d in (ctx.dependencies or []) if d.get("scene_index") == si and d.get("shot_index") == shi), {}, ) dep_loc_refs = (dep_for_shot.get("location_refs") or [])[:1] dep_char_refs = dep_for_shot.get("character_refs") or [] out: List[Dict[str, Any]] = [] for r in dep_loc_refs: # Defensive: legacy int-shape ref skip (dict contract). if not isinstance(r, dict): continue out.append({ "scene_index": r.get("scene_index"), "shot_index": r.get("shot_index"), "ref_usage": r.get("ref_usage", ""), "kind": "location", }) for r in dep_char_refs: if not isinstance(r, dict): continue out.append({ "scene_index": r.get("scene_index"), "shot_index": r.get("shot_index"), "ref_usage": r.get("ref_usage", ""), "kind": "character", }) return out def _derive_card_inputs_from_ctx( *, ctx, seg: Optional[Dict[str, Any]], shot_info: Optional[Dict[str, Any]], ) -> Dict[str, Any]: """G4.1 Wave 4 R4 B1+B2+B3: ctx 단일 source 에서 card 의 7 input 모두 도출. `_analyze_one()` / `verify_completion()` / `_user_edited_card_contract_ violated()` (reuse path) 가 모두 본 helper 를 호출 → drift detection input 이 한 source. 하드코드 empty 인풋 (B1/B2 결함) 차단. 7 ctx-derived input source: - visible_entities: ctx.shot_director_ve / ctx.scene_visible - outlook_pairs: ctx.outlook_data + scene_outlooks + fixed_outfits - staging: ctx.staging_map[f"{si}_{shi}"] - bg_id: ctx.chain_bg_id_by_shot[(si, shi)] - bg_owned: ctx.chain_bg_owned_by_shot[(si, shi)] - bg_camera_meta: ctx.chain_bg_camera_meta_by_shot[(si, shi)] - bg_guide: ctx.chain_bg_guide_by_shot[(si, shi)] - fixed_elements: ctx.fixed_elements_by_scene[si] filtered by applies_to_shots - previous_shot_refs: ctx.dependencies (loc_refs[:1] + char_refs) - forward_zoom_targets: _compute_forward_zoom_targets(ctx.dependencies, ...) derived flags: - is_close_framing: staging.framing_scale == FRAMING_CLOSE (helper read) - background_mode_on: settings.background_mode in {on, floor_plan_anchored} - perception_mode: shot_info.perception_mode (passthrough) Returns dict with keys matching `build_render_prompt_card(...)` signature. `ctx` 본 dict 에 포함 (caller 가 strip — splat call 전). R1-I1 / B4 fail-fast 분기: - shot_info 가 dict 이고 staging 부재 → staging_not_applicable marker 설정 안 함 (builder 가 raise — shot path 의 staging 누락 = contract violation). - shot_info 가 None → legacy / scene-level path → marker dict synthesize 해서 builder 가 not_applicable mode 로 처리. """ from app.core.config import settings as _settings si = 0 if isinstance(seg, dict): si = int(seg.get("scene_index") or seg.get("index") or 0) shi: Optional[int] = None if isinstance(shot_info, dict): raw_shi = shot_info.get("shot_index") if raw_shi is not None: try: shi = int(raw_shi) except (TypeError, ValueError): shi = None perception_mode = None if isinstance(shot_info, dict): perception_mode = shot_info.get("perception_mode") visible_entities = _derive_visible_for_shot(ctx, si, shi) # Patch A — visible_entity_details: ctx.entities 에서 visible sid 로 lookup. # ★★갈래 표를 **여기서 다시 적지 않는다** (2026-09-01). 앞에는 # characters/locations/props 셋을 손으로 적어 둬서, 계약에 네 번째 # (`location_parts`)가 생겨도 이쪽은 몰랐다 — 같은 규칙이 두 곳이면 # 한쪽만 고쳐진다. `ctx.entities` 에 그 칸이 없으면 그대로 무동작이다. # shot_description / representative_moment / t2i_prompts 는 아래 shot_info. from app.modules.pipeline.grounding_carry import ENTITY_KEY_TO_OWNER _patch_a_entity_by_sid: Dict[str, Dict[str, Any]] = {} if ctx is not None and getattr(ctx, "entities", None): for _etype_key, _etype_val in ENTITY_KEY_TO_OWNER: for _e in (ctx.entities.get(_etype_key) or []): _sid = _e.get("short_id") if isinstance(_e, dict) else None if _sid: # Area B (2026-05-13): metadata_json (JSON string in # ctx.entities) → dict. consumer (build_render_contracts via # entity_metadata helper) 가 직접 dict access. parse 실패 / # missing 은 빈 dict — helper 가 fail-fast. _md_raw = _e.get("metadata_json") if isinstance(_md_raw, str): try: _md_dict = json.loads(_md_raw) except (json.JSONDecodeError, TypeError): _md_dict = {} elif isinstance(_md_raw, dict): _md_dict = _md_raw else: _md_dict = {} _patch_a_entity_by_sid[_sid] = { "short_id": _sid, "name": _e.get("name", "") or "", "entity_type": _etype_val, "t2i_prompt": _e.get("t2i_prompt", "") or "", "metadata_json": _md_dict, } visible_entity_details: List[Dict[str, Any]] = [ _patch_a_entity_by_sid[_s] for _s in (visible_entities or []) if _s in _patch_a_entity_by_sid ] outlook_pairs = _derive_outlook_pairs_for_shot(ctx, si, shi) staging = None if shi is not None: cand = (ctx.staging_map or {}).get(f"{si}_{shi}") if isinstance(cand, dict) and cand: staging = cand bg_id = None bg_owned: List[str] = [] bg_camera_meta = None bg_guide = None if shi is not None: bg_id = (ctx.chain_bg_id_by_shot or {}).get((si, shi)) # Codex Important 1 fix — preserve None vs explicit [] distinction. # If bg_id is present but chain_bg_owned_by_shot has no entry for # (si, shi), that is a loader contract violation (upstream did not # populate). raise instead of silently absorbing to []. The # `chain_bg_owned_by_shot or {}` outer wrapper is preserved because # ctx may legitimately have an empty/None map when the whole # chain_bg loader is disabled (background_mode off). _bg_owned_map = ctx.chain_bg_owned_by_shot or {} _bg_owned_raw = _bg_owned_map.get((si, shi)) if bg_id is not None and _bg_owned_raw is None: raise AppError( code="step.contract_violation", message=( f"chain_bg_owned_by_shot missing entry for " f"(scene={si}, shot={shi}) but bg_id={bg_id!r} present — " f"upstream loader did not populate. NO silent fallback. " f"Use explicit [] for intentionally empty (no owned objects)." ), ) bg_owned = list(_bg_owned_raw) if _bg_owned_raw is not None else [] bg_camera_meta = (ctx.chain_bg_camera_meta_by_shot or {}).get((si, shi)) bg_guide = (ctx.chain_bg_guide_by_shot or {}).get((si, shi)) background_mode_on = ( _settings.background_mode in {"on", "floor_plan_anchored"} ) # framing_scale enum SOT v1 (2026-05-15): shot path 만 helper read. legacy # / scene-level path (shot_info=None caller → staging=None, docstring R1-I1 # B4 명시) 는 is_close_framing 무관 → default False (Gate 4 의 "legacy # pass-through" 허용). shot path 에서 staging dict 인데 framing_scale 부재면 # helper 가 fail-fast (legacy cp; force re-run shot_staging step). is_close_framing = False if isinstance(staging, dict): is_close_framing = ( get_framing_scale_or_raise( staging, where=f"detail_steps._build_render_prompt_card_dict S{si}_Shot{shi}", ) == FRAMING_CLOSE ) fixed_elements = _derive_fixed_elements_for_shot(ctx, si, shi) previous_shot_refs = _derive_previous_shot_refs(ctx, si, shi) forward_zoom_targets = _compute_forward_zoom_targets( ctx.dependencies if ctx is not None else None, ctx.shot_scenes_map if ctx is not None else {}, si, shi, ) if ctx is not None else [] # B4: shot path (shot_info 가 dict + shi != None) 에서 staging 부재면 # builder 가 raise — staging_not_applicable marker 안 붙임. # legacy / scene-level path (shot_info 가 None) → marker synthesize. shot_info_for_card = shot_info if shot_info is None: shot_info_for_card = {"staging_not_applicable": True} # Area B (2026-05-13, Task 6): 옛 노운 매칭 필터 폐기 정합. # shot_description / representative_moment / t2i_prompts 모두 build_render_ # prompt_card 시그니처에서 제거됨. visible_entity_details (metadata_json 포함) # 만 render_contracts producer 진입. return { "scene_index": si, "shot_index": shi if shi is not None else 0, "seg": seg, "shot_info": shot_info_for_card, "visible_entities": visible_entities, "visible_entity_details": visible_entity_details, "outlook_pairs": outlook_pairs, "perception_mode": perception_mode, "staging": staging, "bg_id": bg_id, "bg_owned": bg_owned, "bg_camera_meta": bg_camera_meta, "bg_guide": bg_guide, "is_close_framing": is_close_framing, "background_mode_on": background_mode_on, "fixed_elements": fixed_elements, "previous_shot_refs": previous_shot_refs, "forward_zoom_targets": forward_zoom_targets, "name_by_short_id": getattr(ctx, "name_by_short_id", {}) or {}, "episode_reference_policy": getattr( ctx, "episode_reference_policy", None), "ctx": ctx, } def _collect_card_inputs( *, ctx, seg: Optional[Dict[str, Any]], shot_info: Optional[Dict[str, Any]], visible_entities=None, outlook_pairs=None, staging_for_shot=None, bg_id_for_shot=None, bg_owned_for_shot=None, bg_camera_meta_for_shot=None, bg_guide_for_shot=None, is_close_framing=None, background_mode_on=None, fixed_elements_for_shot=None, previous_shot_refs_for_shot=None, forward_zoom_targets_for_shot=None, name_by_short_id=None, ) -> Dict[str, Any]: """G4.1 Phase 5 Task 14 (R2-B3) + Wave 4 R4 (B1/B2/B3): RenderPromptCard builder 의 단일 input source. `_analyze_one()` (Task 14/15), `verify_completion()` (Task 18), 그리고 `_user_edited` reuse path (Task 19) 모두 이 helper 를 호출 → drift detection 의 input 이 일관됨. Wave 4 R4 redesign: ctx 가 제공되면 ``_derive_card_inputs_from_ctx`` 가 7 input 모두 ctx 에서 도출 (B1/B2 hardcode empty 결함 차단). 명시적 keyword 인자가 None 이 아닌 다른 값으로 전달되면 그 값으로 override (legacy callers / unit test 호환). Override semantics (N1 cleanup — explicit, ctx-derived path 기준): - kwarg = ``None`` → derived 값 유지 (override 안 함, sentinel) - kwarg = ``[]`` / ``{}`` / ``False`` / ``0`` → derived 위 override (빈/falsy 도 명시적 no-data 의도이므로 sentinel 아님) - kwarg = 비어있지 않은 값 → override 즉 ``None`` 만 sentinel — falsy 값은 모두 override 적용. bool flag 두 개 (``is_close_framing`` / ``background_mode_on``): - ctx-derived path: ``None`` = sentinel, 아니면 ``bool(...)`` 캐스팅 후 override. - legacy fallback path (ctx 미제공): ``None`` 도 ``bool(None) == False`` 로 즉시 default 셋팅 (sentinel 의미 X). 즉 fallback path 에서는 ``False`` 와 ``None`` 입력이 동일 결과 — 옛 구현과의 호환을 위한 의도적 동작 (review iter1 MINOR 명시화). Returns dict with keys matching `build_render_prompt_card(...)` signature + `ctx` (callers strip `ctx` before splat-call into builder). """ # ctx-driven derivation requires SceneAnalysisContext instance — opaque # test fixtures (e.g. plain `object()`) fall through to explicit kwargs path. _ctx_is_analysis_ctx = isinstance(ctx, SceneAnalysisContext) if _ctx_is_analysis_ctx: derived = _derive_card_inputs_from_ctx( ctx=ctx, seg=seg, shot_info=shot_info, ) else: # Legacy fallback — ctx 없으면 explicit 인자 그대로 forward. si = 0 if isinstance(seg, dict): si = int(seg.get("scene_index") or seg.get("index") or 0) shi = 0 if isinstance(shot_info, dict): try: shi = int(shot_info.get("shot_index") or 0) except (TypeError, ValueError): shi = 0 perception_mode = None if isinstance(shot_info, dict): perception_mode = shot_info.get("perception_mode") # Area B (2026-05-13, Task 6): 옛 노운 매칭 필터 폐기 정합. # shot_description / representative_moment / t2i_prompts 모두 제거 — # build_render_prompt_card 시그니처에서 사라짐. derived = { "scene_index": si, "shot_index": shi, "seg": seg, "shot_info": shot_info, "visible_entities": visible_entities, "visible_entity_details": [], # legacy path: builder default 흡수 (silent []) "outlook_pairs": outlook_pairs, "perception_mode": perception_mode, "staging": staging_for_shot, "bg_id": bg_id_for_shot, "bg_owned": bg_owned_for_shot, "bg_camera_meta": bg_camera_meta_for_shot, "bg_guide": bg_guide_for_shot, "is_close_framing": bool(is_close_framing), "background_mode_on": bool(background_mode_on), "fixed_elements": fixed_elements_for_shot, "previous_shot_refs": previous_shot_refs_for_shot, "forward_zoom_targets": forward_zoom_targets_for_shot, "name_by_short_id": {}, "episode_reference_policy": ( getattr(ctx, "episode_reference_policy", None) if ctx is not None else None ), "ctx": None, } # Explicit overrides for callers that have already computed values # (e.g. `_analyze_one()` after retry-time prompt strip + outfit cleanup). overrides = { "visible_entities": visible_entities, "outlook_pairs": outlook_pairs, "staging": staging_for_shot, "bg_id": bg_id_for_shot, "bg_owned": bg_owned_for_shot, "bg_camera_meta": bg_camera_meta_for_shot, "bg_guide": bg_guide_for_shot, "fixed_elements": fixed_elements_for_shot, "previous_shot_refs": previous_shot_refs_for_shot, "forward_zoom_targets": forward_zoom_targets_for_shot, "name_by_short_id": name_by_short_id, } for k, v in overrides.items(): if v is not None: derived[k] = v if is_close_framing is not None: derived["is_close_framing"] = bool(is_close_framing) if background_mode_on is not None: derived["background_mode_on"] = bool(background_mode_on) return derived def _recompute_card_hash_with_upstream_dependency( *, runner: Any, base_ctx: Any, si: int, shi: Optional[int], used_outlook_pairs: Any, state_variant_chars: Any, ) -> Optional[str]: """FINDING 8 (e2e-bughunt-v1) — render_prompt_card verify dual-source fallback. ``verify_completion()`` 의 card-hash recompute 는 ``SceneContextLoader. _load_dependencies()`` 결과를 dependency source 로 쓴다. 이 loader 는 downstream ``shot_dependency_t2i`` checkpoint 가 존재하면 그것을 우선한다. ``shot_dependency_t2i`` (order 21.71) 는 ``scene_detail`` (order 21.7) *이후* 에 실행되므로, scene_detail 이 카드를 저장한 뒤 ``shot_dependency_t2i`` 가 갱신/생성되면 이후의 verify recompute 가 카드 build 당시와 다른 dependency source 를 읽어 ``previous_shot_refs`` / ``forward_zoom_targets`` 가 달라지고 false ``card_drifted`` hash 가 발생한다. 본 helper 는 1차 recompute 가 mismatch 일 때만 caller 가 호출 — upstream ``shot_dependency`` checkpoint 의 ``data.dependencies`` 를 source 로 한 fallback recompute 를 정확히 1회 수행한다. fallback 은 false drift 를 **구제만** 한다: upstream cp 부재 / 1차와 동일 source / recompute 실패 시 ``None`` 을 반환해 caller 가 원래 drift 신호를 유지하게 한다. 구조 손상 (stored card missing / shape invalid) 은 caller 의 1차 recompute 단계에서 이미 fail-fast 처리되므로 본 helper 는 hash-only mismatch 회복으로 제한된다. """ import copy as _copy shot_dep_cp = runner._load_prev_checkpoint("shot_dependency") upstream_deps = (shot_dep_cp or {}).get("data", {}).get("dependencies") if not upstream_deps: return None if (getattr(base_ctx, "dependencies", None) or []) == upstream_deps: # 1차 recompute 가 이미 upstream shot_dependency source — 2차 source 없음. return None try: fallback_ctx = _copy.copy(base_ctx) fallback_ctx.dependencies = upstream_deps _shot_info = {"shot_index": shi} if shi is not None else None _inputs = _collect_card_inputs( ctx=fallback_ctx, seg={"scene_index": si}, shot_info=_shot_info, ) _builder_inputs = {k: v for k, v in _inputs.items() if k != "ctx"} _builder_inputs["used_outlook_pairs"] = used_outlook_pairs _builder_inputs["state_variant_chars"] = state_variant_chars from app.core.steps.render_prompt_card import ( build_render_prompt_card as _build, compute_card_hash as _hash, ) return _hash(_build(**_builder_inputs)) except Exception as exc: # best-effort rescue — 절대 escalate 안 함. logger.warning( "verify_completion: upstream-dependency card recompute fallback " "failed (s%s_sh%s): %s — keeping original drift signal", si, shi, exc, ) return None def _user_edited_card_contract_violated( *, stored_card: Optional[Dict[str, Any]], stored_hash: Optional[str], card_inputs: Dict[str, Any], ) -> Optional[str]: """G4.1 Phase 7 Task 19 (R2-I6 / R3-I2): `_user_edited` reuse path card drift guard — testable helper. Returns: None → reuse OK (card present + shape OK + hash matches recompute). str → violation reason (caller rejects reuse → fresh path). G3.1 `_evidence_violated()` / G3.2 `_owned_validation_violated()` 패턴 mirror — caller (`_user_edited` reuse 분기) 가 본 helper 의 return 값으로 reuse vs fresh 결정. """ from app.core.steps.render_prompt_card import ( build_render_prompt_card, compute_card_hash, assert_card_shape, ) if stored_card is None or stored_hash is None: return "card_or_hash_missing" try: assert_card_shape(stored_card, where="_user_edited") except AppError as exc: return f"card_shape_invalid: {exc.message}" # Strip ctx before splat into builder (matches `_analyze_one()` pattern). builder_inputs = {k: v for k, v in card_inputs.items() if k != "ctx"} try: recomputed_card = build_render_prompt_card(**builder_inputs) recomputed_hash = compute_card_hash(recomputed_card) except AppError as exc: return f"recompute_failed: {exc.message}" if recomputed_hash != stored_hash: return ( f"hash_drift: stored={stored_hash} vs recomputed={recomputed_hash}" ) return None def _user_edited_owned_contract_violated( variation: Dict[str, Any], expected_owned: List[str], expected_camera_direction: str, is_close: bool, ) -> bool: """G3.2 round 4 IMPORTANT 4 + round 5 BLOCKING 3: ``_user_edited`` reuse 의 owned_validation drift 검사. True 반환 시 reuse 거부 + fresh 재생성. False 면 reuse 정상. 검사 순서 — 어느 한 검증 실패 시 즉시 True: 1. owned_validation field 존재 + shape OK. 2. validator type 이 현재 close framing 상태와 일치. 3. t2i_prompt_hash 일치 (round 5 BLOCKING 3 — user 가 cp 직접 편집해서 prompt 수정 시 차단). 4. owned_hash 일치. 5. camera_direction_hash 일치. """ from app.core.steps._owned_helpers import ( OWNED_VALIDATOR_CLOSE_SKIP, OWNED_VALIDATOR_FULL, assert_owned_sentinel_shape, compute_camera_direction_hash, compute_owned_hash, compute_t2i_prompt_hash, ) sentinel = variation.get("owned_validation") if sentinel is None: return True try: assert_owned_sentinel_shape(sentinel, where="_user_edited") except Exception: return True expected_validator = ( OWNED_VALIDATOR_CLOSE_SKIP if is_close else OWNED_VALIDATOR_FULL ) if sentinel.get("validator") != expected_validator: return True if sentinel.get("t2i_prompt_hash") != compute_t2i_prompt_hash( variation.get("t2i_prompt", "") ): return True if sentinel.get("owned_hash") != compute_owned_hash(expected_owned): return True if sentinel.get("camera_direction_hash") != compute_camera_direction_hash( expected_camera_direction ): return True return False # Patch C / Area A — enum literal → directive template 매핑. # code 가 element 문자열 분류 0. {element} 와 {orientation} 만 substitution. _DIRECTIVE_TEMPLATES = { "content_surface": ( "Orientation constraint: the content-bearing surface of the " "{element} is visible as follows — {orientation}. Render exactly " "this face; do not invent content on a face that is not visible " "to the camera." ), "reflective_surface": ( "Reflection constraint: the reflecting surface of the {element} " "shows — {orientation}. Render this reflection accurately; do not " "invent other reflections or scenes." ), # transparent_surface / directional_3d / non_directional / 미지 enum # 모두 영어 directive 0 (legacy NL inline 만). } def _build_bg_element_line( element: str, state: str, camera_use: str, orientation: str, directionality_class: str = "", ) -> str: """user_prompt 의 'bg element' 한 줄 + 5-class directive 분기. Contract: - orientation NL non-empty → "[방향: {orientation}]" inline (모든 class 공통, 기존 동작 보존). - directionality_class in _DIRECTIVE_TEMPLATES (content_surface / reflective_surface) AND orientation non-empty → 강한 영어 directive 를 다음 줄에 추가. - 그 외 (transparent_surface / directional_3d / non_directional / legacy cp 의 class 누락 / 미지 enum value) → 영어 directive 0. - **code 가 element 문자열을 분류하지 않음** — directionality_class enum literal 만 본다. element 가 어떤 noun 이든 무영향. """ base = f" - {element}: {state} ({camera_use})" if orientation and orientation.strip(): base += f" [방향: {orientation}]" template = _DIRECTIVE_TEMPLATES.get(directionality_class) if template and orientation and orientation.strip(): base += "\n " + template.format(element=element, orientation=orientation) return base def _build_entity_traits_block( visible_entities: list, name_by_short_id: dict, traits_by_short_id: dict, ) -> str: """G4.6 RC-H — visible_entities 안 character base 들의 stable_traits block. Codex iter 1 B1 carry — db 인자 제거, ctx 의 prebuilt map 만 사용. ThreadPool worker 안 SQLAlchemy session race 회피 (Phase 3 prebuild pattern lockstep). prebuild 는 SceneContextLoader. `_load_entity_canon_character_maps` 가 main thread 에서 1회 query. visible_entities 안 character base ID (C##) 만 추출 → prebuilt map 조회 → stable_traits 가 있는 entity 만 block 에 포함. fail-fast: visible_entities 안 character base ID 가 prebuild map 에 없으면 AppError (RO-15 silent bypass 금지). location/prop/outlook ID 는 무시 (block 은 character 한정). Args: visible_entities: shot 의 visible_entities list (C##/L##/P##/C##O##). name_by_short_id: ctx.name_by_short_id — short_id → entity_canon.name. traits_by_short_id: ctx.traits_by_short_id — short_id → parsed list. Returns: "[Entity stable_traits for this shot]\\n- C##: ...\\n\\n" 형태 문자열. block 비어있으면 "" (visible_entities 에 character 없음 또는 모든 entity 의 stable_traits 가 빈 list). Raises: AppError: visible_entities 의 character base ID 가 prebuild map 에 없음 (entity_canon project_id + entity_type='character' scoped prebuild 에서 누락). """ base_ids = { sid.split("O")[0] for sid in visible_entities if isinstance(sid, str) and sid.startswith("C") } if not base_ids: return "" missing = base_ids - set(name_by_short_id.keys()) if missing: raise AppError( code="step.scene_detail.unknown_short_id", message=( f"visible_entities references unknown character short_id: " f"{sorted(missing)} — prebuild EntityCanon map 에 없음 " f"(entity_type='character' scoped). Fix: shot_validator " f"character_ids must reference existing entities." ), status_code=400, ) lines = ["[Entity stable_traits for this shot]"] has_any = False for sid in sorted(base_ids): traits = traits_by_short_id.get(sid) or [] if not traits: continue traits_str = ", ".join(traits) lines.append(f"- {sid} ({name_by_short_id.get(sid, '')}): {traits_str}") has_any = True if not has_any: return "" return "\n".join(lines) + "\n\n" class _DetailStepMixin: def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict]: from app.core.config import settings cp = ( Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / step_id / "manifest.json" ) if cp.exists(): return json.loads(cp.read_text(encoding="utf-8")) return None def _load_fulltext(self) -> str: """raw fulltext 로드 — scene_segmentation 오프셋과 동일한 텍스트 소스.""" from app.models.project import Episode from sqlalchemy.orm import undefer ep = ( self.db.query(Episode) .options(undefer(Episode.fulltext)) .filter(Episode.id == self.episode_id) .first() ) return ep.fulltext if ep else "" class SceneDetailStep(_DetailStepMixin, StepRunner): """Step 16: 씬 상세 분석 (ThreadPool 병렬). visible_entities는 확정 데이터에서 구축.""" # W21B-W7 W-B (2026-06-12): printed_prop anchor 컨텍스트 lazy cache. # flag OFF / cp 부재 / printed_prop 그룹 0 → 빈 dict = 주입 0 + config_hash # payload 불변 (기존 cp byte-identical resume — Codex 판정 ③). _vca_prop_ctx_cache: Optional[Dict[str, Any]] = None # P8 (2026-06-20): immobilized_subject anchor 컨텍스트 lazy cache (loader 분리 — # printed_prop 와 격리). flag OFF / cp 부재 / immobilized 그룹 0 → 빈 dict = # 주입 0 + config_hash payload 불변 (기존 cp byte-identical resume). _vca_immobilized_ctx_cache: Optional[Dict[str, Any]] = None def _vca_printed_prop_context(self) -> Dict[str, Any]: if self._vca_prop_ctx_cache is None: from app.core.steps.visual_continuity_anchor_step import ( load_printed_prop_anchor_context, ) self._vca_prop_ctx_cache = load_printed_prop_anchor_context( self.project_id, self.episode_id) return self._vca_prop_ctx_cache def _vca_immobilized_subject_context(self) -> Dict[str, Any]: if self._vca_immobilized_ctx_cache is None: from app.core.steps.visual_continuity_anchor_step import ( load_immobilized_subject_anchor_context, ) self._vca_immobilized_ctx_cache = load_immobilized_subject_anchor_context( self.project_id, self.episode_id) return self._vca_immobilized_ctx_cache def _config_hash(self) -> str: """v12 (2026-05-02): Rule E/F/G/H + 신규 Rule J Primary Subject Framing (한 shot = 한 framing scale). 모델은 gemini-pro 유지. SCHEMA_VERSION 또는 PROMPT_VERSION bump 시 step_runner 가 stale cp 자동 감지.""" import hashlib import json as _json from app.core.config import settings payload = { "schema_version": SCENE_DETAIL_SCHEMA_VERSION, "prompt_version": SCENE_DETAIL_PROMPT_VERSION, "chain_bg_guide_enabled": settings.chain_bg_guide_enabled, "chain_bg_camera_meta_enabled": settings.chain_bg_camera_meta_enabled, "shot_essence_enabled": settings.shot_essence_enabled, } # W21B-W7 W-B: anchor 적용 시에만 stamp 추가 — flag ON + groups present # 면 hash 가 달라져 resume 이 no-anchor cp 를 조용히 재사용하지 않는다. # 빈 컨텍스트면 payload 불변 → 기존 cp 와 hash 동일 (Codex 판정 ③). _vca = self._vca_printed_prop_context() if _vca: payload["visual_continuity_anchor_stamp"] = _vca.get("stamp", {}) # P8: immobilized 컨텍스트도 stamp (있을 때만 — 빈 컨텍스트면 payload 불변). _vca_imm = self._vca_immobilized_subject_context() if _vca_imm: payload["immobilized_subject_anchor_stamp"] = _vca_imm.get("stamp", {}) # 카드 공간 규칙 축약 — **켰을 때만** stamp (2026-08-26 Codex BLOCK-2). # # 이 플래그는 카드의 `spatial_consistency` 를 바꾼다. hash 에 안 접으면 # ON/OFF 를 바꿔도 **옛 checkpoint 가 current 로 읽혀** A/B 두 판이 # 사실은 같은 것을 보고, 운영 resume 도 새 카드 계약을 안 태운다. # ★위 두 stamp 와 같은 관례 — OFF 면 payload 가 불변이라 **기존 판의 # hash 가 그대로 보존**된다. if settings.card_spatial_rules_scoped_enabled: payload["card_spatial_rules_scoped"] = True # ★★★2026-09-02 D cutover (Codex BLOCK) — 이 스텝이 이제 카드에 # **고증 참조 sidecar 를 적는다**. 그 산출 계약을 지문에 안 접으면 # 이미 `completed` 인 CP 가 **sidecar 없는 옛 카드를 그대로 재사용**해 # cutover 가 코드에만 있고 기존 프로젝트에는 안 닿는다. # ★위 세 stamp 와 같은 관례로 **켠 판에서만** 넣는다 — 안 켠 판은 # payload 가 불변이라 기존 지문이 그대로 보존된다(legacy 비회귀). # ★구매 신원이 아니라 **처리 지문**이다. from app.core.grounding_mode import (resolve_grounding_mode, uses_chunk_producer) # ★`project_config` 가 없는 호출자도 있다(옛 시험의 대역) — 그것은 # 옛 판이다. 없다고 여기서 터지면 안 된다. if uses_chunk_producer(resolve_grounding_mode( getattr(self, "project_config", None) or {})): from app.modules.pipeline import grounding_bundle_projection as _bp from app.modules.pipeline import grounding_central_acquisition \ as _ca from app.modules.pipeline import grounding_reference_bundle as _rb from app.modules.pipeline import grounding_sidecar_writer as _sw payload["grounding_sidecar_contract"] = { "writer": _sw.SIDECAR_WRITER_CONTRACT_VERSION, "projection": _bp.PROJECTION_CONTRACT_VERSION, "acquisition_projection": _ca.ACQUISITION_PROJECTION_VERSION, # ★묶는 규칙 자체 — dedupe 열쇠·주인 하나 자리·값 모양 "bundle": _rb.BUNDLE_CONTRACT_VERSION, # ★HITL 0 (2026-09-03): 사람 판정 지문(`fidelity_reviews`)은 **뺐다** — # production 지문이 사람 표에 매이면 그 표가 운영 절차가 된다. } return hashlib.sha256( _json.dumps(payload, sort_keys=True).encode("utf-8") ).hexdigest()[:16] def _load_d6_consumed_hashes(self) -> Dict[str, str]: """D6 T5d: master_plan cp 의 sibling field (bg_catalog_hash + shot_binding_hash) 를 dict 로 반환 — `_execute` return data 에 stamp. review iter6 minor note: scene_detail 은 binding hash 위주 (어느 shot 이 어느 bg 에 매핑인지가 핵심). catalog hash 도 보조 stamp (drift 추적). legacy master_plan cp (D6 sibling 부재) / cp 자체 부재 → 둘 다 빈 문자열. preflight (T8) 가 빈 문자열을 mismatch 로 처리. """ plans_cp = self._load_prev_checkpoint("background_master_plan") or {} plans_data = plans_cp.get("data", {}) or {} return { "consumed_bg_catalog_hash": plans_data.get("bg_catalog_hash", "") or "", "consumed_shot_binding_hash": plans_data.get("shot_binding_hash", "") or "", } def _run_shots_with_retry( self, *, tasks: List[Tuple[Dict[str, Any], Optional[Dict[str, Any]]]], ctx, system: str, schema: Dict[str, Any], results: List[Dict[str, Any]], failed: int, primary_max_workers: int = 5, retry_max_workers: int = 3, retry_sleep_seconds: float = 2.0, ) -> Tuple[List[Dict[str, Any]], int, List[Dict[str, Any]]]: """1차 ThreadPool + retry path 의 shot-level typed failure handling. 본 helper 는 `_execute` 에서 분리된 retry 코어. unit test 용이 + 운영 retry policy 의 단일 진입점. 흐름: 1. 1차 ThreadPool (primary_max_workers): 각 shot 의 `_analyze_one` 호출. AppError / 일반 Exception 모두 try/except 로 잡고 typed failure record 보존, retry 대상에 포함. 2. failed > 0 시 retry path (retry_max_workers): 1차 실패 shot 만 재실행. retry_sleep_seconds 만큼 sleep 후. 같은 typed failure 처리. 3. retry 후에도 failed > 0 면 `AppError(code= 'step.scene_detail.contract_violations_after_retry')` raise. Returns: (results, failed) — results 는 성공한 _analyze_one 결과 append 된 리스트, failed 는 최종 실패 count (성공 시 0). Raises: AppError(code="step.scene_detail.contract_violations_after_retry") — retry 후에도 fail 잔존 시. typed summary message 포함. """ failed_shots: List[Dict[str, Any]] = [] def _record_failure(_seg, _sh, _exc, _stage: str): _code = getattr(_exc, "code", None) or "step.scene_detail.unexpected" failed_shots.append({ "scene_index": _seg.get("scene_index", 0), "shot_index": _sh.get("shot_index") if _sh else None, "code": _code, "stage": _stage, "message": str(_exc), }) # 정지 표를 worker 로 실어 보낸다. 표는 스레드마다 따로라, 안 나르면 # 팬아웃 안에서 「멈춰라」가 통째로 안 들린다. from app.core.image_call_budget import bind_current_budget with ThreadPoolExecutor(max_workers=primary_max_workers) as pool: _one = bind_current_budget(self._analyze_one) futures = { pool.submit(_one, seg, sh, ctx, system, schema): (seg, sh) for seg, sh in tasks } for f in as_completed(futures): _seg, _sh = futures[f] try: r = f.result() except AppError as exc: # 멈추라는 말과 락을 놓친 것은 **이 샷 하나의 실패가 아니다.** # 실패로 적으면 아래 재시도가 다시 부르고, 나머지 샷도 계속 # 돌아 돈이 나간다. 위로 올려 스텝을 세운다. from app.core.run_control import is_abort # ★코드 목록은 **한 곳**(`ABORT_CODES`)이다 — # 여기 다시 적으면 한쪽만 고쳐진다 if is_abort(exc): logger.warning("scene_detail 중단 — %s", exc.message) raise _record_failure(_seg, _sh, exc, "primary") logger.warning( "Scene %s Shot %s contract violation (primary): %s", _seg.get("scene_index"), _sh.get("shot_index") if _sh else None, exc, ) r = None except Exception as exc: _record_failure(_seg, _sh, exc, "primary_unexpected") logger.error( "Scene %s Shot %s unexpected (primary): %s", _seg.get("scene_index"), _sh.get("shot_index") if _sh else None, exc, ) r = None if r: results.append(r) else: failed += 1 # 실패 1회 retry — contract violation 도 포함 (LLM stochastic 회복 기회). # 단, echo mismatch 는 inner `_check_prompts` retry 가 이미 1회 처리 → # outer retry budget 부여 안 함 (plan §Task 6 의 "1 회 retry 후에도 echo # set mismatch 면 AppError raise. graceful fix 불가능" 정책 일관). _NON_RETRYABLE_CODES = {"scene_detail.frame_spatial_contract_echo_mismatch"} if failed > 0: done_keys = set() for r in results: si = r.get("scene_index") shot_idx = r.get("_shot_index") done_keys.add((si, shot_idx)) non_retryable_keys = { (f["scene_index"], f["shot_index"]) for f in failed_shots if f.get("code") in _NON_RETRYABLE_CODES } retry_tasks = [ (seg, sh) for seg, sh in tasks if (seg.get("scene_index", 0), sh.get("shot_index") if sh else None) not in done_keys and (seg.get("scene_index", 0), sh.get("shot_index") if sh else None) not in non_retryable_keys ] if retry_tasks: _last5 = [ f"S{f['scene_index']}_Shot{f['shot_index']}({f.get('code','?')})" for f in failed_shots[-5:] ] logger.warning( "Retrying %d failed tasks (typed last 5: %s)", len(retry_tasks), _last5, ) # ★직전 실패의 계약 위반 문구를 샷마다 모아 재시도에 싣는다 — 같은 입력이면 같은 답이다. _why: Dict[tuple, List[str]] = {} for _f in failed_shots: if str(_f.get("code") or "").startswith("step.scene_detail.contract_violation"): _why.setdefault((_f["scene_index"], _f["shot_index"]), []).append(_f["message"]) time.sleep(retry_sleep_seconds) with ThreadPoolExecutor(max_workers=retry_max_workers) as pool2: _one2 = bind_current_budget(self._analyze_one) futures2 = { pool2.submit(_one2, seg, sh, ctx, system, schema, corrections=_why.get((seg.get("scene_index", 0), sh.get("shot_index") if sh else None))): (seg, sh) for seg, sh in retry_tasks } for f in as_completed(futures2): _seg, _sh = futures2[f] try: r = f.result() except AppError as exc: # 여기서 삼키면 멈추라는 말을 듣고도 **다시 부른다.** from app.core.run_control import is_abort # ★코드 목록은 **한 곳**(`ABORT_CODES`)이다 — # 여기 다시 적으면 한쪽만 고쳐진다 if is_abort(exc): logger.warning( "scene_detail 재시도 중단 — %s", exc.message) raise _record_failure(_seg, _sh, exc, "retry") logger.error( "Scene %s Shot %s contract violation (retry): %s", _seg.get("scene_index"), _sh.get("shot_index") if _sh else None, exc, ) continue except Exception as exc: _record_failure(_seg, _sh, exc, "retry_unexpected") logger.error("Retry unexpected exception: %s", exc) continue if r: results.append(r) failed -= 1 # retry 후에도 failure 가 남으면 **성공분을 버리지 않고** partial 로 # 넘긴다 (2026-08-04). 이전에는 여기서 raise 해 `_execute` 의 return 에 # 도달하지 못했고, step_runner 의 checkpoint 저장이 반환 이후라 성공한 # shot 전량이 폐기됐다 — 실측: 255 중 1 실패에 254 건의 유료 호출과 # scene_still 0행. # # "silent continue 금지" 라는 원래 의도는 유지된다 — step_manifest 의 # `allow_partial_downstream: False` 가 partial 일 때 하류(shot_dependency_t2i # / t2i_review)를 차단하므로, 계약 위반이 박힌 t2i_prompt 가 이미지 # 단계로 흘러가지 않는다. 실패는 typed 목록으로 결과에 실어 진단을 남긴다. # `failed_shots` 는 **시도 이력**이라 retry 로 회복된 shot 도 남아 있다 # (primary 에서 append 하고 retry 성공 시 지우지 않는다). 그대로 CP 에 # 실으면 다음 resume 이 그 키를 실패로 읽어 **이미 성공한 shot 을 다시 # 유료로 부른다.** 그래서 최종 results 에 없는 것만 unresolved 로 남긴다. # 같은 shot 이 primary·retry 두 번 기록되므로 **키 기준으로 뒤엣것만** # 남긴다 (retry stage 정보가 더 최신이다). 안 그러면 미해결 1건인데 # 목록이 2건으로 보여 로그·CP 를 오독하게 된다. _resolved = { (r.get("scene_index"), r.get("_shot_index")) for r in results } _by_key: Dict[tuple, Dict[str, Any]] = {} for f in failed_shots: k = (f.get("scene_index"), f.get("shot_index")) if k in _resolved: continue _by_key[k] = f unresolved = list(_by_key.values()) if failed > 0: _summary = "; ".join( f"S{f['scene_index']}_Shot{f['shot_index']} " f"[{f.get('stage','?')}/{f.get('code','?')}]: {f['message'][:120]}" for f in unresolved[-5:] ) logger.error( "scene_detail: %d shots failed after 1 retry (미해결 %d · 시도 " "이력 %d) — 성공 %d건은 partial 로 보존한다. Last 5: %s", failed, len(unresolved), len(failed_shots), len(results), _summary, ) return results, failed, unresolved def _build_llm_inputs(self, ctx) -> Tuple[str, Dict[str, Any]]: """system prompt + JSON schema 빌드. 2026-05-11 redo-shot service 재사용용 helper 로 추출. 본 메서드는 `_execute` 의 setup 단계와 1:1 동일 — system prompt 에 visual_world_rules / planning_context 를 동적으로 inject 한다. Returns: (system, schema) — system 은 inject 완료된 최종 system prompt. """ # v3: 전용 프롬프트 → scene_extractor_v2 fallback try: system = load_prompt("scene_detail", "system") schema = load_schema("scene_detail", "detail_schema") except Exception: system = load_prompt("scene_extractor_v2", "turn_scene_detail") schema = load_schema("scene_extractor_v2", "scene_detail_schema") # visual_world_rules → system prompt 에 시대/지역/의상 규칙 주입 rd = ctx.world_rules or {} if rd: world_lines = [] if rd.get("era"): world_lines.append(f"시대: {rd['era']}") if rd.get("region"): world_lines.append(f"지역/국가: {rd['region']}") for r in rd.get("rules", []): if r.get("rule_type") in ("time_period", "costume", "technology"): world_lines.append(f"[{r['rule_type']}] {r.get('visual_guideline', '')}") if world_lines: system += ( "\n\n## 시각적 세계관 (T2I 프롬프트 생성 시 반드시 반영)" "\n아래 설정을 모든 T2I 프롬프트에 반영하세요." "\n특히 지역/국가 설정에 맞는 인종, 건축양식, 복장, 소품을 사용하세요." "\n참조 이미지가 없는 인물/배경/소품은 아래 세계관에 기반하여 최대한 구체적으로 묘사하세요." "\n" + "\n".join(world_lines) ) # t2i_context — 시각 스타일 참고 (복사 금지) _t2i_ctx = rd.get("t2i_context", "") if _t2i_ctx: system += ( "\n\n## T2I 시각 스타일 참고 (복사 금지)" "\n아래는 프로젝트 전체의 시각 톤입니다. t2i_prompt 에 이 문장을 그대로 복사하지 마세요." "\n색감/조명/분위기만 간접 반영하고, 지명이나 장소 열거는 t2i_prompt 에 넣지 마세요." f"\n{_t2i_ctx}" ) # 기획서 톤/비주얼 컨셉 → system prompt 에 주입 pctx = ctx.planning_context if pctx is not None: system += pctx.inject_if_available("tone_mood", "## 기획서: 톤/분위기") system += pctx.inject_if_available("visual_concepts", "## 기획서: 비주얼 컨셉") # 제작자 정정 채널 (wave3) — 정정 없으면 빈 문자열 = byte-identical from app.core.creator_corrections import project_corrections_block system += project_corrections_block(self.project_id) return system, schema def _execute(self, mode="resume") -> Dict[str, Any]: # Phase 3.6 DTO: 20+ 체크포인트 로드를 SceneContextLoader로 위임. # Phase 3b.6: _analyze_one을 class method로 승격 + closure → ctx.X. ctx = SceneContextLoader(self).load_all() # _execute 본체에서 직접 쓰이는 필드만 로컬 alias (가독성). segments = ctx.segments selected_map = ctx.selected_map shot_scenes_map = ctx.shot_scenes_map # v3: system/schema + world_rules + planning context — helper 로 추출 # (redo-shot service 가 재사용 — 2026-05-11 사용자 권장안 #2 축소형). system, schema = self._build_llm_inputs(ctx) results = [] failed = 0 # v4: 현재 선택된 shot 키 세트 구축 (교차 필터용) current_shot_keys: set = set() for si, shots in shot_scenes_map.items(): if shots: for sh in shots: current_shot_keys.add((si, sh.get("shot_index"))) else: # shot이 없는 씬 = selected_shot_indices=[] 이면 스킵 대상 # selected_map에 키 자체가 없으면 legacy (shot 미사용 씬) → 포함 if si not in selected_map: current_shot_keys.add((si, None)) # v4: 사용자 편집 보존 — _user_edited 마킹된 shot은 스킵 (현재 selection과 교차 필터) existing_detail_cp = self._load_prev_checkpoint("scene_detail") edited_results = [] edited_keys = set() if existing_detail_cp and existing_detail_cp.get("data", {}).get("scenes"): for s in existing_detail_cp["data"]["scenes"]: if s.get("_user_edited"): key = (s.get("scene_index"), s.get("_shot_index")) # 현재 selection에 있는 shot만 보존 if key in current_shot_keys: # G3.1 PROBLEM #4: _user_edited reuse 는 strict schema 우회 path — # 옛 v13 cp 가 새 cp 로 그대로 적재되면 4 필드 없이 downstream 흘러감. # 옛 cp 처럼 normalize → 4 필드 default + confidence='legacy' 마킹. # 새로 LLM 이 만든 v14 cp 면 4 필드 이미 존재 → noop. _normalize_scene_detail_result( s, where="scene_detail._user_edited" ) # G3.1 (Codex IMPORTANT 1): _user_edited reuse 의 4 필드가 # legacy 가 아닌데 contract 위반 (source_facts=[] AND # confidence='high' 등) 인 경우 — user 가 cp 직접 편집한 # 케이스 — 그대로 새 cp 에 보존되는 silent fallback 차단. # legacy 는 옛 cp 정상 backfill 이라 skip. # G3.2 (round 4 IMPORTANT 4): owned drift reload 위한 # owned + camera_direction ground truth. _owned_for_var = ctx.chain_bg_owned_by_shot.get( (s.get("scene_index"), s.get("_shot_index")) ) or [] _staging = ctx.staging_map.get( f"{s.get('scene_index')}_{s.get('_shot_index')}" ) or {} _cam_dir = ( _staging.get("camera_direction", "") if isinstance(_staging, dict) else "" ) # framing_scale enum SOT v1 (2026-05-15): shot path # (_staging dict) 만 helper read. legacy reuse path # (_staging=None) 는 default False (Gate 4 legacy # pass-through). _cam_dir 변수는 보존 — # build_owned_sentinel 의 camera_direction arg 용. _is_close = False if isinstance(_staging, dict): _is_close = ( get_framing_scale_or_raise( _staging, where=( f"detail_steps._user_edited_reuse " f"S{s.get('scene_index')}_" f"Shot{s.get('_shot_index')}" ), ) == FRAMING_CLOSE ) # G4.1 Phase 7 Task 19 (5-point modify scope guard 5/5, # R2-I6 / R3-I2) + Wave 4 R4 B2: card hash drift guard. # ctx-driven single source — outlook/visible/fixed/forward # 모두 ctx 에서 derive (옛 hardcode empty 제거). _g41_shot_idx_ue = s.get("_shot_index") _g41_card_inputs_ue = _collect_card_inputs( ctx=ctx, seg={"scene_index": s.get("scene_index")}, shot_info=( {"shot_index": _g41_shot_idx_ue} if _g41_shot_idx_ue is not None else None ), ) # Fix B (2026-05-10): _user_edited reuse path 도 narrow # args 적용 — stored RPC (narrow) 와 recomputed RPC 가 # 같은 narrow 파라미터로 비교되어야 hash drift 회피. _used_pairs_ue = _compute_used_outlook_pairs( s.get("t2i_variations", []) or [], ) _visible_ue = _derive_visible_for_shot( ctx, s.get("scene_index"), _g41_shot_idx_ue, ) _visible_chars_ue: Set[str] = { sid.split("O")[0] for sid in (_visible_ue or []) if isinstance(sid, str) and sid.startswith("C") } _state_var_chars_ue = _detect_state_variant_chars( _staging if isinstance(_staging, dict) else None, _visible_chars_ue, ctx, ) _g41_card_inputs_ue["used_outlook_pairs"] = _used_pairs_ue _g41_card_inputs_ue["state_variant_chars"] = _state_var_chars_ue _g41_card_violation = _user_edited_card_contract_violated( stored_card=s.get("render_prompt_card"), stored_hash=s.get("render_prompt_card_hash"), card_inputs=_g41_card_inputs_ue, ) # Gate via stored card presence: only enforce when the # cp explicitly carries a card field (v16+ resume). # v15 cp resume 는 step_runner schema gate 가 force # escalate — verify path 의 false-positive 차단. if ( _g41_card_violation is not None and s.get("render_prompt_card") is not None ): logger.warning( "scene_detail _user_edited %s: card contract " "violated (%s) — reuse 거부, fresh 재생성.", key, _g41_card_violation, ) # break out of variation loop equivalent — do not # add to edited_keys. continue for _var in s.get("t2i_variations", []) or []: if not isinstance(_var, dict): continue if _var.get("confidence") == "legacy": continue try: assert_fresh_llm_evidence(_var, "scene_detail._user_edited") except AppError as exc: if exc.code != "step.contract_violation": raise logger.warning( "scene_detail _user_edited %s: contract violation — %s", key, exc.message, ) # contract 위반 _user_edited 는 reuse 하지 않고 fresh # LLM 으로 재생성 (key 를 edited_keys 에 안 넣음). break # G3.2 round 4 IMPORTANT 4: owned drift 검증. # mismatch 시 reuse 거부 + fresh 재생성 (G3.1 패턴). if _user_edited_owned_contract_violated( _var, _owned_for_var, _cam_dir, _is_close, ): logger.warning( "scene_detail _user_edited %s: owned drift — " "reuse 거부, fresh 재생성.", key, ) break else: edited_keys.add(key) edited_results.append(s) continue # 위 break 일 때만 도달 — fresh LLM 으로 재생성 진행. logger.info( "scene_detail: user-edited shot %s contract violation — refresh", key, ) else: logger.info("scene_detail: user-edited shot %s no longer selected — dropping", key) if edited_keys: logger.info("scene_detail: %d user-edited shots preserved", len(edited_keys)) # partial resume 재사용 (2026-08-04) — 직전 실행이 partial 로 끝났으면 # 그때 성공한 shot 을 다시 유료로 부르지 않는다. # # 게이트는 `config_hash` 일치 하나로 충분하다 — payload 에 # SCENE_DETAIL_PROMPT_VERSION / SCHEMA_VERSION / anchor stamp 가 모두 # 들어 있어(_config_hash 참조), 판이 바뀌거나 상류 anchor 가 바뀌면 # 해시가 달라져 재사용이 저절로 꺼진다. `mode='force'` 는 전량 재생성이 # 의도이므로 제외한다. # ★게이트에 `status == "partial"` 이 반드시 있어야 한다. `mode` 만으로는 # 부족하다 — step_runner 가 prior_state(stale/failed) 를 force 로 격상해도 # `_execute_rerun_self()` 는 `_execute(mode="resume")` 를 부르므로 # (step_runner.py:1129,1153) 여기 mode 는 여전히 "resume" 이다. 그리고 # `config_hash` 는 **계약 해시**(schema/prompt/설정/anchor)이지 입력 해시가 # 아니라 씬·shot selection 변화를 잡지 못한다. 직전이 partial 로 끝났을 # 때만 재사용해야 상류 변경으로 stale 된 옛 결과를 되살리지 않는다. reused_keys: set = set() if ( mode != "force" and existing_detail_cp and existing_detail_cp.get("status") == "partial" ): _prev_hash = existing_detail_cp.get("config_hash") if _prev_hash and _prev_hash == self._config_hash(): _prev_failed = { (f.get("scene_index"), f.get("shot_index")) for f in (existing_detail_cp.get("failed_shots") or []) } for s in (existing_detail_cp.get("data", {}) or {}).get("scenes", []): key = (s.get("scene_index"), s.get("_shot_index")) if key in edited_keys or key in reused_keys: continue # 현재 selection 밖 / 직전에 실패한 shot 은 새로 만든다. if key not in current_shot_keys or key in _prev_failed: continue # 산출이 비어 있으면 성공분이 아니다 (verify_completion 의 # partial 판정 기준과 동일). if not s.get("t2i_variations"): continue # ★규칙 위반으로 표시된 컷도 다시 만든다 (2026-08-07). # # 종전에는 "결과물이 있으면 성공"으로 보아 그대로 재사용했다. # 그런데 `contract_violation` 은 결과물은 있고 규칙만 어긴 # 상태다 — 재사용하면 그 컷이 영원히 위반으로 남고, # `verify_completion` 이 그것 하나 때문에 단계 전체를 # "일부만 완료"로 표시해 다음 단계가 막힌다. 실제로 256개 중 # 255개가 정상인데 한 컷 때문에 파이프라인이 섰고, resume 을 # 몇 번 돌려도 같은 자리에서 다시 막혔다. if s.get("status") == "contract_violation": logger.info( "scene_detail: S%s shot%s 는 규칙 위반 상태 — " "재사용하지 않고 다시 만든다", s.get("scene_index"), s.get("_shot_index")) continue _normalize_scene_detail_result( s, where="scene_detail._partial_reuse" ) edited_results.append(s) reused_keys.add(key) if reused_keys: logger.info( "scene_detail: partial resume — 직전 성공 %d shot 재사용 " "(config_hash 일치, 실패 %d shot 만 재호출)", len(reused_keys), len(_prev_failed), ) elif _prev_hash: logger.info( "scene_detail: 직전 cp config_hash 불일치 — 전량 재생성 " "(prompt/schema/anchor 변경)", ) # v4: shot 단위 태스크 구성 tasks = [] for seg in segments: si = seg.get("scene_index", 0) shots_for_scene = shot_scenes_map.get(si, []) if shots_for_scene: for sh in shots_for_scene: key = (si, sh.get("shot_index")) if key in edited_keys or key in reused_keys: continue # 사용자 편집 보존 / partial resume 재사용 tasks.append((seg, sh)) elif si not in selected_map: # legacy: shot_selection 자체가 없는 씬 → scene-level fallback if (si, None) not in edited_keys and (si, None) not in reused_keys: tasks.append((seg, None)) # else: selected_shot_indices=[] → 전부 deselect → 이 씬 스킵 # 편집된 결과를 미리 추가 results.extend(edited_results) logger.info("scene_detail: %d segments → %d tasks (shot-based, %d edited skipped)", len(segments), len(tasks), len(edited_keys)) # 2026-05-11 runner fix — shot-level typed failure + retry. # 기존 1차 executor 가 try 없이 f.result() 호출 → _analyze_one 의 raise # AppError (Rule X-2 등 contract violation) 가 step 전체 abort 시킴. # 57 shot 중 1 LLM 응답 contract 위반 = 56 shot 의 successful 응답까지 # 모두 폐기 → 큰 비용 손실. 본 fix 는 retry helper 로 추출 — 1차 + # retry path 양쪽에서 AppError 를 shot-level typed failure 로 잡고 # retry 대상에 넣음. retry 후에도 잔존 failure 가 있으면 성공분을 # partial 로 보존하고 typed 실패 목록을 결과에 실어 보낸다 # (2026-08-04 — 이전에는 여기서 raise 해 성공분이 통째로 폐기됐다). results, failed, failed_shots = self._run_shots_with_retry( tasks=tasks, ctx=ctx, system=system, schema=schema, results=results, failed=failed, ) results.sort(key=lambda r: (r.get("scene_index", 0), r.get("_shot_index") or 0)) new_results_count = len(results) - len(edited_results) # D6 T5d: master_plan cp 의 hash 둘 다 stamp. d6_hashes = self._load_d6_consumed_hashes() out = { "completed_count": len(results), "applicable_count": len(tasks) + len(edited_results), "failed_count": max(0, len(tasks) - new_results_count), "schema_version": SCENE_DETAIL_SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"scenes": results, **d6_hashes}, } # partial 진단 — 어느 shot 이 어떤 typed code 로 남았는지 CP 에 보존한다. # 다음 resume 이 이 목록으로 실패분만 다시 부른다 (아래 재사용 게이트). if failed_shots: out["failed_shots"] = [ {k: v for k, v in f.items() if k != "message"} | {"message": (f.get("message") or "")[:300]} for f in failed_shots ] return out def verify_completion(self): """Group 1 #3 + G3.2: scene_detail 산출물 검증. 검증 항목: - data.scenes 존재. - 각 result 에 ``t2i_variations`` 1+개 (빈 배열은 partial). - (G3.2 Task 19) 각 variation 에 owned_validation sentinel 존재 + shape OK. - (G3.2 Task 19 / round 4 BLOCKING 2 / round 5 BLOCKING 3, 4) 4 종 drift 검증: validator type / t2i_prompt_hash / owned_hash / camera_direction_hash. close-skip variation 도 hash 일치 강제. - (G3.2 Task 19) ``result["status"] == "contract_violation"`` 이면 partial 마킹. source 우선순위: step_runner 가 보존한 ``_last_execute_result`` → cp fallback. 둘 다 없으면 missing. framing_scale enum SOT v1 (2026-05-15): staging.framing_scale enum helper read — 생성 path 와 동일한 enum 소비 (regex 폐기). partial / blocked 시 manifest ``allow_partial_downstream=False`` 가 downstream 차단. """ from app.core.integrity_report import CompletionReport from app.core.steps._owned_helpers import ( OWNED_VALIDATOR_CLOSE_SKIP, OWNED_VALIDATOR_FULL, assert_owned_sentinel_shape, compute_camera_direction_hash, compute_owned_hash, compute_t2i_prompt_hash, ) result = getattr(self, "_last_execute_result", None) if result is None: cp = self._load_prev_checkpoint("scene_detail") if not cp: return CompletionReport( is_complete=False, missing=["scene_detail cp + result 모두 부재"], severity="missing", metadata={"total_results": 0, "failed_count": 0, "failed_indices": []}, ) result = cp scenes = result.get("data", {}).get("scenes", []) or [] # G3.1: 옛 cp (4 필드 누락) lazy backfill — t2i_variations 각 variation 에 # source_facts/visual_inferences/creative_decisions/confidence default 주입, # confidence='legacy' 마킹. 새 LLM 출력은 _execute 가 strict schema + assert # 로 보장하므로 normalize 가 도달해도 noop. for r in scenes: _normalize_scene_detail_result(r, where="scene_detail.verify_completion") if not scenes: return CompletionReport( is_complete=False, missing=["scene_detail data.scenes 빈 리스트"], severity="missing", metadata={"total_results": 0, "failed_count": 0, "failed_indices": []}, ) # G3.2: owned + camera_direction reload — verify 시점의 ground truth. # 옛 cp / partial v5 cp 차단은 loader 가 fail-fast (AppError raise). # Wave 6 BLOCKING fix: blanket except 가 contract_violation 을 silent # 흡수하던 패턴 제거. bg-off 면 loader 가 정상 {} 리턴 → try 불요. # bg-on + AppError → CompletionReport(missing) 로 전환 (silent {} 차단). # 그 외 모든 예외는 surface (test/infra 누수 검출). from app.core.config import settings from app.core.steps.scene_context_loader import SceneContextLoader _loader = SceneContextLoader(self) _bg_on = settings.background_mode in {"on", "floor_plan_anchored"} loader_violations: List[str] = [] if _bg_on: try: owned_by_shot = _loader._load_chain_bg_owned_by_shot() except AppError as exc: # contract_violation surface — silent {} fallback 금지. owned_by_shot = {} loader_violations.append( f"owned loader contract violation: {exc.message}" ) else: # bg-off path: loader 자체가 안전한 {} 반환 (옛 v4 cp 잔존도 통과). owned_by_shot = _loader._load_chain_bg_owned_by_shot() # staging_map: shot_staging cp 부재 시 loader 가 자연스러운 {} 반환. # owned 와 동일한 bg-mode 분기로 일관성 유지 — bg-off 면 try 불요, # bg-on 일 때만 미래 contract 강화 대비 AppError catch (silent 흡수 X). if _bg_on: try: staging_map = _loader._load_staging_map() except AppError as exc: staging_map = {} loader_violations.append( f"staging loader contract violation: {exc.message}" ) else: staging_map = _loader._load_staging_map() # G4.1 Wave 4 R4 B1: card recompute 가 진짜 ctx-derived input 사용해야 # `_analyze_one()` 와 일치한다. owned/staging 만 reload 하던 옛 path 는 # outlook_pairs/visible/bg_id/fixed_elements/forward_zoom_targets 모두 # 빈/None 으로 hardcode 해 production 에서 false drift cascade 발생 → # 전체 SceneContextLoader.load_all 호출. lazy: 실제로 card check 가 # active 인 result 가 1+개 있을 때만 호출 (옛 fixture / pre-v16 cp 회귀 # 차단 — db stub 없는 test 가 load_all 트리거 X). _verify_ctx = None _verify_ctx_loaded = False def _ensure_verify_ctx(): nonlocal _verify_ctx, _verify_ctx_loaded if _verify_ctx_loaded: return _verify_ctx _verify_ctx_loaded = True try: _verify_ctx = _loader.load_all() except AppError as exc: _verify_ctx = None loader_violations.append( f"scene_context loader contract violation: {exc.message}" ) except ( AttributeError, TypeError, KeyError, ValueError, RuntimeError, OSError, ImportError, SQLAlchemyError, ) as exc: # N4 cleanup (+ review iter1): blanket `except Exception` 을 # specific class set 으로 좁힘. KeyboardInterrupt / SystemExit / # MemoryError 같은 critical signal 은 propagate. 본 catch 대상: # - DB stub 없는 test fixture 누수 (AttributeError / TypeError) # - 잘못된 cp shape / missing key (KeyError / ValueError) # - 일반 infra 누수 (RuntimeError / OSError) # - lazy import 실패 (ImportError) # - DB connection 단절 / 타임아웃 / 쿼리 오류 # (SQLAlchemyError — load_all 이 db.execute() 호출하므로 # DBAPIError / OperationalError / InterfaceError 모두 cover. # review iter1 IMPORTANT — 이전 tuple 은 SQLAlchemy 계열을 # 누수시켜 verify_completion 이 graceful missing 대신 # uncaught raise 했음). _verify_ctx = None loader_violations.append( f"scene_context loader infra error: {exc}" ) return _verify_ctx failed_indices: list = [] contract_violations: list = [] sentinel_drifted: list = [] # G4.1 Phase 6 Task 18: card drift detection — shot-level (G3.2 sentinel # 패턴 mirror, but card 는 shot-level + sentinel 은 variation-level). # Gate: result.schema_version == 7 (v16 fresh write). v15 cp 는 step_runner # _config_hash mismatch detection 에서 mode='force' 로 escalate (R1-B1 # / R3-B1) — verify_completion 가 v15 cp 를 직접 받는 것은 test fixture # 시나리오 (Wave 3 fixture cleanup 대기). 정상 production path 에서는 # 항상 schema_version=7 result 가 도달. card_drifted: list = [] _result_schema_version = result.get("schema_version", 0) _card_check_active = (_result_schema_version == SCENE_DETAIL_SCHEMA_VERSION) # I5 cleanup: card check bypass warning 을 step 당 1회만 emit # (이전: per-r loop 내부 로깅 → N scene 마다 redundant). schema_version 은 # outer scope 의 result 에서 도출되므로 모든 scene r 에 동일 — 한 번만 # 발사하고 정확한 affected count 를 함께 보고. if not _card_check_active and scenes: logger.warning( "verify_completion: card check bypass — result schema_version=%s " "!= current=%s (step_runner force escalate path expected, " "%d scene results affected).", _result_schema_version, SCENE_DETAIL_SCHEMA_VERSION, len(scenes), ) from app.core.steps.render_prompt_card import ( assert_card_shape as _g41_assert_card_shape, build_render_prompt_card as _g41_build_card, compute_card_hash as _g41_compute_card_hash, ) for r in scenes: tvars = r.get("t2i_variations", []) or [] si = r.get("scene_index") shi = r.get("_shot_index") if not tvars: failed_indices.append((si, shi)) continue if r.get("status") == "contract_violation": contract_violations.append((si, shi)) # G4.1: card drift check (shot-level, BEFORE per-variation sentinel # check). card 는 sentinel 과 분리된 scope — coexist. Gate via # _card_check_active — v15 cp / pre-G4.1 fixture 는 step_runner gate # 가 처리하므로 verify_completion 단에서 false-positive 방지. if not _card_check_active: # I5 cleanup: warning 은 for-loop 진입 전 1회 발사됨 (위쪽 가드). # 여기는 per-r skip path — 추가 로깅 없음. pass else: _stored_card = r.get("render_prompt_card") _stored_card_hash = r.get("render_prompt_card_hash") if _stored_card is None or _stored_card_hash is None: # v15 cp 잔존 — schema_version=7 expected. card_drifted.append((si, shi, "missing")) else: try: _g41_assert_card_shape( _stored_card, where=f"verify_completion(s{si}_sh{shi})", ) except Exception: card_drifted.append((si, shi, "shape")) else: # Codex 합의 BLOCK3 1단계 (2026-08-12): 저장 카드를 # 다시 해시해 저장 해시와 대조 — live ctx 와 무관해 # 하류 재산출로 오염되지 않으면서 저장 손상(한쪽 # 오염·잘못된 덮어쓰기)을 잡는다. 구조 태그 = 차단. # live recompute 는 그대로 진행한다 — 손상 케이스에 # "hash" 태그가 함께 남아도 self_hash 가 차단을 확정 # 하고, 둘 다 남는 것이 감사에는 더 정확하다. if _g41_compute_card_hash(_stored_card) != _stored_card_hash: card_drifted.append((si, shi, "self_hash")) # G4.1 Wave 4 R4 B1: ctx-derived recompute (single source # = `_analyze_one()` 와 동일 helper). 옛 hardcode empty # 인풋 path (outlook_pairs=[]/bg_id=None/fixed_elements=[] # 등) 가 false-pass cascade 만든 결함 fix. _shot_info_for_v: Optional[Dict[str, Any]] = None if shi is not None: _shot_info_for_v = {"shot_index": shi} # B4: shot path → staging 부재면 builder raise # (staging_not_applicable marker 자동 X). # legacy / shi=None 만 _derive 가 marker synthesize. _ctx_for_recompute = _ensure_verify_ctx() if _ctx_for_recompute is None: # Loader 실패 surface — recompute skip + drift. # NB: outer for-r 의 sentinel check 도 진행해야 하므로 # `continue` 안 함 (옛 owned check path 유지). card_drifted.append((si, shi, "recompute_failed")) else: try: _card_inputs_v = _collect_card_inputs( ctx=_ctx_for_recompute, seg={"scene_index": si}, shot_info=_shot_info_for_v, ) except AppError: card_drifted.append((si, shi, "recompute_failed")) else: _builder_inputs_v = { k: v for k, v in _card_inputs_v.items() if k != "ctx" } # Fix B (2026-05-10) — narrow args 산출. _analyze_one # 와 동일 helper 사용 → narrow stored hash 와 정합 # (verify hash drift false-positive 회피). _used_pairs_v = _compute_used_outlook_pairs(tvars) _visible_v = _derive_visible_for_shot( _ctx_for_recompute, si, shi, ) _visible_chars_v: Set[str] = { sid.split("O")[0] for sid in (_visible_v or []) if isinstance(sid, str) and sid.startswith("C") } _staging_v = ( staging_map.get(f"{si}_{shi}") if isinstance(staging_map, dict) else None ) _state_var_chars_v = _detect_state_variant_chars( _staging_v, _visible_chars_v, _ctx_for_recompute, ) _builder_inputs_v["used_outlook_pairs"] = _used_pairs_v _builder_inputs_v["state_variant_chars"] = _state_var_chars_v # Block B B8 (plan v2.1.3 / spec V5 §4.3): card # recompute exception 분기. # - AppError → card_drifted "recompute_failed" # (structural tag — origin=contract_drift via # aggregation). 기존 severity 보존 (loader_violations # 누적 X — production 회귀 방지). # - 그 외 Exception → AppError(step.verify_crashed) # raise (자동 force 금지, fail-fast). try: _recomputed_card = _g41_build_card( **_builder_inputs_v ) _recomputed_hash = _g41_compute_card_hash( _recomputed_card ) except AppError: card_drifted.append( (si, shi, "recompute_failed") ) except Exception as _exc: raise AppError( code="step.verify_crashed", message=( f"build_render_prompt_card crashed " f"(s{si}_sh{shi}): " f"{type(_exc).__name__}: {_exc}" ), ) from _exc else: if _recomputed_hash != _stored_card_hash: # FINDING 8: dual-source verify. 1차 # recompute 가 downstream # shot_dependency_t2i source 를 읽어 # false drift 일 수 있음 — upstream # shot_dependency 로 1회 fallback # recompute 후 일치하면 drift 아님. _fallback_hash = ( _recompute_card_hash_with_upstream_dependency( runner=self, base_ctx=_ctx_for_recompute, si=si, shi=shi, used_outlook_pairs=_used_pairs_v, state_variant_chars=( _state_var_chars_v ), ) ) if _fallback_hash != _stored_card_hash: card_drifted.append( (si, shi, "hash") ) else: logger.info( "verify_completion: " "s%s_sh%s card hash drift " "resolved by upstream " "shot_dependency recompute " "(downstream " "shot_dependency_t2i " "false-positive)", si, shi, ) for var in tvars: sentinel = var.get("owned_validation") if sentinel is None: sentinel_drifted.append((si, shi, "missing")) continue try: assert_owned_sentinel_shape( sentinel, where=f"scene_detail.verify s{si}_shot{shi}", ) except Exception: sentinel_drifted.append((si, shi, "shape")) continue # round 4 BLOCKING 2 + round 5 BLOCKING 3, 4: 4 종 drift 검증. # 1) validator type 이 현재 close framing 상태와 일치. expected_owned = owned_by_shot.get((si, shi), []) staging_key = f"{si}_{shi}" stg = staging_map.get(staging_key) if isinstance(staging_map, dict) else None cam_dir_now = "" if isinstance(stg, dict): cam_dir_now = stg.get("camera_direction", "") or "" # framing_scale enum SOT v1 (2026-05-15): shot path (stg dict) # 만 helper read. legacy verify path (staging_map entry 부재 — # producer 가 shot 단위 staging 미보유) 는 expected_close # default False (Gate 4 legacy pass-through). cam_dir_now 변수 # 는 보존 — 4 번 drift check 의 camera_direction_hash 계산 # 용 (line 1755). expected_close = False if isinstance(stg, dict): expected_close = ( get_framing_scale_or_raise( stg, where=f"detail_steps.verify_completion S{si}_Shot{shi}", ) == FRAMING_CLOSE ) expected_validator = ( OWNED_VALIDATOR_CLOSE_SKIP if expected_close else OWNED_VALIDATOR_FULL ) if sentinel.get("validator") != expected_validator: sentinel_drifted.append((si, shi, "validator_type")) continue # 2) round 5 BLOCKING 3: t2i_prompt_hash 일치. v_t2i_now = var.get("t2i_prompt", "") if sentinel.get("t2i_prompt_hash") != compute_t2i_prompt_hash(v_t2i_now): sentinel_drifted.append((si, shi, "t2i_prompt_hash")) continue # 3) owned_hash 일치 (close-skip 도 검증 — round 3 #4). if sentinel.get("owned_hash") != compute_owned_hash(expected_owned): sentinel_drifted.append((si, shi, "owned_hash")) continue # 4) camera_direction_hash 일치. if sentinel.get("camera_direction_hash") != compute_camera_direction_hash( cam_dir_now ): sentinel_drifted.append((si, shi, "camera_direction_hash")) total = len(scenes) missing_msgs: List[str] = [] # Wave 6 BLOCKING fix: loader contract violations 를 missing 으로 surface. if loader_violations: missing_msgs.extend(loader_violations) if failed_indices: missing_msgs.append( f"{len(failed_indices)}/{total} scene/shot result(s) have empty " f"t2i_variations: {failed_indices[:5]}" ) if contract_violations: missing_msgs.append( f"{len(contract_violations)} owned contract_violation: " f"{contract_violations[:5]}" ) if sentinel_drifted: # I4 (R3-I1 mirror): production literal `sentinel_drifted` 포함 — # test substring assert 와 정확 일치. 옛 "sentinel drift/missing" # 표현은 R3-I1 literal contract 위반. missing_msgs.append( f"{len(sentinel_drifted)} owned_validation sentinel_drifted: " f"{sentinel_drifted[:5]}" ) # G4.1 Phase 6 Task 18: card drift surfaced as partial signal (G3.2 # sentinel 패턴 mirror). # 2026-08-11 (FINDING 8 일반화 — '사랑했지만' 프로덕션 실측): # hash-only card drift 는 차단하지 않는다. scene_detail 완료 이후 # 재생성되는 하류 소스(shot_dependency_t2i·world/state 류)의 LLM # 재산출을 verify 가 live cp 로 재계산해 생기는 가짜 드리프트로, # force 재실행이 하류 재생성을 다시 유발해 영원히 재드리프트한다 # (순환 실측: 108장 → force → 71장). 저장 카드는 build 시점 소스와 # 정합. 구조 손상(missing/shape/recompute_failed)만 차단 신호. _card_structural = [t for t in card_drifted if t[2] != "hash"] _card_hash_only = [t for t in card_drifted if t[2] == "hash"] if _card_hash_only: logger.warning( "verify_completion: %d render_prompt_card hash-only drift — " "build 이후 하류 재산출로 인한 가짜 드리프트, 차단하지 않음 " "(예: %s)", len(_card_hash_only), _card_hash_only[:3], ) if _card_structural: missing_msgs.append( f"{len(_card_structural)} render_prompt_card drift/missing: " f"{_card_structural[:5]}" ) if missing_msgs: # Wave 6 iter2 fix: loader contract violation 은 전체 batch 무효 → # missing (partial 보다 강한 신호). 그 외는 기존 partial/missing 분기. # G4.1: card-missing 이 모든 result 에 걸쳐 발생한 케이스 (v15 cp # 잔존) 은 force escalate 가 필요 → missing 등급. partial card drift # (일부) 는 partial 등급 유지. _all_card_missing = bool(_card_structural) and all( tag == "missing" for _si, _shi, tag in _card_structural ) and len(_card_structural) >= total severity = ( "missing" if loader_violations or len(failed_indices) >= total or _all_card_missing else "partial" ) # Block B B8 (plan v2.1.3 / spec V5 §4.3): origin 분류. # 우선순위 — loader/contract/structural 위반 (contract_drift) > # hash drift (invariant_drift) > default (contract_drift safer). _STRUCTURAL_TAGS = {"missing", "shape", "validator_type", "recompute_failed", "self_hash"} _has_structural_sentinel = any( tag in _STRUCTURAL_TAGS for _si, _shi, tag in sentinel_drifted ) _has_structural_card = any( tag in _STRUCTURAL_TAGS for _si, _shi, tag in _card_structural ) if loader_violations or contract_violations: origin = "contract_drift" elif _has_structural_sentinel or _has_structural_card: origin = "contract_drift" elif sentinel_drifted or _card_structural: origin = "invariant_drift" # hash-only drift else: origin = "contract_drift" # failed_indices only — safer default return CompletionReport( is_complete=False, missing=missing_msgs, severity=severity, metadata={ "total_results": total, "failed_count": ( len(failed_indices) + len(contract_violations) + len(sentinel_drifted) + len(_card_structural) ), "failed_indices": failed_indices, "contract_violations": contract_violations, "sentinel_drifted": sentinel_drifted, "card_drifted": card_drifted, "loader_violations": loader_violations, }, origin=origin, ) _clean_meta: Dict[str, Any] = { "total_results": total, "failed_count": 0, "failed_indices": [], } if _card_hash_only: # Codex 합의 BLOCK3 (2026-08-12): 무시한 hash-only 드리프트를 # 감사 기록으로 남긴다 — 이전엔 warning 로그 한 줄뿐이었다. _clean_meta["ignored_card_hash_drift"] = _card_hash_only return CompletionReport( is_complete=True, missing=[], severity="clean", metadata=_clean_meta, ) def _attempt_owned_redraw_repair( self, *, variation, original_t2i, owned, judge_violations, cam_dir, result, name_by_short_id, check_prompts_fn, si, shot_idx, visible_entities=None, prop_term_map=None, retried_model=False, ): """FINDING 7 (e2e-bughunt-v1): owned redraw violation 1-attempt atomic repair. owned judge 가 검출한 redraw_violation evidence 로 repair LLM 을 1회 호출 — 위반 t2i_prompt 를 owned redraw → anchor/omit 으로 좁게 rewrite. 후보를 owned merge + owned judge + ID contract + visible_entities contract 로 전부 재검증하고, **모두 통과할 때만** variation 에 적용한다 (atomic). 어느 하나라도 실패하면 원본 variation 을 보존하고 None 을 반환 → caller 가 기존 대로 contract_violation 마킹. retry budget = 1 — repair LLM 은 1회만 호출, 무한 루프 없음. 본 메서드는 non-close + owned 1+ 분기에서만 호출된다 (close framing 은 owned redraw 불가, owned 부재는 violation 불가). Args: variation: ``result["t2i_variations"]`` 안의 대상 variation dict (성공 시 in-place 갱신). original_t2i: repair 전 t2i_prompt. owned: chain_bg owned 객체 목록. judge_violations: 1차 ``run_owned_judge`` 결과 violations (verdict 포함). cam_dir: shot 카메라 방향. result: scene_detail shot 결과 dict (재검증 validator 입력). name_by_short_id: ``ctx.name_by_short_id`` (visible validator 입력). check_prompts_fn: ``_analyze_one`` 의 ``_check_prompts`` closure. si / shot_idx: 로깅용. Returns: ``(repaired_t2i, repaired_merged_usage, repaired_violations)`` — 전부 재검증 통과 시. ``variation["t2i_prompt"]`` / ``["owned_object_usage"]`` 도 함께 in-place 갱신된다. ``None`` — repair 실패/revert. 원본 variation 불변. """ from app.core.steps._owned_helpers import ( has_redraw_violation, merge_owned_object_usage, reconcile_owned_prop_namespace_overlap, ) from app.core.steps._owned_judge import run_owned_judge from app.core.steps._owned_repair import run_owned_repair from app.core.visible_entities_validator import ( validate_visible_entities_contract, ) redraw_evidence = [ v for v in judge_violations if isinstance(v, dict) and v.get("verdict") == "redraw_violation" ] if not redraw_evidence: # 방어 — caller 가 has_redraw_violation True 로만 진입했어야 함. return None where = f"scene_detail owned_repair S{si}_Shot{shot_idx}" # 1) repair LLM 1-call. best-effort recovery — repair LLM 경계의 모든 # 실패 (AppError / LLM·runtime 계열 exception) 를 catch 후 revert. # 원본은 어차피 contract_violation/partial 행이므로 repair 자체 실패가 # 상태를 crash 로 악화시키면 안 된다 (Codex NARROW amend 1). silent # 아님 — logger.error 로 surface, 원본 contract_violation 은 caller 가 # 아래 `if has_redraw_violation(violations)` 분기로 유지. broad except # 는 repair LLM 경계에만 한정 — 원본 1차 owned judge path (else 분기, # 본 메서드 밖) 는 기존 fail-fast 유지. # 수리는 가벼운 모델(gpt-mini)이 맡는다. 실패하면 **같은 방식으로 다시 # 묻지 않고 더 나은 모델로 한 번 더** 시도한다 — 같은 모델에 같은 식으로 # 다시 물으면 같은 답이 나올 확률이 높다(2026-08-07 사용자 지시). # # 실제로 겪은 실패: 수리본에 장소 번호(`L07`)가 섞여 ID 규칙에 걸려 # 되돌려졌고, 그 한 컷 때문에 256개 중 255개가 정상인데도 단계 전체가 # "일부만 완료"로 남아 다음 단계가 막혔다. _RETRY_MODEL = "gpt" # analysis_sub 기본(gpt-mini)보다 상위 _TAG = "scene_detail_owned_repair" repair_out = None _upper = {**(self.project_config or {}), _TAG: {"model": _RETRY_MODEL}} # 검증 거부로 되돌아온 재호출이면 곧장 상위 모델만 쓴다. _plan = (_upper,) if retried_model else (self.project_config, _upper) for attempt, pc in enumerate(_plan): try: repair_out = run_owned_repair( t2i_prompt=original_t2i, owned=owned, redraw_violations=redraw_evidence, camera_direction=cam_dir, call_structured_fn=call_structured, project_config=pc, opik_metadata=self.build_opik_metadata( extra_tags=["owned_repair"] + ( ["owned_repair_retry"] if attempt else []), extra_metadata={"scene_index": si, "shot_index": shot_idx, "attempt": attempt + 1}, ), ) break except Exception as exc: # repair LLM 경계 best-effort recovery. if attempt == 0 and len(_plan) > 1: logger.warning( "%s: 1차 수리 실패 — 상위 모델(%s)로 재시도: %s", where, _RETRY_MODEL, exc) continue logger.error("%s: owned repair 재시도까지 실패 — revert: %s", where, exc) return None if repair_out is None: return None cand_t2i = repair_out["t2i_prompt"] cand_echo_raw = repair_out["owned_object_usage"] # 2) owned merge 재검증 — unknown token 등 contract 위반이면 repair 폐기. # merge 는 pure helper (LLM 호출 아님) — 선언된 AppError contract 만 catch. try: cand_merged = merge_owned_object_usage( cand_echo_raw, owned, is_close_framing=False, where=where, ) except AppError as exc: logger.warning("%s: repaired owned_object_usage merge 실패 — revert: %s", where, exc) return None # 3) owned judge 재검증 — repair 후보 judge 호출도 LLM 경계 (LLM 호출 + # schema 검증) 이므로 모든 exception 을 catch 후 revert (Codex NARROW # amend 1 — repair 검증 단계의 judge 실패가 원본 partial 을 crash 로 # 악화시키면 안 됨). 통과 후 여전히 redraw 면 repair 실패로 revert. try: cand_violations = run_owned_judge( t2i_prompt=cand_t2i, owned=owned, owned_object_usage=cand_merged, camera_direction=cam_dir, call_structured_fn=call_structured, project_config=self.project_config, opik_metadata=self.build_opik_metadata( extra_tags=["owned_judge", "owned_repair_verify"], extra_metadata={"scene_index": si, "shot_index": shot_idx}, ), # judge v5: repair 재검증도 동일한 shot-intent 증거 사용. shot_intent=( result.get("representative_moment") or result.get("still_frame_prompt") or "" ), ) except Exception as exc: # repair 검증 judge 경계 best-effort recovery. logger.error("%s: repair 후보 owned judge 재검증 실패 — revert: %s", where, exc) return None # Task 14 Wave 2 — same deterministic reclass applied on the repair- # verify path so that an honest anchor mention (e.g. owned="map" on # a visible "P06=종이 지도" shot whose prompt still names P06) doesn't # spuriously fail the repair gate. visible_entities / prop_term_map # may be None when the caller is a legacy path that didn't thread # them — helper short-circuits in that case (G10a). cand_violations = reconcile_owned_prop_namespace_overlap( violations=cand_violations, owned_object_usage=cand_merged, t2i_prompt=cand_t2i, visible_entities=visible_entities or [], prop_term_map=prop_term_map, ) if has_redraw_violation(cand_violations): logger.warning("%s: repair 후에도 redraw_violation 잔존 — revert", where) return None # 4) ID + visible_entities contract 재검증 — 후보를 임시 swap 후 기존 # validator 호출. committed flag + finally 로, commit 전 어떤 경로로 # 빠져나가도 t2i_prompt swap 이 남지 않도록 revert 보장. _orig_prompt = variation["t2i_prompt"] variation["t2i_prompt"] = cand_t2i committed = False try: id_bad = check_prompts_fn(result) if id_bad: # ★수리본이 **검증에서** 거부된 경우도 다른 모델로 한 번 더 # (2026-08-07). LLM 호출은 성공했으므로 위 재시도 루프는 # 타지 않는다 — 실제로 겪은 실패가 정확히 이 경로였다 # (수리본에 `L07` 이 섞여 ID 규칙 위반 → 곧장 revert → # 그 한 컷 때문에 단계 전체가 막힘). if not retried_model: logger.warning( "%s: 수리본 ID 위반 %s — 상위 모델(%s)로 재수리", where, sorted(id_bad), _RETRY_MODEL) variation["t2i_prompt"] = _orig_prompt return self._attempt_owned_redraw_repair( result=result, variation=variation, si=si, shot_idx=shot_idx, owned=owned, judge_violations=judge_violations, cam_dir=cam_dir, original_t2i=original_t2i, check_prompts_fn=check_prompts_fn, name_by_short_id=name_by_short_id, visible_entities=visible_entities, prop_term_map=prop_term_map, retried_model=True, ) logger.warning( "%s: 재수리본도 ID 위반 %s — revert", where, sorted(id_bad)) return None validate_visible_entities_contract(result, name_by_short_id) # 5) 전부 통과 — atomic commit. owned_object_usage 는 정상 flow 와 # 동일하게 merged 형태로 variation 에 저장. variation["owned_object_usage"] = cand_merged committed = True logger.info("%s: owned redraw repair 성공 — %d redraw_violation 해소", where, len(redraw_evidence)) return (cand_t2i, cand_merged, cand_violations) except AppError as exc: logger.warning("%s: repaired prompt visible/id 재검증 실패 — revert: %s", where, exc) return None finally: if not committed: variation["t2i_prompt"] = _orig_prompt def _analyze_one(self, seg, shot_info, ctx, system, schema, corrections=None): si = seg.get("scene_index", 0) scene_text = seg.get("text", "") all_visible = ctx.scene_visible.get(si, []) summary = ctx.summaries.get(si, "") # shot_director 결과가 있으면 shot별 VE 사용 (변형 확정됨) shot_idx = shot_info.get("shot_index") if shot_info else None if shot_idx is not None and (si, shot_idx) in ctx.shot_director_ve: visible = ctx.shot_director_ve[(si, shot_idx)] else: visible = all_visible # outlook 조회 맵: outlook_id(O01) + name 양쪽으로 키잉 outlook_id_to_info: Dict[str, tuple] = {} if ctx.outlook_data: for ol in ctx.outlook_data.get("outlooks", []): osid = ol.get("outlook_id") or ol.get("short_id", "") oname = ol.get("name", "") entry = (osid, oname) if osid: outlook_id_to_info[osid] = entry if oname: outlook_id_to_info[oname] = entry # 이름으로도 조회 가능 # 현재 씬의 아웃룩만 수집 (전체 씬 X → 교차 배정 방지) scene_outlooks: Dict[str, list] = {} # cid → [(osid, oname), ...] if ctx.outlook_data: for sa in ctx.outlook_data.get("scene_assignments", []): if sa.get("scene_index") != si: continue for a in sa.get("assignments", []): cid = a.get("character_id", "") # outlook_id 또는 outlook_name 중 있는 것으로 조회 osid_key = a.get("outlook_id") or a.get("outlook_name", "") if cid and osid_key and osid_key in outlook_id_to_info: entry = outlook_id_to_info[osid_key] if cid not in scene_outlooks: scene_outlooks[cid] = [] if entry not in scene_outlooks[cid]: scene_outlooks[cid].append(entry) # 씬 아웃룩 전체 사용 — LLM이 shot description 기반으로 선택 visible_chars = [sid for sid in visible if sid.startswith("C")] # 엔티티 목록 (bare ID만) entity_block_lines = [] for sid in visible: name = "" for etype in ["characters", "locations", "props"]: for e in ctx.entities.get(etype, []): if e.get("short_id") == sid: name = e.get("name", "") break entity_block_lines.append(f" {sid} ({name})" if name else f" {sid}") # 변형 관계 감지 (이름 기반) — 같은 인물의 변형 캐릭터 쌍 찾기 char_id_name: Dict[str, str] = {} for sid in visible_chars: for e in ctx.entities.get("characters", []): if e.get("short_id") == sid: char_id_name[sid] = e.get("name", "") break # 괄호 앞 base name 추출 + 전체 엔티티에서 비교 def _base_name(name: str) -> str: return re.split(r'\s*[\(（]', name)[0].strip() # visible 이외 캐릭터도 포함하여 base_name 그룹핑 all_char_names: Dict[str, str] = {} # sid → name (전체) for e in ctx.entities.get("characters", []): sid = e.get("short_id", "") if sid: all_char_names[sid] = e.get("name", "") # base_name → [sid, ...] 그룹 base_groups: Dict[str, list] = {} for sid, name in all_char_names.items(): bn = _base_name(name) if bn: base_groups.setdefault(bn, []).append((sid, name)) # visible 캐릭터 중 같은 base_name 그룹에 속하는 쌍 찾기 variant_pairs: list = [] # [(base_sid, variant_sid, base_name, variant_name)] for bn, members in base_groups.items(): visible_members = [(sid, name) for sid, name in members if sid in char_id_name] if len(visible_members) < 2: continue # short_id 순서가 빠른 것(원본)을 base로 visible_members.sort(key=lambda x: x[0]) base_sid, base_nm = visible_members[0] for var_sid, var_nm in visible_members[1:]: variant_pairs.append((base_sid, var_sid, base_nm, var_nm)) # 아웃룩 선택지 — 1개면 선택 불필요 표시, 2개면 선택지 제공 outlook_lines = [] fixed_outfits: Dict[str, str] = {} # cid → osid (1개일 때 자동 배정용) for cid in visible_chars: outlooks = scene_outlooks.get(cid, []) if not outlooks: # ★★이 씬에 아웃룩 배정이 **하나도 없는** 인물 (2026-09-17 컨트리로드). # 의상 단계가 회상·모니터 속 인물 등을 비워 둔다. 줄을 안 적으면 프롬프트 # 규칙(허용 짝이 없으면 C##O00)과 아래 허용 짝 검사가 부딪혀 재시도 → # 강제 제거로 인물이 그림 문장에서 사라진다. O00 은 이미지 단계에서 # 인물 기본 참조를 그대로 붙인다(scene_reference_service). # ★`_derive_outlook_pairs_for_shot` 에도 같은 규칙 — 한쪽만 고치면 지문이 어긋난다. fixed_outfits[cid] = "O00" outlook_lines.append(f" {cid} 아웃룩: 없음 (이 씬에 배정된 아웃룩 없음 — {cid}O00)") # O00 (Null Outlook) — 비인간형 캐릭터, 아웃룩 불필요 elif len(outlooks) == 1 and outlooks[0][0] == "O00": fixed_outfits[cid] = "O00" outlook_lines.append(f" {cid} 아웃룩: 없음 (비인간형/변형체 — 아웃룩 불필요, bare ID 사용)") elif len(outlooks) == 1: fixed_outfits[cid] = outlooks[0][0] outlook_lines.append(f" {cid} 아웃룩: {outlooks[0][0]}({outlooks[0][1]}) [확정]") elif len(outlooks) >= 2: # O00이 포함된 경우 O00 제외하고 선택지 제공 non_null = [(osid, oname) for osid, oname in outlooks if osid != "O00"] if len(non_null) == 1: fixed_outfits[cid] = non_null[0][0] outlook_lines.append(f" {cid} 아웃룩: {non_null[0][0]}({non_null[0][1]}) [확정]") elif non_null: opts = ", ".join(f"{osid}({oname})" for osid, oname in non_null) outlook_lines.append(f" {cid} 아웃룩 선택지: {opts} [variation별로 선택]") else: # 전부 O00뿐 fixed_outfits[cid] = "O00" outlook_lines.append(f" {cid} 아웃룩: 없음 (비인간형 — bare ID 사용)") from app.core.config import settings # v4: shot별 variation 수 (ENV: SHOT_VARIATION_COUNT) if shot_info: var_count = settings.shot_variation_count shot_idx = shot_info.get("shot_index", 0) else: shots = ctx.scene_shots_map.get(si, []) var_count = len(shots) if shots else settings.scene_variation_count shot_idx = None # Phase 2: essence (Phase 1b consumer) + chain_bg_guide prepend # legacy 흐름 (shot_info=None, scene 단위 분석)에서는 prepend skip — (si, 1) 우연 매핑 방지 if shot_info is None: prepend_blocks = "" else: shi = shot_info.get("shot_index", shot_info.get("_shot_index", 1)) # framing_scale enum SOT v1 (2026-05-15): staging_map 에서 dict 조회 # 후 helper (`get_framing_scale_or_raise`) 로 framing_scale enum read # → close 분류 bool 도출. cam_dir_for_block variable 은 보존 — # _g41 + owned_judge callsite 에서 camera_direction hash 계산 / arg # 전달 용도. staging_for_block = ctx.staging_map.get(f"{si}_{shi}") or {} cam_dir_for_block = staging_for_block.get("camera_direction", "") if isinstance(staging_for_block, dict) else "" is_close_for_block = ( get_framing_scale_or_raise( staging_for_block, where=f"detail_steps.phase2_prepend_caller S{si}_Shot{shi}", ) == FRAMING_CLOSE ) prepend_blocks = _build_phase2_prepend_blocks( si=si, shi=shi, essence_by_shot=ctx.essence_by_shot, chain_bg_guide_by_shot=ctx.chain_bg_guide_by_shot, chain_bg_camera_meta_by_shot=ctx.chain_bg_camera_meta_by_shot, chain_bg_owned_by_shot=ctx.chain_bg_owned_by_shot, shot_essence_enabled=settings.shot_essence_enabled, chain_bg_guide_enabled=settings.chain_bg_guide_enabled, chain_bg_camera_meta_enabled=settings.chain_bg_camera_meta_enabled, is_close_framing=is_close_for_block, ) user_prompt = prepend_blocks + ( f"씬 {si} 요약: {summary}\n\n" ) # v4: shot 정보 if shot_info: user_prompt += ( f"[분석 대상 Shot]\n" f" Shot {shot_info['shot_index']}: {shot_info.get('description', '')}\n" ) beat_idx = shot_info.get("based_on_beat") if beat_idx: beat_data = ctx.beats_by_scene.get(si, {}).get(beat_idx) if beat_data: user_prompt += ( f" Beat #{beat_idx} [{beat_data.get('change_type', '')}]: " f"{beat_data.get('before_state', '')} → {beat_data.get('after_state', '')}\n" ) else: user_prompt += f" Beat 근거: #{beat_idx}\n" # shot별 dependency (앞쪽 연관 shot) dep_for_shot = next( (d for d in ctx.dependencies if d.get("scene_index") == si and d.get("shot_index") == shot_idx), {} ) # schema: location_refs maxItems=1 dep_loc_refs = dep_for_shot.get("location_refs", [])[:1] dep_char_refs = dep_for_shot.get("character_refs", []) if dep_loc_refs or dep_char_refs: user_prompt += " [앞쪽 연관 Shot — 연속성 유지]\n" _ru_labels = { "zoom_in_detail": " [zoom_in_detail — 같은 시공간, 카메라 확대/이동만]", "exact_background": " [exact_background — 같은 방, 다른 순간]", "atmosphere_reference": " [atmosphere_reference — 톤만 참고]", } for ref in dep_loc_refs: _ru = ref.get("ref_usage", "") _ru_label = _ru_labels.get(_ru, f" [{_ru}]" if _ru else "") user_prompt += f" 배경 참조: S{ref['scene_index']}_Shot{ref['shot_index']}{_ru_label}\n" for ref in dep_char_refs[:2]: user_prompt += f" 인물 참조: S{ref['scene_index']}_Shot{ref['shot_index']}\n" user_prompt += ( "[중요 — 순간 고정]\n" "t2i_prompt는 위 Shot description이 묘사하는 정확한 순간만 캡처하세요.\n" "씬의 다른 시점이나 다른 장면은 포함하지 마세요.\n" "이 shot에서 카메라에 찍히는 것만 영어로 묘사하세요.\n\n" ) # ── 결함 A fix: forward-look continuity ──────────────────────── # 현재 shot이 후속 zoom_in_detail의 source(앞 샷)이면, 후속 close-up 대상 # 영역이 현재 frame에 visible 위치로 잡혀야 한다. # _compute_forward_zoom_targets — module-level pure helper (단위 테스트 가능). # 일반화 메커니즘 — 시나리오 의존 0. forward_zoom_targets = _compute_forward_zoom_targets( ctx.dependencies, ctx.shot_scenes_map, si, shot_idx, ) if forward_zoom_targets: _FZT_CAP = 6 # 4+ 케이스도 흔하지 않으나 prompt bloat 미미 if len(forward_zoom_targets) > _FZT_CAP: logger.warning( "scene_detail S%s_Shot%s: %d forward zoom targets — capped to %d", si, shot_idx, len(forward_zoom_targets), _FZT_CAP, ) user_prompt += ( "[중요 — 후속 zoom_in_detail 대상 영역 (forward-look continuity)]\n" "이 shot 이후에 후속 shot이 이 frame의 특정 영역을 close-up할 예정입니다. " "후속 close-up이 이 frame에서 자연스럽게 zoom-in 가능하도록, 그 대상 영역이 " "현재 shot의 frame에 visible 위치로 잡히고 가려지지 않게 배치하세요. " "특히 죽은/의식불명/움직일 수 없는 대상이면 위치/자세를 frame 안에 명확히 노출시키세요.\n" ) _KEEP_CAP = 5 # 후속 keep_elements 가 통상 < 5. 초과 시 warning + cap. for fzt in forward_zoom_targets[:_FZT_CAP]: # CLAUDE.md 절대 규칙: LLM 컨텍스트 truncation 금지 (전문 inject). desc = fzt["description"] user_prompt += ( f"- 후속 S{fzt['scene_index']}_Shot{fzt['shot_index']} 묘사: {desc}\n" ) keeps = fzt.get("keep_elements") or [] if keeps: if len(keeps) > _KEEP_CAP: logger.warning( "scene_detail S%s_Shot%s forward zoom keep_elements=%d > cap=%d — capping", si, shot_idx, len(keeps), _KEEP_CAP, ) # Area D-next-min — dict only path. shot_dependency_t2i # v7 producer SOT 가 list[{label, kind∈{environment, # static_prop}}] dict shape 강제 (schema_version=3 # bump). 옛 str + description key 폐기. character / # person 묘사는 keep_elements 가 다루지 않음 (별도 # layer: scene_consistency / character_state_variant / # semantic_contract_router 책임). keep_summary = "; ".join( k["label"] for k in keeps[:_KEEP_CAP] ) user_prompt += f" 보존 요소: {keep_summary}\n" user_prompt += "\n" # scene_consistency: 교차 샷 고정 요소 주입 _shot_idx_for_fe = shot_info.get("shot_index", shot_info.get("_shot_index", 1)) try: _shot_idx_for_fe_int = int(_shot_idx_for_fe) except (TypeError, ValueError): _shot_idx_for_fe_int = None fixed_for_shot = [] if _shot_idx_for_fe_int is not None and si in ctx.fixed_elements_by_scene: for fe in ctx.fixed_elements_by_scene[si]: # LLM이 string으로 반환할 경우 대비 — int 캐스팅 _raw = fe.get("applies_to_shots", []) or [] _applies: set = set() for _v in _raw: try: _applies.add(int(_v)) except (TypeError, ValueError): continue if _shot_idx_for_fe_int in _applies: fixed_for_shot.append(fe) if fixed_for_shot: _type_labels = { "character_state": "인물 상태", "environment_state": "환경 상태", "persistent_prop": "고정 소품", } # character_name → C## 역매핑 (visible 엔티티에서 이름 매칭) _name_to_cid: Dict[str, str] = {} for _eid in visible: if _eid.startswith("C"): for _e in ctx.entities.get("characters", []): if _e.get("short_id") == _eid: _name_to_cid[_e.get("name", "")] = _eid break user_prompt += "[교차 샷 고정 요소]\n\n" for fe in fixed_for_shot: fe_type = fe.get("element_type", "") label = _type_labels.get(fe_type, fe_type) header = f" [{label}] {fe.get('element_id', '')}" char_name = fe.get("character_name", "") if char_name: matched_cid = _name_to_cid.get(char_name, "") if matched_cid: header += f" ({char_name} = {matched_cid})" else: header += f" ({char_name})" user_prompt += f"{header}:\n {fe.get('description', '')}\n\n" # v0.5.10: location_consistency 주입 비활성화. # 이전에는 location_visuals_by_id를 t2i_prompt에 강제 삽입했으나, # shot_dependency_t2i의 이전 샷 참조 + ignore/keep label 방식이 이미 # 더 효과적으로 씬 간 배경 일관성을 보장함. # location_consistency 텍스트 주입은 프롬프트 비대화(v5 +40% 단어 수)와 # T2I 모델 혼란을 유발해 오히려 품질을 저하시켰음. if entity_block_lines: user_prompt += f"[사용 가능한 엔티티]\n" + "\n".join(entity_block_lines) + "\n\n" else: user_prompt += "[사용 가능한 엔티티]\n 없음 — 이 씬에 할당된 엔티티가 없으므로 엔티티 ID(C##, L##, P## 등)를 사용하지 마세요. outfit_assignments도 비워두세요. 씬 텍스트에 묘사된 장면을 그대로 영어 프롬프트로 작성하세요.\n\n" if outlook_lines: user_prompt += "인물별 아웃룩 선택지 (각 variation의 outfit_assignments에 인물별 1개씩 선택):\n" + "\n".join(outlook_lines) + "\n\n" # v4: 촬영 기법 + 조명/색감 if shot_info: # shot_staging 결과가 있으면 우선 사용 staging_key = f"{si}_{shot_info.get('shot_index', shot_info.get('_shot_index', 1))}" staging = ctx.staging_map.get(staging_key) if staging: cam_dir = staging.get("camera_direction", "") light = staging.get("lighting_mood", "") bg_elems = staging.get("key_bg_elements", []) perspective = staging.get("perspective", "") pov_char = staging.get("pov_character", "") user_prompt += ( "[촬영 감독(DP) 연출 지시 — 반드시 t2i_prompt에 반영하세요]\n" f"카메라 관점(POV): {perspective}\n" ) if pov_char: user_prompt += f"POV 인물: {pov_char} — 이 인물의 시점이므로 t2i_prompt에서 이 인물을 제외하세요. 이 인물의 참조 이미지도 사용하지 마세요.\n" perception = staging.get("perception_mode", "direct") if perception and perception != "direct": # C1 v1: helper-owned _PERCEPTION_GUIDES (detail_steps local map 제거). # get_perception_guide 2-stage check: unknown → AppError("perception_mode.unknown") / # direct → AppError("perception_mode.guide_not_applicable") (caller가 non-direct branch 안에서만 호출). _pguide = get_perception_guide(perception) user_prompt += f"지각 모드: {perception} — {_pguide}\n" # ★`카메라/프레이밍: {cam_dir}` 줄 제거 (2026-08-25). # # 이 값은 **카드에 이미 있다** — `render_strategy.camera_direction`. # 둘은 같은 `staging` dict 에서 온다 # (`_g41_staging = locals()["staging"]`, 카드 쪽은 # `cam_dir = staging_cam or shot_cam` 이라 오히려 더 완전하다). # 그런데 여기서 다시 넣으면 user 메시지 **뒤쪽**에 # 「반드시 t2i_prompt에 반영하세요」를 달고 한 번 더 나가고, # 그 recency 지시를 system.md 의 어떤 문구도 못 이긴다. # # 실측(Opik span 전수): t2i_prompt 로 샌 촬영 용어 12건 중 # **9건이 이 camera_direction 에 있던 낱말**이었다. 그 줄만 # 빼고 같은 payload 로 ABBA A/B (source-positive 4컷×6회): # A 중복 있음 12컷 · 등급낱말 6 · 자리12 거리3 배치12 관계4 · 1,046자 # B 그 줄 뺌 12컷 · 등급낱말 3 · 자리12 거리4 배치12 관계4 · 953자 # 누출 절반 · 의미 네 축 손실 없음(거리·잘림은 오히려 증가). # # `조명/색감`(=`render_strategy.lighting_mood`)도 같은 중복이나 # 조명 왜곡 축은 현재 0건이라 관측된 해가 없다 — 손대지 않는다. user_prompt += f"조명/색감: {light}\n" if bg_elems: bg_lines = [] for e in bg_elems: bg_lines.append(_build_bg_element_line( element=e.get("element", ""), state=e.get("state", ""), camera_use=e.get("camera_use", ""), orientation=e.get("orientation", ""), directionality_class=e.get("directionality_class", ""), )) user_prompt += f"배경 핵심 요소:\n" + "\n".join(bg_lines) + "\n" char_angles = staging.get("character_angles", []) if char_angles: # 엔티티 미등록 인물(엑스트라)은 보통명사 주석 추가 _registered_names = set() for etype in ["characters", "locations", "props"]: for e in ctx.entities.get(etype, []): if e.get("name"): _registered_names.add(e["name"]) angle_lines = [] for a in char_angles: _cname = a.get('character', '') _angle = a.get('angle', '') _pose = a.get('body_pose', '') # Area #2 W5: v13 3 field labeling — gaze_direction_kind + gaze_target_id + subject_state # raw enum/ID label 그대로 노출 (code semantic translation 금지, Gate 1). _kind = a["gaze_direction_kind"] # required by v13 schema, KeyError = schema violation (Gate 4 fail-fast) _target_id = a.get('gaze_target_id') or '' # conditional nullable/omitted OK (schema oneOf) _state = a["subject_state"] # required by v13 schema, KeyError = schema violation (Gate 4 fail-fast) _gaze_str = f", gaze={_kind}" + (f"→{_target_id}" if _target_id else "") _state_str = f", state={_state}" if _cname and _cname not in _registered_names: angle_lines.append(f" {_cname} (엔티티 미등록 — t2i에서 보통명사로 묘사): {_angle}, {_pose}{_gaze_str}{_state_str}") else: angle_lines.append(f" {_cname}: {_angle}, {_pose}{_gaze_str}{_state_str}") user_prompt += ( "[인물-카메라 각도 + 시선 + 상태 — 절대 무시 금지, t2i_prompt에 반드시 반영]\n" "아래 각 인물의 카메라 대비 방향, 자세, 시선 방향, 신체 상태를 t2i_prompt에 영어로 구체적으로 명시하세요:\n" "- facing_camera → 'facing the camera' / back_to_camera → 'with back to camera'\n" "- profile_left → 'in left profile' / three_quarter_right → 'at three-quarter angle from the right'\n" "- gaze=camera → 'looking at the camera' / gaze=down → 'looking down' / gaze=up → 'looking up'\n" "- gaze=distant → 'gazing into the distance' / gaze=closed_eyes → 'with eyes closed'\n" "- gaze=off_screen → 'looking off-screen' / gaze=looks_at_character→C## → 'eyes fixed on [that character]'\n" "- gaze=looks_at_object→P##/B## → 'eyes fixed on [that prop/background element]'\n" "- state=alive → 평소 자세/표정 / state=unconscious → 'unconscious, slack and limp'\n" "- state=dead → 'lying motionless, lifeless' / state=severely_injured → 'visibly wounded, pained expression'\n" "인물이 2명 이상이면 각각의 방향, 시선, 상태를 따로 명시하세요.\n" "- 엔티티 미등록 인물은 C## ID 대신 보통명사 + demographic descriptor (인종/연령대/성별)로 묘사하세요. 인종은 visual_world_rules.region 이 명시한 ethnicity 그대로 carry — production code / prompt 어디에도 hardcoded ethnicity 어휘 list 를 강요하지 마세요 (시나리오/region 마다 다름, A-prime binding).\n" f"{chr(10).join(angle_lines)}\n" ) user_prompt += "\n" elif ctx.shot_types_block: # fallback: 기존 DB shot_type 목록 user_prompt += ( "[촬영 기법 선택] — 아래 기법 중 이 shot에 가장 적합한 것을 선택하여 t2i_prompt에 반영하세요:\n" + ctx.shot_types_block + "\n\n" ) user_prompt += ( "[카메라 포커싱]\n" "t2i_prompt에 카메라가 무엇에 초점을 맞추는지 반드시 명시하세요:\n" "- 인물의 특정 신체 부위/표정 (영어 명사구로 구체적 묘사 — gender / emotional label 은 본 beat 의 staging.subject_state 와 character_angles 에서 가져오기)\n" "- 본 shot 의 핵심 narrative 객체 (해당 beat 에서 등장하는 prop 또는 background element)\n" "- 두 인물 간 공간/시선 관계 (subject 간 frame 안 spatial tension)\n" "- 단순히 앵글만 쓰지 말고, 그 앵글로 무엇을 강조하는지 반드시 포함할 것\n\n" "[조명/색감 판단]\n" "beat의 상태 변화와 씬 분위기를 고려하여 t2i_prompt에 조명과 색감을 반영하세요:\n" "- 시간대(낮/밤/새벽/석양), 자연광/인공광, 색온도(따뜻한/차가운)\n" "- 색감은 beat의 정서와 씬 분위기에서 직접 derive하고, 특정 정서에 고정 색조를 대응시키도록 강요하지 마세요\n" "- 핵심 인물/물체에 대한 조명 방향(역광/사광/정면광)\n\n" ) else: # legacy: 씬별 모드 scene_shots_info = ctx.shot_scenes_map.get(si, []) if scene_shots_info: shot_desc_lines = [] for sh in scene_shots_info: shot_desc_lines.append( f" Shot {sh['shot_index']}: {sh['description']}" ) user_prompt += "[씬의 Shot 분석]\n" + "\n".join(shot_desc_lines) + "\n\n" shots = ctx.scene_shots_map.get(si, []) if shots: shot_lines = [] for vi, shot_t in enumerate(shots, 1): shot_idx_info = f" (Shot {shot_t.get('_shot_index', '')})" if shot_t.get('_shot_index') else "" shot_lines.append( f" variation {vi}{shot_idx_info}: {shot_t.get('name', '')} — {shot_t.get('llm_description', '')} (focus: {shot_t.get('focus', '')})" ) user_prompt += ( f"[촬영 감독 추천 기법] — 각 variation의 t2i_prompt에 이 촬영 기법의 카메라 앵글/구도를 반영하세요:\n" + "\n".join(shot_lines) + "\n\n" ) # 변형 관계 주의 블록 if variant_pairs: # shot_director가 VE를 확정한 경우: 간소화 (이미 올바른 형태만 visible에 포함) _has_shot_director = shot_idx is not None and (si, shot_idx) in ctx.shot_director_ve if _has_shot_director: vlines = ["[변형 관계 참고]"] vlines.append("아래 인물들은 같은 인물의 변형 관계입니다. 이 shot의 visible에 이미 올바른 형태만 포함되어 있습니다:") for base_sid, var_sid, base_name, var_name in variant_pairs: vlines.append(f" {var_sid}({var_name}) ← {base_sid}({base_name})의 변형") vlines.append("- visible에 있는 형태의 ID만 사용하세요.") vlines.append("- visible에 없는 형태의 ID는 절대 사용 금지.") else: # fallback: shot_director 없으면 기존 방식 vlines = ["[변형 관계 주의]"] vlines.append("아래 인물들은 같은 인물의 변형 관계입니다:") for base_sid, var_sid, base_name, var_name in variant_pairs: vlines.append(f" {var_sid}({var_name}) ← {base_sid}({base_name})의 변형") vlines.append("") vlines.append("각 t2i_prompt에서 선택한 순간에, 해당 인물이 어떤 형태인지 씬 텍스트에서 판단하세요.") vlines.append("- 변형/변신 이후 시점이면 반드시 변형 ID를 사용하세요. 원본 ID 사용 금지.") vlines.append("- 변형 전 시점이면 원본 ID만 사용하세요. 변형 ID 사용 금지.") vlines.append("- 같은 순간에 두 형태가 물리적으로 동시에 존재하는 경우에만(클론, 분신, 환각 등) 둘 다 사용 가능합니다.") vlines.append("- 시간 순서로 전환되는 변신/변형은 절대 한 프롬프트에 같이 넣지 마세요.") if shot_info: vlines.append("") vlines.append("[아웃룩 선택 규칙 — beat 상태 확인]") vlines.append("- beat의 상태 변화(before_state → after_state)를 확인하세요.") vlines.append("- 변신/변형 이전 순간이면 → 원본 캐릭터(C##)의 아웃룩을 outfit_assignments에 사용") vlines.append("- 변신/변형 이후 순간이면 → 변형 캐릭터(C##)의 아웃룩을 outfit_assignments에 사용") user_prompt += "\n".join(vlines) + "\n\n" user_prompt += ( f"[규칙]\n" f"- 인물+아웃룩은 복합 ID(C01O02)를 사용 — 합성 단계가 'the character from Image N'으로 자동 치환.\n" f"- 신체 부위 클로즈업/사진·거울 속 인물 등 system prompt가 명시한 예외 구도에서는 보통명사로 대체 (system 룰 참조).\n" f"- outfit_assignments에 각 인물이 이 variation에서 입는 아웃룩 ID를 지정 — t2i_prompt의 복합 ID와 같은 정보를 명시.\n" f"- t2i_variations 정확히 {var_count}개.\n" + (f"- 각 variation의 t2i_prompt는 해당 촬영 기법의 카메라 구도를 반영할 것.\n" if (shot_info or ctx.scene_shots_map.get(si)) else "") + f"\n씬 텍스트:\n{scene_text}" ) # G4.1 Phase 5 (R2-B3 5-point modify scope guard): RenderPromptCard # collection + build + user_prompt inject (points 1/2/3). 기존 변수 # (visible / scene_outlooks / fixed_outfits / staging / cam_dir_for_block / # forward_zoom_targets / fixed_for_shot / dep_loc_refs / dep_char_refs) # 만 read-only 참조. 새 변수 도출 0. from app.core.steps.render_prompt_card import ( build_render_prompt_card as _g41_build, compute_card_hash as _g41_hash, ) from app.core.config import settings as _g41_settings # outlook_pairs (List[Dict[str,str]]) — scene_outlooks/fixed_outfits 로부터. # builder 가 expected shape: [{"character_id": cid, "outlook_id": oid}, ...] # Codex iter 1 B4 carry — shot-specific contract: visible_chars (이 shot # 의 visible_entities character base) 안 character_id 만 통과. 같은 # scene 의 다른 shot 에만 등장하는 character 의 pair 는 본 shot 의 # render_prompt_card 에서 제거 — visible_entities 와 정합 (validator # Source 3b 의 fail-fast 가 정상 shot 을 false-fail 시키지 않도록). _visible_chars_for_pairs: set = { sid.split("O")[0] for sid in (visible or []) if isinstance(sid, str) and sid.startswith("C") } _g41_outlook_pairs: List[Dict[str, str]] = [] _g41_seen: set = set() for _g41_cid, _g41_ols in scene_outlooks.items(): if _g41_cid not in _visible_chars_for_pairs: continue # B4: shot-specific filter for _g41_osid, _g41_ in _g41_ols: if _g41_osid and (_g41_cid, _g41_osid) not in _g41_seen: _g41_outlook_pairs.append( {"character_id": _g41_cid, "outlook_id": _g41_osid} ) _g41_seen.add((_g41_cid, _g41_osid)) for _g41_cid, _g41_osid in fixed_outfits.items(): if _g41_cid not in _visible_chars_for_pairs: continue # B4: shot-specific filter if _g41_osid and (_g41_cid, _g41_osid) not in _g41_seen: _g41_outlook_pairs.append( {"character_id": _g41_cid, "outlook_id": _g41_osid} ) _g41_seen.add((_g41_cid, _g41_osid)) # shot-specific values: legacy (shot_info=None) flow 에서는 비어있음. # `_local` lookup via locals() — 변수명 변경 0 (R2-B3 scope guard). _g41_locals = locals() _g41_has_shot = bool(shot_info) _g41_staging = ( _g41_locals.get("staging") if _g41_has_shot else None ) # framing_scale enum SOT v1 (2026-05-15): _g41_staging.framing_scale enum # helper read. legacy / scene-level path (_g41_has_shot=False) 만 # default False (staging 자체 무관). _g41_cam_dir variable 제거 — 더 # 이상 정의 안 함 (regex 폐기 후 사용처 0). _g41_is_close = ( ( get_framing_scale_or_raise( _g41_staging, where=f"detail_steps._g41 S{si}_Shot{shot_idx}", ) == FRAMING_CLOSE ) if _g41_has_shot else False ) # G4.1 Wave 4 R4 B3: bg_id source = ctx.chain_bg_id_by_shot (loader # 가 background_prompt cp 의 backgrounds[bid] 의 bid 를 shot 매핑). # 옛 hardcode None 은 bg-on + non-close + ref-attached shot 에서도 # builder mode='not_applicable' 만 만들어 background reference 미사용 # 결함 발생. _g41_bg_id = None if _g41_has_shot and shot_idx is not None: _g41_bg_id = (ctx.chain_bg_id_by_shot or {}).get((si, shot_idx)) # Codex Important 1 fix — preserve None vs explicit [] distinction. # If bg_id is present but chain_bg_owned_by_shot has no entry for # (si, shot_idx), that is a loader contract violation (upstream # did not populate). raise instead of silently absorbing to []. _g41_bg_owned_raw = ctx.chain_bg_owned_by_shot.get((si, shot_idx)) if _g41_bg_id is not None and _g41_bg_owned_raw is None: raise AppError( code="step.contract_violation", message=( f"chain_bg_owned_by_shot missing entry for " f"(scene={si}, shot={shot_idx}) but " f"bg_id={_g41_bg_id!r} present — upstream loader " f"did not populate. NO silent fallback. Use explicit " f"[] for intentionally empty (no owned objects)." ), ) _g41_bg_owned = ( list(_g41_bg_owned_raw) if _g41_bg_owned_raw is not None else [] ) _g41_bg_camera_meta = ctx.chain_bg_camera_meta_by_shot.get( (si, shot_idx) ) _g41_bg_guide = ctx.chain_bg_guide_by_shot.get((si, shot_idx)) else: _g41_bg_owned = [] _g41_bg_camera_meta = None _g41_bg_guide = None _g41_background_mode_on = ( _g41_settings.background_mode in {"on", "floor_plan_anchored"} ) # continuity inputs — full lists (R1-I7: no truncation in card payload). _g41_fixed_elements = ( _g41_locals.get("fixed_for_shot") if _g41_has_shot else None ) or [] _g41_prev_refs: List[Dict[str, Any]] = [] if _g41_has_shot: for _r in (_g41_locals.get("dep_loc_refs") or []): _g41_prev_refs.append({ "scene_index": _r.get("scene_index"), "shot_index": _r.get("shot_index"), "ref_usage": _r.get("ref_usage", ""), "kind": "location", }) for _r in (_g41_locals.get("dep_char_refs") or []): _g41_prev_refs.append({ "scene_index": _r.get("scene_index"), "shot_index": _r.get("shot_index"), "ref_usage": _r.get("ref_usage", ""), "kind": "character", }) _g41_forward_zoom = ( _g41_locals.get("forward_zoom_targets") if _g41_has_shot else None ) or [] # G4.1 Wave 4 R4 B4: build_render_strategy() 의 staging 부재 fail-fast # 분기. shot path (shot_info dict) + staging 부재 → builder raise # (R1-I1: silent fallback / not_applicable promotion 금지). # legacy / scene-level path (shot_info=None) 만 staging_not_applicable # marker synthesize 허용 (의도적 not-applicable case). _g41_shot_info_for_card = shot_info if _g41_shot_info_for_card is None: # legacy: synthesize minimal shot_info marker so builder treats as # explicit not-applicable (scene-level path 는 staging 자체가 무관). _g41_shot_info_for_card = {"staging_not_applicable": True} # else: shot_info dict 그대로 forward — staging 부재 시 builder raise. # Point 1 (5-point modify scope guard 1/5): single-source helper. _g41_card_inputs = _collect_card_inputs( ctx=ctx, seg=seg, shot_info=_g41_shot_info_for_card, visible_entities=visible, # 기존 변수 outlook_pairs=_g41_outlook_pairs, staging_for_shot=_g41_staging, bg_id_for_shot=_g41_bg_id, bg_owned_for_shot=_g41_bg_owned, bg_camera_meta_for_shot=_g41_bg_camera_meta, bg_guide_for_shot=_g41_bg_guide, is_close_framing=_g41_is_close, background_mode_on=_g41_background_mode_on, fixed_elements_for_shot=_g41_fixed_elements, previous_shot_refs_for_shot=_g41_prev_refs, forward_zoom_targets_for_shot=_g41_forward_zoom, ) # Point 2 (5-point modify scope guard 2/5): card build + self-check. # build_render_prompt_card 가 내부에서 assert_card_shape 호출 — sibling # 호출 불요. 실패 시 AppError → 외부 try/except 가 None 반환 → retry path. _g41_builder_inputs = { k: v for k, v in _g41_card_inputs.items() if k != "ctx" } _g41_render_prompt_card = _g41_build(**_g41_builder_inputs) _g41_render_prompt_card_hash = _g41_hash(_g41_render_prompt_card) # Point 3 (5-point modify scope guard 3/5): user_prompt inject — first # block. spec §5.1 + R1-I4 card-wins precedence. canonical JSON # (sort_keys=True, ensure_ascii=False) — hash 계산과 동일. # G4.2 R2-I4: inject 직전 _card_metadata strip — LLM 입력은 envelope 7 # contract field 만 노출. _card_metadata (debug-only) 는 CP 에는 보존되지만 # inject JSON 에는 제외 (token budget + LLM confusion 방지). hash 격리는 # canonicalize_render_prompt_card() 가 이미 처리하므로 strip 은 inject path # 한정. _g41_card_for_inject = { k: v for k, v in _g41_render_prompt_card.items() if k != "_card_metadata" } _g41_card_block = ( "[RenderPromptCard v1]\n" + json.dumps( _g41_card_for_inject, sort_keys=True, ensure_ascii=False, separators=(",", ":"), ) + "\n\n" ) # G4.6 Wave A3 RC-H — entity_traits block prepend after RenderPromptCard. # block empty 면 빈 string return — 자연 skip. Rule X (variant trait # inject) + entity-aware silhouette policy 가 이 block 을 carry source # 로 사용. visible_entities 의 character base 가 prebuild map 에 없으면 # _build_entity_traits_block 가 fail-fast (silent bypass 금지). # Codex iter 1 B1 carry — ThreadPool worker 안 self.db query 금지, # ctx.name_by_short_id / traits_by_short_id (main thread prebuild) 사용. _g46_entity_traits_block = _build_entity_traits_block( visible, ctx.name_by_short_id, ctx.traits_by_short_id, ) # W21B-W7 W-B (2026-06-12): printed_prop visual-continuity anchor 주입 — # 이 shot 이 printed_prop 그룹 멤버면 anchor 의 printed_content / # physical_form / scale_contract 를 계약으로 전달. t2i 작문이 ref 와 # 모순되는 인쇄 내용·포스터화 스케일을 만들지 않게 한다 (설계 D-②). # C(zoom_continuity) 주입은 W-C 별도 wave — 여기서 다루지 않는다. _vca = self._vca_printed_prop_context() if _vca and shot_idx is not None: _vca_sids = _vca.get("members_by_shot", {}).get((si, shot_idx), []) _vca_lines = [] for _sid in _vca_sids: _a = _vca.get("by_prop", {}).get(_sid) if not _a: continue _vca_lines.append( f" {_sid}: printed/displayed content = {_a.get('printed_content', '')} | " f"physical form = {_a.get('physical_form', '')} | " f"scale contract = {_a.get('scale_contract', '')}" ) if _vca_lines: user_prompt += ( "[고정 소품 시각 계약 — visual continuity anchor]\n" "아래 소품의 인쇄/표시 내용과 실물 크기는 에피소드 전체에서 고정된 계약입니다. " "t2i_prompt 에서 이 소품을 묘사할 때 이 계약과 모순되는 내용(다른 인쇄 내용, " "포스터/문서 크기로의 확대 등)을 쓰지 마세요. 크기는 환경 대비 실물 크기를 유지하세요. " "인쇄된 글자의 자구는 고정하지 않습니다 — 구조(장소+인물 구성)만 지키세요.\n" + "\n".join(_vca_lines) + "\n\n" ) # P8 (2026-06-20): immobilized_subject 연속성 계약 주입 — 이 shot 이 # immobilized 인물 그룹 멤버면 그룹 공유 포즈/상태 계약(자세·위치·방향· # 소지/근접 소품)을 멤버 전샷에 동일 주입한다. 카메라/프레이밍은 잠그지 # 않는다 — 크롭은 자유, 물리 상태만 고정 (Codex 합의 ⑥). 별도 flag/loader. _vca_imm = self._vca_immobilized_subject_context() if _vca_imm and shot_idx is not None: # members_by_shot 은 group_id 를 가리킨다 (BLOCKING1 fix) — 같은 인물이 # 여러 씬 group 을 가져도 이 샷의 그룹 계약만 정확히 집는다. _imm_gids = _vca_imm.get("members_by_shot", {}).get((si, shot_idx), []) _imm_lines: list = [] for _gid in _imm_gids: _sa = _vca_imm.get("anchors_by_group", {}).get(_gid) if not _sa: continue _contract = str(_sa.get("shared_state_contract") or "").strip() if not _contract: # MINOR1 방어 — 빈 계약 무의미 주입 차단 continue _csid = _sa.get("character_short_id", "") _focus = (_sa.get("per_shot_visible_focus") or {}).get(str(shot_idx)) _line = f" {_csid} [{_sa.get('subject_state', '')}]: {_contract}" if _focus: _line += f" | 이 샷 프레임에 보이는 부분: {_focus}" _imm_lines.append(_line) if _imm_lines: user_prompt += ( "[부동 피사체 연속성 계약 — immobilized subject continuity]\n" "아래 인물은 같은 씬의 여러 컷에 걸쳐 스스로 움직이지 못하는 상태입니다. " "자세·신체 방향·위치, 그리고 쥐고 있거나 곁에 닿은 소품의 상태는 컷마다 동일해야 " "하는 고정 계약입니다. 카메라 앵글/프레이밍/크롭은 이 샷의 묘사대로 자유이나(가까이 " "잡아 일부만 보여도 됨), 이 물리적 상태/자세/소품 상태는 바꾸거나 새로 지어내지 마세요.\n" + "\n".join(_imm_lines) + "\n\n" ) user_prompt = _g41_card_block + _g46_entity_traits_block + user_prompt if corrections: # ★직전 시도가 걸린 계약 위반을 **그대로** 짚어 다시 묻는다 (2026-09-18 컨트리로드). # 같은 입력으로 다시 물으면 같은 답이다 — 손 클로즈업 샷들이 두 번 모두 인물 # 표식을 빼서 base_id_missing 으로 섰다. 규칙은 system·카드 그대로다. user_prompt = ("[수정 요청] 직전 답이 계약을 어겼다:\n" + "".join(f"- {c}\n" for c in corrections) + "\n" + user_prompt) try: # Group 1 #1 (2026-05-02): composite ID(C##O##) 가 default — base C## # (또는 L##/P##) 가 allowed_set 에 있으면 통과. outlook 부분 (O##) 도 # scene allowed pair (cid, oid) 로 별 검사 (Codex review IMPORTANT — # invalid outlook 이 silent 로 downstream 에 흘러가 wrong/fallback ref # 에 연결되는 회귀 차단). fail-fast 원칙. allowed_set = set(visible) # base ID 만 (C01, L02, P03 등) # ★배경 표식(`L143B01` = background_binding.bg_id)의 앞부분을 장소 표식으로 읽지 않는다 # (2026-09-18 컨트리로드). 팩은 `[L…: …]` 라벨을 허용하는데 이 규칙이 `L143` 만 떼어 # VE 밖이라 보고 재시도·강제 제거(`\b` 라 실제로는 못 지움)·owned 수리본 폐기를 불렀다. # 같은 착오를 visible_entities_validator 는 이미 고쳤다(required_refs kind 분기). _ve_pattern = re.compile(r'[CLP]\d{2,3}(?:O\d{2,3})?(?![\dB])') _base_pattern = re.compile(r'[CLP]\d{2,3}') # scene 단위 valid (character, outlook) pair 집합. 다음 source 통합: # - scene_outlooks: scene assignment 에서 LLM 이 배정한 cid → outlook 후보 # - fixed_outfits: 단일 outlook 자동 배정 결과 (O00 포함 — 비인간형/변형체 또는 fallback) # 주의: 인간형 cid 에 O00 무조건 허용 안 한다 (Codex/Claude review IMPORTANT) # — fixed_outfits/scene_outlooks 가 명시한 (cid, O00) 만 통과. 인간형이 # explicit non-null outlook 갖고 있는데 LLM 이 C##O00 출력하면 retry. allowed_outlook_pairs: set = set() for _cid, _ols in scene_outlooks.items(): for _osid, _ in _ols: if _osid: allowed_outlook_pairs.add((_cid, _osid)) for _cid, _osid in fixed_outfits.items(): if _osid: allowed_outlook_pairs.add((_cid, _osid)) def _extract_base(sid: str) -> str: """ID 에서 base 추출. composite (C01O02) → C01, bare 는 그대로. sid[:3] 단순 truncate 는 P123 같은 3-digit ID 에서 P12 로 misalign. 정확한 prefix regex 매치 사용. """ m = _base_pattern.match(sid) return m.group(0) if m else sid def _check_prompts(res): """위반 검사 — base ID invalid 또는 composite outlook invalid 면 to_remove. composite ID(C##O##) 의 outlook 부분 (O##) 검증 — (cid, oid) pair 가 allowed_outlook_pairs 에 있어야 valid. fail-fast 원칙으로 invalid outlook 을 silent 통과 안 시키고 retry 강제. P##O## / L##O## 같은 non-character composite 는 valid v13 출력 형식 아님 — 무조건 reject. """ to_remove = set() for var in res.get("t2i_variations", []): found = set(_ve_pattern.findall(var.get("t2i_prompt", ""))) for sid in found: base_sid = _extract_base(sid) if base_sid not in allowed_set: to_remove.add(sid) continue # composite case (sid 가 base 보다 김 → outlook suffix 있음). if sid != base_sid: # P##/L## 에 outlook suffix 는 valid 형식 아님 (Codex review). if not base_sid.startswith("C"): to_remove.add(sid) continue outlook_sid = sid[len(base_sid):] if (base_sid, outlook_sid) not in allowed_outlook_pairs: to_remove.add(sid) return to_remove # 글로벌 3-tier fallback (call_structured 내부) — gemini → sanitize → gpt 자동. result = call_structured( step="scene_detail", system_prompt=system, user_prompt=user_prompt, response_schema=schema, project_config=self.project_config, schema_name=f"scene_detail_{si}", opik_metadata=self.build_opik_metadata(extra_metadata={"scene_index": si}), ) to_remove = _check_prompts(result) # T6 (area-frame-spatial-contract): post-validation echo set 검증. # validate_echoes 는 violation list 반환 (raise X). retry trigger 는 # to_remove 또는 fsc_echo_violations 어느 한쪽이라도 있으면 발동. # _g41_render_prompt_card 는 같은 _analyze_one 안에서 build_render_prompt_card 가 # 생성한 카드 (line 2461). render_strategy.frame_spatial_contract 가 None 이면 # injected_ids=set(), 모든 variation echo=[] 강제 — helper 가 처리. fsc_echo_violations = _fsc_validate_echoes(_g41_render_prompt_card, result) if to_remove or fsc_echo_violations: corrections = [] if to_remove: corrections.append(f"제거할 ID: {', '.join(sorted(to_remove))}") for v in fsc_echo_violations: corrections.append( f"variation '{v['variant_label']}': applied_frame_spatial_constraint_ids " f"echo 가 contract id set 과 다릅니다. " f"missing={v['missing']} extra={v['extra']} " f"(injected={v['injected_ids']})" ) logger.warning("Scene %d: %s — retry", si, " / ".join(corrections)) retry_prompt = ( f"[수정 요청]\n" + "\n".join(f"- {c}" for c in corrections) + "\n\n" + user_prompt ) result = call_structured( step="scene_detail", system_prompt=system, user_prompt=retry_prompt, response_schema=schema, project_config=self.project_config, schema_name=f"scene_detail_{si}_retry", opik_metadata=self.build_opik_metadata(extra_tags=["retry"], extra_metadata={"scene_index": si}), ) # retry 후에도 위반 → 변형 치환 또는 강제 제거 post_remove = _check_prompts(result) if post_remove: # variant_resolved가 있으면 base→variant 치환 _vr = ctx.shot_director_vr.get((si, shot_idx), {}) replace_map = {} pure_remove = set() for sid in post_remove: base_id = _extract_base(sid) # C01O02 → C01, P123 → P123 (정확) resolved = _vr.get(base_id) if resolved and resolved != base_id: replace_map[base_id] = resolved else: pure_remove.add(sid) for var in result.get("t2i_variations", []): prompt = var.get("t2i_prompt", "") # 변형 치환 — composite (base+O##) + bare 양쪽 모두 base 만 치환, # outlook 부분은 lookahead 로 보존. (Codex/Claude review BLOCKING fix) # 옛 ``\bC02\b`` 단독 패턴은 ``C02O01`` 안의 C02 와 매치 안 함 # (O 가 \w → \b 미충족) → composite 통과 회귀 위험. for old_id, new_id in replace_map.items(): prompt = re.sub( r'\b' + re.escape(old_id) + r'(?=O\d{2,3}\b|\b)', new_id, prompt, ) # 나머지 강제 제거 if pure_remove: prompt = _strip_invalid_sids(prompt, pure_remove) var["t2i_prompt"] = prompt # outfit_assignments에서도 치환 for oa in var.get("outfit_assignments", []): cid = oa.get("character_id", "") if cid in replace_map: oa["character_id"] = replace_map[cid] # pure_remove 의 base ID 가 character_id 면 outfit_assignments # 에서도 drop — t2i_prompt 와 outfit_assignments 정합성 보장 # (Claude review IMPORTANT). 단순 strip 만 했을 때 t2i_prompt # 에는 없는데 assignments 에 살아남는 회귀 차단. if pure_remove: _pr_bases = {_extract_base(s) for s in pure_remove} var["outfit_assignments"] = [ oa for oa in var.get("outfit_assignments", []) if oa.get("character_id", "") not in _pr_bases ] if replace_map: logger.info("Scene %d Shot %s: variant replaced %s", si, shot_idx, replace_map) if pure_remove: logger.warning("Scene %d: retry 후에도 위반 %s — 강제 제거", si, pure_remove) # T6 (area-frame-spatial-contract): post-retry echo 재검사. # graceful fix (변형 치환 / 강제 제거) 는 t2i_prompt 안 base ID 만 # 다루므로 applied_frame_spatial_constraint_ids 의 echo set drift 는 # 자체적으로 복구 안 된다. 1 회 retry 후에도 echo mismatch 면 # AppError raise — outer try/except 가 None 반환 → shot-level retry # path 로 전달 (line 1025-1037 architecture: contract violation 도 # typed failure 로 보존). post_fsc = _fsc_validate_echoes(_g41_render_prompt_card, result) if post_fsc: raise AppError( code="scene_detail.frame_spatial_contract_echo_mismatch", message=( f"scene_detail S{si} shot{shot_idx}: per-variation " f"applied_frame_spatial_constraint_ids echo mismatch " f"after 1 retry. graceful fix 불가능." ), details={"violations": post_fsc}, ) # T6 (area-frame-spatial-contract): phrase diagnostic — soft warning # only. raise 없음. constraint 의 label/zone/depth phrase 가 t2i_prompt # 안 미충족 시 logger.warning 만 기록 — v1 soft, v2 hard 승격 가능. # _g41_render_prompt_card.render_strategy.frame_spatial_contract 가 # None 이면 constraints=[] 로 자연 skip. _fsc_rs = (_g41_render_prompt_card or {}).get("render_strategy") or {} _fsc_obj = _fsc_rs.get("frame_spatial_contract") if _fsc_obj is not None: _fsc_constraints = _fsc_obj.get("constraints") or [] for var in result.get("t2i_variations", []) or []: diag = _fsc_phrase_diagnostic(var.get("t2i_prompt", ""), _fsc_constraints) for d in diag: logger.warning( "scene_detail S%d shot%s variation %s frame_spatial_contract " "phrase diagnostic: %s", si, shot_idx, var.get("variant_label"), d, ) # representative_moment VE 위반 강제 제거. # 쓰기는 representative_moment만 사용(현행 스키마). # 읽기는 레거시 체크포인트(still_frame_prompt) fallback 유지. sfp = result.get("representative_moment", "") or result.get("still_frame_prompt", "") if sfp: # _extract_base 사용 — sid[:3] 단순 truncate 는 3-digit ID 회귀 위험. bad_in_sfp = {sid for sid in _ve_pattern.findall(sfp) if _extract_base(sid) not in allowed_set} if bad_in_sfp: sfp = _strip_invalid_sids(sfp, bad_in_sfp) result["representative_moment"] = sfp # 레거시 키가 남아 있으면 혼란 방지를 위해 제거 result.pop("still_frame_prompt", None) result["scene_index"] = si result["_shot_index"] = shot_idx if shot_info else None result["visible_entities"] = visible # G4.1 Phase 5 Task 17 (5-point modify scope guard 4/5): top-level # result fields — Open Q1 RESOLVED (top-level, NOT per-variation). # variation drift 는 G3.2 sentinel 가 담당 (shot-level vs variation-level # 분리 — coexist). result["render_prompt_card"] = _g41_render_prompt_card result["render_prompt_card_hash"] = _g41_render_prompt_card_hash # variation 개수 강제 t2i_vars = result.get("t2i_variations", []) if len(t2i_vars) > var_count: result["t2i_variations"] = t2i_vars[:var_count] t2i_vars = result["t2i_variations"] # outfit_assignments → 코드에서 bare C## → C##O## 조합 # 이 씬에서 유효한 아웃룩 세트 (캐릭터별) valid_outfits: Dict[str, set] = {} for cid, ols in scene_outlooks.items(): valid_outfits[cid] = {osid for osid, _ in ols} for var in t2i_vars: assignments = var.get("outfit_assignments", []) # 1) visible_entities에 없는 캐릭터 제거 if assignments: assignments = [a for a in assignments if a.get("character_id", "") in allowed_set] # 2) 이 씬의 유효 아웃룩이 아닌 경우 → 교체 또는 제거 cleaned_assignments = [] for a in assignments: cid = a.get("character_id", "") oid = a.get("outlook_id", "") valid = valid_outfits.get(cid, set()) if not valid: # 이 캐릭터에 아웃룩 없음 → 할당 제거 logger.warning("씬 %d: %s 아웃룩 없음 — 할당 제거", si, cid) continue if oid not in valid: correct = scene_outlooks[cid][0][0] logger.warning("씬 %d: %s 유효하지 않은 아웃룩 %s → %s로 교체", si, cid, oid, correct) a["outlook_id"] = correct cleaned_assignments.append(a) assignments = cleaned_assignments var["outfit_assignments"] = assignments # 3) 확정 아웃룩(1개) 자동 배정 + 빠진 캐릭터 보충 # prompt에 등장하는 캐릭터만 대상 (씬 VE 전체가 아닌 shot 단위) # Codex/Claude review IMPORTANT (Group 1 #1): v13 alignment 후 LLM 이 # composite C##O## 를 default 로 출력. (?!O\d) negative lookahead 는 # composite 형식의 base C## 를 추출 못 해 outfit_assignments 가 모두 drop # 됨 (DB/UI sync 손실). lookahead 제거 — bare + composite 의 base 모두 추출. assigned_chars = {a.get("character_id", "") for a in assignments} prompt_chars = set(re.findall(r'(? bool: rule = self.manifest.get("applicability", "always") if rule != "if_multi_char_scenes": return True detail_cp = self._load_prev_checkpoint("scene_detail") if not detail_cp: return False scenes = detail_cp.get("data", {}).get("scenes", []) return any(len(s.get("visible_entities", [])) >= 2 for s in scenes) def _execute(self, mode="resume") -> Dict[str, Any]: detail_cp = self._load_prev_checkpoint("scene_detail") scenes = detail_cp.get("data", {}).get("scenes", []) if detail_cp else [] save_cp = self._load_prev_checkpoint("scene_save") segments = save_cp.get("data", {}).get("segments", []) if save_cp else [] system = load_prompt("scene_verify", "system") schema = load_schema("scene_verify", "verify_schema") results = [] failed = 0 def _verify_one(scene): si = scene.get("scene_index", 0) shot_idx = scene.get("_shot_index") visible = scene.get("visible_entities", []) if len(visible) < 2: return scene # skip single-entity scenes # Get previous 2 scenes text prev_texts = [] for j in range(max(1, si - 2), si): prev_seg = next( (s for s in segments if s.get("scene_index") == j), None ) if prev_seg: prev_texts.append(prev_seg.get("text", "")) cur_seg = next( (s for s in segments if s.get("scene_index") == si), None ) scene_text = cur_seg.get("text", "") if cur_seg else "" # v4: shot 정보 포함 shot_context = "" if shot_idx is not None: # representative_moment 우선, 레거시 체크포인트(still_frame_prompt) fallback shot_desc = scene.get("representative_moment", "") or scene.get("still_frame_prompt", "") shot_context = f"\n[검증 대상 Shot {shot_idx}]: {shot_desc}\n" user_prompt = ( f"앞 씬 컨텍스트:\n{'---'.join(prev_texts)}\n\n" f"현재 씬 {si}:\n{scene_text}\n" f"{shot_context}\n" f"확정 엔티티: {', '.join(visible)}\n" "위 엔티티가 물리적으로 이 씬(또는 shot)에 존재하는지 검증해주세요." ) try: verify_result = call_structured( step="scene_verify", system_prompt=system, user_prompt=user_prompt, response_schema=schema, project_config=self.project_config, schema_name=f"scene_verify_{si}", opik_metadata=self.build_opik_metadata(extra_tags=["verify"], extra_metadata={"scene_index": si}), ) scene["verification"] = verify_result return scene except Exception as exc: logger.warning("Scene %d verify failed: %s", si, exc) return scene with ThreadPoolExecutor(max_workers=5) as pool: futures = {pool.submit(_verify_one, s): s for s in scenes} for f in as_completed(futures): r = f.result() if r: results.append(r) else: failed += 1 results.sort(key=lambda r: r.get("scene_index", 0)) return { "completed_count": len(results), "applicable_count": len(scenes), "failed_count": failed, "data": {"scenes": results}, }