"""W21B-W7 (2026-06-12) — visual_continuity_anchor deterministic core. B-run 피드백 버킷 C(zoom pose 모순)/D(printed prop identity·scale) 의 production 1차. 이 모듈은 순수 deterministic 영역만 담는다: - seed 탐지 (LLM 0): C(zoom_continuity) = refined ``ref_usage == "zoom_in_detail"`` dep pair (같은 scene 의 selected shot pair). exact same-moment(``exact_background``)는 production 소비 없이 diagnostics 후보로만 기록 (Codex 판정 ①). D = prop short_id 가 2개 이상의 selected shots 의 visible_entity_ids / required_refs(kind=prop) 에 등장. framing 은 hard gate 가 아니라 priority score (Codex 판정 ④) — 코드는 structured enum / ID 조인만 수행하고 단어 regex / substring 의미 판별은 절대 하지 않는다. - cap 선별: per-episode LLM 콜 cap. 초과 seed 는 skipped(reason="cap") 로 진단 기록 — silent drop 금지 (bg_space_partition cost-cap 선례). - manifest 조립 + 조인 무결성 검증 (의미 게이트 없음 — Codex 가이드 5). LLM anchor 추출(스파이크 1/2 포팅)은 ``visual_continuity_anchor_provider`` 가 담당한다. anchor 품질은 deterministic 테스트 비대상 — canary + 육안. 설계: docs/w21b-w7-visual-continuity-anchor-production-brief-20260612/. """ from __future__ import annotations import logging import re from typing import Any, Callable, Dict, List, Optional, Set, Tuple from app.core.subject_state import is_immobilized_state logger = logging.getLogger(__name__) SCHEMA_VERSION: int = 1 DEFAULT_GROUP_CAP: int = 8 # P8 (2026-06-20): immobilized-subject continuity anchors are a separate SOT with # their own cap so they are not starved by printed_prop seeds (Codex 합의 ⑧). DEFAULT_IMMOBILIZED_CAP: int = 4 ANCHOR_TYPE_ZOOM = "zoom_continuity" ANCHOR_TYPE_PROP = "printed_prop" # P8 dynamic-entity continuity: a character who is immobilized (dead / unconscious / # severely_injured per subject_state enum) and recurs across >=2 selected shots of # the same scene. Its pose / position / orientation / held-or-nearby props must stay # consistent shot-to-shot even though framing may crop it. Orthogonal to zoom # continuity (which is tied to camera_relation / crop strategy). ANCHOR_TYPE_IMMOBILIZED = "immobilized_subject" REF_USAGE_ZOOM = "zoom_in_detail" REF_USAGE_EXACT_BG = "exact_background" ROLE_SOURCE_WIDE = "source_wide" ROLE_ZOOM = "zoom" ROLE_CLOSE_INSERT = "close_insert" ROLE_ENVIRONMENT = "environment" # Cinematography 축 (2026-06-12 사용자 지시) — anchor LLM 의 structured 판정 enum. # punch-in 만 pixel crop 적격; 그 외 zoom 은 자기 camera plan(t2i) 유지. CAMERA_RELATION_PUNCH_IN = "same_axis_punch_in" CAMERA_RELATION_DIFFERENT = "different_camera_same_moment" CAMERA_RELATION_UNCERTAIN = "uncertain" # shot_staging.framing_scale 의 structured enum 중 "디테일 지배" 계열 — # D seed priority score 신호로만 쓴다 (hard gate 아님, Codex 판정 ④). _CLOSE_FRAMING_SCALES = frozenset({"close", "insert"}) ShotKey = Tuple[int, int] # (scene_index, shot_index) # ─────────────────────────── C seed (zoom_continuity) ─────────────────────────── def detect_c_seeds( dependencies: List[Dict[str, Any]], selected_map: Dict[int, Set[int]], *, source_still_exists: Optional[Callable[[int, int], bool]] = None, ) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]], List[Dict[str, Any]]]: """zoom_in_detail dep pair 에서 C seed 를 deterministic 하게 뽑는다. Args: dependencies: shot_dependency(_t2i) cp 의 ``data.dependencies`` — [{scene_index, shot_index, location_refs:[{scene_index, shot_index, ref_usage, ...}], ...}]. ref_usage 는 현재 refined cp (shot_dependency_t2i) 에만 실재한다 — caller 가 source 선택을 책임. selected_map: {scene_index: {selected shot_index}}. source_still_exists: (si, shi) → bool. source 가 selected 가 아니어도 기존 still 이 실존하면 seed 성립 (Codex 판정 ① 추가 조건). None 이면 selected 만 인정. Returns: (seeds, skipped, exact_candidates) — seeds: [{"zoom": [si, shi], "source": [si, shi]}] skipped: [{"seed": str, "reason": str}] exact_candidates: production 소비 없는 same-scene exact_background pair 진단 후보 (``candidate_same_moment_exact_background``). """ seeds: List[Dict[str, Any]] = [] skipped: List[Dict[str, Any]] = [] exact_candidates: List[Dict[str, Any]] = [] seen_zoom: Set[ShotKey] = set() for dep in dependencies or []: zoom_si = dep.get("scene_index") zoom_shi = dep.get("shot_index") if not isinstance(zoom_si, int) or not isinstance(zoom_shi, int): continue for ref in dep.get("location_refs") or []: usage = ref.get("ref_usage") src_si = ref.get("scene_index") src_shi = ref.get("shot_index") if not isinstance(src_si, int) or not isinstance(src_shi, int): continue label = f"S{zoom_si}sh{zoom_shi}<-S{src_si}sh{src_shi}" if usage == REF_USAGE_EXACT_BG and src_si == zoom_si: exact_candidates.append( {"zoom": [zoom_si, zoom_shi], "source": [src_si, src_shi]} ) continue if usage != REF_USAGE_ZOOM: continue if zoom_shi not in selected_map.get(zoom_si, set()): skipped.append({"seed": label, "reason": "zoom_not_selected"}) continue if src_si != zoom_si: # v1 narrow scope: 같은 씬의 같은 순간 pair 만 (Codex 판정 ①). skipped.append({"seed": label, "reason": "cross_scene_zoom"}) continue if (zoom_si, zoom_shi) in seen_zoom: skipped.append({"seed": label, "reason": "duplicate_zoom_member"}) continue source_ok = src_shi in selected_map.get(src_si, set()) if not source_ok and source_still_exists is not None: source_ok = bool(source_still_exists(src_si, src_shi)) if not source_ok: skipped.append({"seed": label, "reason": "source_still_unavailable"}) continue seen_zoom.add((zoom_si, zoom_shi)) seeds.append({"zoom": [zoom_si, zoom_shi], "source": [src_si, src_shi]}) return seeds, skipped, exact_candidates # ─────────────────────────── D seed (printed_prop) ─────────────────────────── def detect_d_seeds( ve_by_shot: Dict[ShotKey, List[str]], selected_map: Dict[int, Set[int]], prop_short_ids: Set[str], *, framing_by_shot: Optional[Dict[ShotKey, str]] = None, required_prop_refs_by_shot: Optional[Dict[ShotKey, Set[str]]] = None, ) -> List[Dict[str, Any]]: """prop 반복 등장 기반 D seed + priority score (Codex 판정 ④). base = prop short_id 가 2개 이상의 selected shots 에서 visible_entity_ids 또는 required_refs(kind=prop) 로 등장. score 는 ID/enum 조인만: - required_refs(kind=prop) 등장 샷 수 × 10 (가장 강한 신호) - framing_scale ∈ {close, insert} 샷 수 × 3 (structured enum) - 멤버 샷 수 × 1 required_prop_refs_by_shot 은 scene_detail cp soft 역참조 — fresh run 에는 None (신호 부재일 뿐 seed 차단 아님). """ framing_by_shot = framing_by_shot or {} required_prop_refs_by_shot = required_prop_refs_by_shot or {} member_shots: Dict[str, List[ShotKey]] = {} for si, selected in selected_map.items(): for shi in sorted(selected): key = (si, shi) present: Set[str] = set() for sid in ve_by_shot.get(key, []) or []: if sid in prop_short_ids: present.add(sid) for sid in required_prop_refs_by_shot.get(key, set()): if sid in prop_short_ids: present.add(sid) for sid in present: member_shots.setdefault(sid, []).append(key) seeds: List[Dict[str, Any]] = [] for sid in sorted(member_shots): shots = sorted(member_shots[sid]) if len(shots) < 2: continue required_count = sum( 1 for key in shots if sid in required_prop_refs_by_shot.get(key, set()) ) close_count = sum( 1 for key in shots if framing_by_shot.get(key) in _CLOSE_FRAMING_SCALES ) seeds.append({ "prop_short_id": sid, "member_shots": [list(key) for key in shots], "score": required_count * 10 + close_count * 3 + len(shots), }) # score 내림차순, 동점은 short_id 오름차순 — deterministic. seeds.sort(key=lambda s: (-s["score"], s["prop_short_id"])) return seeds # ─────────────────────────── P8 seed (immobilized_subject) ─────────────────────────── def build_identity_family_by_sid( relations: List[Dict[str, Any]], *, entity_type: str = "character", ) -> Dict[str, Set[str]]: """entity_relation 체크포인트 relations → short_id identity-family 맵 (union-find). 변형 EntityCanon(예: 같은 인물의 환영/시신 외형)은 base 와 동일 인물이라는 데이터 계약(base_short_id ↔ variant_short_id)을 short-id 공간의 등가집합으로 노출한다. VE/staging 이 서로 다른 variant 표기를 쓸 때 exact-ID 게이트가 같은 인물을 놓치는 것을 막는 SOT (2026-07-02, S12 시신 seed 전멸 수정). - ``entity_type`` 정확 일치(enum 비교) 항목만 결합 — prop/location 미적용. - 반환: 모든 family 멤버 sid → 그 family 의 전체 sid set(자기 포함). 관계가 없거나 입력이 비면 빈 dict (게이트는 기존 exact 동작으로 환원). """ parent: Dict[str, str] = {} def _find(x: str) -> str: while parent[x] != x: parent[x] = parent[parent[x]] x = parent[x] return x def _union(a: str, b: str) -> None: parent.setdefault(a, a) parent.setdefault(b, b) ra, rb = _find(a), _find(b) if ra != rb: parent[rb] = ra for r in relations or []: if not isinstance(r, dict): continue if str(r.get("entity_type") or "") != entity_type: continue base = str(r.get("base_short_id") or "").strip() variant = str(r.get("variant_short_id") or "").strip() if not base or not variant or base == variant: continue _union(base, variant) roots: Dict[str, Set[str]] = {} for sid in parent: roots.setdefault(_find(sid), set()).add(sid) out: Dict[str, Set[str]] = {} for members in roots.values(): for sid in members: out[sid] = set(members) return out def detect_immobilized_subject_seeds( staging_by_shot: Dict[ShotKey, Dict[str, Any]], ve_by_shot: Dict[ShotKey, List[str]], selected_map: Dict[int, Set[int]], name_to_sids: Dict[str, Set[str]], *, framing_by_shot: Optional[Dict[ShotKey, str]] = None, identity_family_by_sid: Optional[Dict[str, Set[str]]] = None, ) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]: """immobilized 인물의 씬 내 다중 샷 연속성 seed (Codex 합의 P8 v1 gate). gate (전부 구조 신호 — 글자/시나리오 판정 0): - shot_staging ``character_angles[*].subject_state`` 가 IMMOBILIZED_STATES (``is_immobilized_state``) 인 인물. - 그 인물 이름(``character``) → short_id 를 ``name_to_sids`` exact 조인. 모호(2+)/미상(0) 이름은 seed 제외 + diagnostic (substring 매칭 금지 — entity catalog display-name exact equality 만, 기존 state/ref 흐름과 동일). - 그 short_id 가 그 샷의 ``visible_entity_ids`` 에 실재해야 멤버 (VE 누락 샷 억지 포함 금지 — producer 오류 은폐 방지, Codex 세부의견). ★identity-variant aware (2026-07-02): subject sid 가 VE 에 없어도 ``identity_family_by_sid``(entity_relation 데이터 계약, character 한정)의 같은 family 멤버가 VE 에 실재하면 같은 인물로 보고 멤버 인정 — 매칭 내역은 skipped 채널에 ``subject_matched_via_identity_variant`` 진단으로 기록(subject/ve_member/family 감사 가능). - 같은 ``scene_index`` 의 같은 short_id 가 ≥2 selected shots 에 등장 → seed. shot_index 인접 조건 없음 (insert/cutaway 건너뛰는 연속성, Codex Q2). - 한 그룹의 멤버 subject_state 가 섞이면(예: dead + unconscious) ``mixed_immobilized_states`` diagnostic + v1 skip (Codex 보수안). score (cap 선별용, deterministic): close/insert framing 가중 + 멤버수. Returns: (seeds, skipped). seed 항목: {character_short_id, scene_index, subject_state, member_shots: [[si, shi], ...], score} """ framing_by_shot = framing_by_shot or {} # (scene_index, char_sid) → [{shot_index, subject_state}] grouped: Dict[Tuple[int, str], List[Dict[str, Any]]] = {} skipped: List[Dict[str, Any]] = [] for si, selected in selected_map.items(): for shi in sorted(selected): key = (si, shi) staging = staging_by_shot.get(key) if not staging: continue ve = set(ve_by_shot.get(key, []) or []) for ca in staging.get("character_angles") or []: state = ca.get("subject_state") if not is_immobilized_state(state or ""): continue name = ca.get("character", "") or "" sids = name_to_sids.get(name) or set() if len(sids) != 1: skipped.append({ "seed": f"S{si}sh{shi}:{name or ''}", "reason": ("ambiguous_subject_name" if sids else "unknown_subject_name"), }) continue sid = next(iter(sids)) if sid not in ve: fam = (identity_family_by_sid or {}).get(sid) or set() fam_hit = sorted(fam & ve) if not fam_hit: entry: Dict[str, Any] = { "seed": f"S{si}sh{shi}:{sid}", "reason": "subject_not_in_visible_entities", } if fam: entry["identity_family"] = sorted(fam) skipped.append(entry) continue # 같은 인물의 variant EntityCanon 이 VE 에 실재 — 멤버 인정. # (진단 채널에 기록해 manifest 에서 감사 가능하게 유지.) skipped.append({ "seed": f"S{si}sh{shi}:{sid}", "reason": "subject_matched_via_identity_variant", "ve_member": fam_hit[0], "identity_family": sorted(fam), }) grouped.setdefault((si, sid), []).append( {"shot_index": shi, "subject_state": state}) seeds: List[Dict[str, Any]] = [] for (si, sid), members in sorted(grouped.items()): # dedup shot_index (한 샷의 character_angles 에 같은 인물 중복 방어). by_shi: Dict[int, str] = {} for m in members: by_shi.setdefault(m["shot_index"], m["subject_state"]) if len(by_shi) < 2: continue states = set(by_shi.values()) if len(states) > 1: skipped.append({ "seed": f"S{si}:{sid}", "reason": "mixed_immobilized_states", "states": sorted(states), }) continue shots = sorted(by_shi) close_count = sum( 1 for shi in shots if framing_by_shot.get((si, shi)) in _CLOSE_FRAMING_SCALES ) seeds.append({ "character_short_id": sid, "scene_index": si, "subject_state": next(iter(states)), "member_shots": [[si, shi] for shi in shots], "score": close_count * 3 + len(shots), }) seeds.sort(key=lambda s: (-s["score"], s["scene_index"], s["character_short_id"])) return seeds, skipped def select_immobilized_seeds_with_cap( seeds: List[Dict[str, Any]], cap: int, ) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]: """immobilized seed 를 자체 cap 으로 선별 (printed_prop cap 과 독립 — Codex ⑧). 초과분은 silent drop 금지 — skipped(reason="immobilized_cap") + score 기록.""" cap = max(cap, 0) take = min(len(seeds), cap) selected = seeds[:take] skipped = [ {"seed": f"immobilized:S{s['scene_index']}:{s['character_short_id']}", "reason": "immobilized_cap", "score": s.get("score")} for s in seeds[take:] ] return selected, skipped def build_immobilized_subject_group( seed: Dict[str, Any], llm_result: Dict[str, Any], *, framing_by_shot: Optional[Dict[ShotKey, str]] = None, related_zoom_group_ids: Optional[List[str]] = None, ) -> Dict[str, Any]: """P8 seed + LLM 추출 → immobilized_subject group. member role 은 framing_scale structured enum 으로만 분류 (close/insert → close_insert, 그 외 → environment) — 의미 추론 없음. subject_anchor 는 그룹 전체 공유 SOT (locked_elements = evidence-backed 공유 시각 사실) + per-shot visible focus (insert 샷에 wide body 강요 방지). zoom 그룹은 SOT 가 아니라 참고 링크(related_zoom_group_ids) 로만 남긴다 (Codex Q1).""" framing_by_shot = framing_by_shot or {} sid = seed["character_short_id"] si = seed["scene_index"] members: List[Dict[str, Any]] = [] hints: Dict[str, Dict[str, Any]] = {} for msi, mshi in (tuple(m) for m in seed["member_shots"]): role = ( ROLE_CLOSE_INSERT if framing_by_shot.get((msi, mshi)) in _CLOSE_FRAMING_SCALES else ROLE_ENVIRONMENT ) members.append({"scene_index": msi, "shot_index": mshi, "role": role}) hints[shot_hint_key(msi, mshi)] = {"enforce_state_continuity": True} # per_shot_visible_focus: LLM 이 멤버 샷별로 "프레임에 실제 보이는 초점"을 # [{shot_index, visible_focus}] 로 반환 — shot_index 키 dict 로 정규화하되 # seed 멤버 샷만 채택 (낯선 키 무시). member_shis = {mshi for _, mshi in (tuple(m) for m in seed["member_shots"])} focus_by_shot: Dict[str, str] = {} for entry in llm_result.get("per_shot_visible_focus") or []: if not isinstance(entry, dict): continue f_shi = entry.get("shot_index") f_text = str(entry.get("visible_focus") or "").strip() if isinstance(f_shi, int) and f_shi in member_shis and f_text: focus_by_shot[str(f_shi)] = f_text return { "group_id": f"vca-s{si}-{sid.lower()}-immobilized", "anchor_type": ANCHOR_TYPE_IMMOBILIZED, "members": members, "locked_elements": llm_result.get("locked_elements") or [], "subject_anchor": { "character_short_id": sid, "subject_state": seed["subject_state"], "shared_state_contract": str( llm_result.get("shared_state_contract") or "").strip(), "per_shot_visible_focus": focus_by_shot, }, "related_zoom_group_ids": list(related_zoom_group_ids or []), "per_shot_consumption_hints": hints, } def build_immobilized_prev_frame_plan( members_by_group: Dict[str, List[Dict[str, Any]]], ) -> Dict[ShotKey, Dict[str, Any]]: """A5 (2026-07-02) — immobilized 그룹 prev 완성프레임 chaining 계획 (pure). star-to-environment (Codex 합의): 각 멤버 샷을 (scene_index, shot_index) 순으로 정렬해, 자신보다 **앞선 가장 가까운 role=environment 멤버**를 anchor_source 로 매핑한다. immobilized state 는 정의상 불변이라 환경(wide) 프레임이 room state· 지지면·소품 연속성의 가장 안정적인 SOT — close/insert 프레임은 앵커로 쓰지 않는다(직전-멤버 체인은 insert 가 끼면 정보가 줄고 오류 전파가 큼). - 선행 env 멤버가 없는 멤버(그룹 첫 env 이전 샷 포함)는 엔트리 없음 → 소비자 no-op. env 멤버 자기 자신(anchor==target)은 제외. - 한 샷이 여러 그룹 멤버면 group_id 정렬 순 첫 매핑 유지 (결정론). - role 은 manifest 의 structured enum 만 조인 — 의미 추론 0. """ plan: Dict[ShotKey, Dict[str, Any]] = {} for gid in sorted(members_by_group.keys()): members = members_by_group.get(gid) or [] ordered: List[Tuple[ShotKey, str]] = [] seen: Set[ShotKey] = set() for m in members: msi, mshi = m.get("scene_index"), m.get("shot_index") if not (isinstance(msi, int) and isinstance(mshi, int)): continue key: ShotKey = (msi, mshi) if key in seen: continue seen.add(key) ordered.append((key, str(m.get("role") or ""))) ordered.sort(key=lambda e: e[0]) last_env: Optional[ShotKey] = None for key, role in ordered: if last_env is not None and key != last_env and key not in plan: plan[key] = { "anchor_source": last_env, "group_id": gid, "anchor_role": ROLE_ENVIRONMENT, } if role == ROLE_ENVIRONMENT: last_env = key return plan # ─────────────────────────── cap 선별 ─────────────────────────── DEFAULT_D_RESERVED_SLOTS: int = 2 def select_seeds_with_cap( c_seeds: List[Dict[str, Any]], d_seeds: List[Dict[str, Any]], cap: int, *, d_reserved: int = DEFAULT_D_RESERVED_SLOTS, ) -> Tuple[List[Tuple[str, Dict[str, Any]]], List[Dict[str, Any]]]: """cap 안에서 (anchor_type, seed) 선별 — type 별 공정성 장치 포함. Codex W_A_STAGE_REVIEW (2026-06-12): C 가 많은 에피소드에서 1차 production 핵심인 D 가 cap 에 밀리지 않도록 D(printed_prop) 에 최소 예약 슬롯을 둔다. 1. D 가 먼저 min(d_reserved, len(d_seeds), cap) 슬롯 확보 (score 순) 2. 남은 슬롯을 C 가 채움 (dep 순) 3. 그래도 남으면 D 추가분 (score 순) 초과분은 skipped(reason="cap_after_priority_score") + score 기록 — silent drop 금지.""" cap = max(cap, 0) d_take = min(max(d_reserved, 0), len(d_seeds), cap) c_take = min(len(c_seeds), cap - d_take) d_take = min(len(d_seeds), cap - c_take) # 남은 슬롯은 D 가 더 채움 selected: List[Tuple[str, Dict[str, Any]]] = ( [(ANCHOR_TYPE_ZOOM, s) for s in c_seeds[:c_take]] + [(ANCHOR_TYPE_PROP, s) for s in d_seeds[:d_take]] ) skipped = [ {"seed": _seed_label(ANCHOR_TYPE_ZOOM, s), "reason": "cap_after_priority_score"} for s in c_seeds[c_take:] ] + [ {"seed": _seed_label(ANCHOR_TYPE_PROP, s), "reason": "cap_after_priority_score", "score": s.get("score")} for s in d_seeds[d_take:] ] return selected, skipped def _seed_label(anchor_type: str, seed: Dict[str, Any]) -> str: if anchor_type == ANCHOR_TYPE_ZOOM: z = seed.get("zoom", ["?", "?"]) return f"zoom:S{z[0]}sh{z[1]}" return f"prop:{seed.get('prop_short_id', '?')}" # ─────────────────────────── group 조립 ─────────────────────────── def shot_hint_key(scene_index: int, shot_index: int) -> str: return f"{scene_index}:{shot_index}" def build_zoom_group( seed: Dict[str, Any], llm_result: Dict[str, Any], ordinal: int, ) -> Dict[str, Any]: """C seed + LLM 추출 결과 → zoom_continuity group. hint 는 deterministic: source_wide → include_in_wide / zoom → crop_from_source (camera_relation 이 punch-in 일 때만 — Cinematography 축. 부재(legacy/stub)는 punch-in 취급 = 기존 동작 보존).""" src = seed["source"] zoom = seed["zoom"] relation = llm_result.get("camera_relation") or CAMERA_RELATION_PUNCH_IN return { "group_id": f"vca-s{zoom[0]}-zoom-{ordinal}", "anchor_type": ANCHOR_TYPE_ZOOM, "members": [ {"scene_index": src[0], "shot_index": src[1], "role": ROLE_SOURCE_WIDE}, {"scene_index": zoom[0], "shot_index": zoom[1], "role": ROLE_ZOOM}, ], "locked_elements": llm_result.get("locked_elements") or [], "wide_shot_contract": llm_result.get("wide_shot_contract") or [], "camera_relation": relation, "camera_relation_evidence": llm_result.get("camera_relation_evidence", ""), "per_shot_consumption_hints": { shot_hint_key(*src): {"include_in_wide": True}, shot_hint_key(*zoom): { "crop_from_source": relation == CAMERA_RELATION_PUNCH_IN, }, }, } def build_prop_group( seed: Dict[str, Any], llm_result: Dict[str, Any], *, framing_by_shot: Optional[Dict[ShotKey, str]] = None, ) -> Dict[str, Any]: """D seed + LLM 추출 결과 → printed_prop group. member role 은 framing_scale structured enum 으로만 분류 (close/insert → close_insert, 그 외 → environment) — 의미 추론 없음.""" framing_by_shot = framing_by_shot or {} sid = seed["prop_short_id"] members: List[Dict[str, Any]] = [] hints: Dict[str, Dict[str, Any]] = {} for si, shi in (tuple(m) for m in seed["member_shots"]): role = ( ROLE_CLOSE_INSERT if framing_by_shot.get((si, shi)) in _CLOSE_FRAMING_SCALES else ROLE_ENVIRONMENT ) members.append({"scene_index": si, "shot_index": shi, "role": role}) hints[shot_hint_key(si, shi)] = {"keep_real_scale": True} return { "group_id": f"vca-{sid.lower()}-prop", "anchor_type": ANCHOR_TYPE_PROP, "members": members, "locked_elements": llm_result.get("locked_elements") or [], "prop_anchor": { "prop_short_id": sid, # carries_printed_content / applicability_evidence MUST be persisted so # the loader gate (06-19 conflation fix) actually fires on fresh # checkpoints. Missing in llm_result (legacy provider) → active default. "carries_printed_content": llm_result.get("carries_printed_content", True), "applicability_evidence": llm_result.get("applicability_evidence", ""), "printed_content": llm_result.get("printed_content", ""), "physical_form": llm_result.get("physical_form", ""), "scale_contract": llm_result.get("scale_contract", ""), }, "per_shot_consumption_hints": hints, } def build_manifest( groups: List[Dict[str, Any]], diagnostics: Dict[str, Any], ) -> Dict[str, Any]: return { "schema_version": SCHEMA_VERSION, "groups": groups, "diagnostics": diagnostics, } # ─────────────────────────── W-B: prop ref prompt overlay ─────────────────────────── def build_prop_ref_prompt_overlay( prop_anchor: Dict[str, Any], canon_t2i: str, *, include_framing: bool = True, ) -> str: """printed_prop anchor → reference image generation prompt overlay (deterministic). Codex 판정 (W_A_STAGE_REVIEW / 설계 D-①): anchor 의 printed_content / physical_form / scale_contract 가 **우선**하되 기존 canon t2i 를 버리지 않고 스타일/맥락 source 로 뒤에 보존한다. canon 자체(EntityCanon.t2i_prompt / entity_t2i cp)는 절대 변경하지 않는다 — 그것이 provenance. include_framing=False 는 description 슬롯용 — ref_image_pipeline 의 prop 템플릿이 product-photo framing 을 이미 제공하므로 (entity_description 으로 조립, t2i_prompt 는 템플릿 부재시 fallback — 2026-06-12 실측) framing 상수를 중복하지 않는다. LLM 재호출 0 — 스파이크2 ref_prompt 패턴의 문자열 조립만. """ # carries_printed_content gate (06-19): a prop that bears no printed/displayed # content of its own must never overlay a printed-content anchor onto its ref — # that conflates it with a different co-occurring printed prop and overrides its # own canon. Such a prop falls back to its own canon prompt. Missing field # (legacy data) defaults to active for backward compatibility. if prop_anchor.get("carries_printed_content", True) is False: return (canon_t2i or "").strip() parts = [] if include_framing: parts.append( "Photorealistic product photo of a single physical prop, isolated on a " "plain neutral background." ) if prop_anchor.get("physical_form"): parts.append(str(prop_anchor["physical_form"]).strip()) if prop_anchor.get("printed_content"): parts.append( "The content printed/displayed on it: " + str(prop_anchor["printed_content"]).strip() ) if prop_anchor.get("scale_contract"): parts.append(str(prop_anchor["scale_contract"]).strip()) overlay = " ".join(parts) canon_t2i = (canon_t2i or "").strip() if canon_t2i: overlay += ( "\nOriginal canon description (style and context reference — where it " "conflicts with the anchor above, the anchor wins): " + canon_t2i ) return overlay # ─────────────────────────── W-C1: revised wide prompt 토큰 audit ─────────────────────────── # 데이터 contract ID 매칭 — detail_steps `_ve_pattern` 과 동일 규약. 의미 추출이 # 아니라 entity short_id 토큰 집합 비교다 (Codex W_C_NARROW_REVIEW 안전안 guard: # revised prompt 는 새 entity ID 토큰 / 새 ref 요구를 추가하면 안 됨 — # ref-contract SOT 는 scene_detail required_refs 불변). _ENTITY_TOKEN_RE = re.compile(r"[CLP]\d{2,3}(?:O\d{2,3})?") def extract_entity_tokens(text: str) -> Set[str]: return set(_ENTITY_TOKEN_RE.findall(text or "")) def revised_prompt_new_tokens(original: str, revised: str) -> Set[str]: """revised 가 original 에 없던 entity ID 토큰을 도입했으면 그 집합 반환. 비어있지 않으면 audit 위반 — caller 는 revised 를 폐기(substitution 미적용) 하고 diagnostic 으로 기록한다 (비차단).""" return extract_entity_tokens(revised) - extract_entity_tokens(original) # ───────────────────── S29 fix: zoom 인물 토큰 ⊆ source 프롬프트 gate ───────────────────── # # 2026-06-12 실측 (S29 외국인 현상): 인물 0 인 wide t2i 가 인물 클로즈업의 # zoom_in_detail source 로 지정되면(상류 모순) REWRITE 가 wide 계약을 지키려 # 인물을 이름으로 주입 → ID 토큰이 없어 ref 미부착 → 모델이 얼굴을 발명. # gate 는 ID 조인만: zoom 멤버 t2i 의 인물(C) 토큰 base(의상 O suffix 제거)가 # source variation 의 토큰 base 에 모두 존재해야 그 variation 재작문/그룹 성립. # scene_detail 의 VE 위반 검사(프롬프트 토큰 ⊆ VE) 불변식 하에서 이 토큰 조인은 # "zoom VE 인물 ⊆ source VE" 조인을 strictly 포함한다 (7그룹 실측 판정 동일). _CHAR_BASE_RE = re.compile(r"^C\d{2,3}") def char_token_bases(text: str) -> Set[str]: """인물(C) entity ID 토큰의 base 집합 — C08O06 → C08. L/P 토큰 제외.""" bases: Set[str] = set() for tok in extract_entity_tokens(text): m = _CHAR_BASE_RE.match(tok) if m: bases.add(m.group(0)) return bases def zoom_subject_missing_chars( zoom_prompts: List[str], source_prompt: str, ) -> Set[str]: """zoom 멤버 프롬프트들의 인물 토큰 base 중 source 프롬프트에 없는 것. 비어있지 않으면 그 source variation 은 zoom 의 피사체를 담을 수 없다 — caller 는 해당 variation 재작문을 skip 하고, 전 variation 이 그러면 seed 자체를 skip(reason="zoom_subject_absent_in_source") 한다.""" zoom_chars: Set[str] = set() for p in zoom_prompts or []: zoom_chars |= char_token_bases(p) return zoom_chars - char_token_bases(source_prompt or "") # ─────────────────────────── 조인 무결성 검증 (얇게) ─────────────────────────── def validate_anchor_manifest( manifest: Dict[str, Any], selected_map: Dict[int, Set[int]], known_prop_short_ids: Set[str], *, extra_known_shots: Optional[Set[ShotKey]] = None, known_char_short_ids: Optional[Set[str]] = None, ) -> List[str]: """schema parse + 조인 무결성만 — 의미 게이트 없음 (Codex 가이드 5). Returns: violation 문자열 목록 (빈 목록 = 통과). """ violations: List[str] = [] extra_known_shots = extra_known_shots or set() groups = manifest.get("groups") if not isinstance(groups, list): return ["groups must be a list"] seen_group_ids: Set[str] = set() for g in groups: gid = g.get("group_id") or "" if gid in seen_group_ids: violations.append(f"{gid}: duplicate group_id") seen_group_ids.add(gid) atype = g.get("anchor_type") if atype not in (ANCHOR_TYPE_ZOOM, ANCHOR_TYPE_PROP, ANCHOR_TYPE_IMMOBILIZED): violations.append(f"{gid}: invalid anchor_type {atype!r}") continue members = g.get("members") or [] if len(members) < 2: violations.append(f"{gid}: needs >=2 members") roles: List[str] = [] for m in members: si, shi = m.get("scene_index"), m.get("shot_index") roles.append(m.get("role", "")) key = (si, shi) if ( not isinstance(si, int) or not isinstance(shi, int) or (shi not in selected_map.get(si, set()) and key not in extra_known_shots) ): violations.append(f"{gid}: member S{si}sh{shi} not a known selected shot") for le in g.get("locked_elements") or []: if not str(le.get("evidence_quote", "")).strip(): violations.append(f"{gid}: locked_element without evidence_quote") hints = g.get("per_shot_consumption_hints") or {} if atype == ANCHOR_TYPE_ZOOM: if ROLE_SOURCE_WIDE not in roles or ROLE_ZOOM not in roles: violations.append(f"{gid}: zoom group needs source_wide + zoom roles") for key_s, hint in hints.items(): if hint.get("crop_from_source") and ROLE_SOURCE_WIDE not in roles: violations.append(f"{gid}: crop_from_source without source_wide ({key_s})") elif atype == ANCHOR_TYPE_PROP: sid = (g.get("prop_anchor") or {}).get("prop_short_id") if sid not in known_prop_short_ids: violations.append(f"{gid}: unknown prop_short_id {sid!r}") else: # ANCHOR_TYPE_IMMOBILIZED csid = (g.get("subject_anchor") or {}).get("character_short_id") if not csid: violations.append(f"{gid}: immobilized group missing character_short_id") elif known_char_short_ids is not None and csid not in known_char_short_ids: violations.append(f"{gid}: unknown character_short_id {csid!r}") return violations