"""W21B-W7 (2026-06-12): VisualContinuityAnchorStep — 시각 연속성 anchor manifest. order 21.66 (episode_reference_policy 21.65 직후 / scene_detail 21.70 직전). ★v1 production scope = printed_prop(D) anchors 전용 — zoom continuity(C) 는 zoom_continuity_anchor(21.73)로 이동 (W-C1, Codex C3 변형: refined ref_usage 가 shot_dependency_t2i 에만 실재해 이 위치에선 fresh-run C seed 0 — SOT 1곳). 이 step 의 zoom/exact 후보는 diagnostics 로만 기록한다. consumer (W-B: ref_image_gen prop overlay + scene_detail scale 라벨) 는 cp 를 읽기만 한다. 흐름: 1. seed 탐지 (deterministic, ``visual_continuity_anchor_plan``): C = refined ref_usage==zoom_in_detail dep pair — shot_dependency_t2i (21.71) cp 의 soft 역참조. ref_usage 는 refined cp 에만 실재 (shot_dependency 18.1 은 None — 2026-06-12 실측). fresh run 첫 pass 엔 C seed 0 + 진단 (2-pass 구조는 scene_detail 의 기존 consumes_downstream 패턴과 동일). D = prop short_id 의 selected-shot 반복 (VE + scene_detail required_refs soft) + framing structured-enum priority score. 2. 선별 seed 마다 LLM anchor 추출 (provider, 그룹당 1콜, 씬 원문 전체 — 자르지 않음). LLM 실패는 skipped(llm_error) 비차단. 3. manifest 조립 + 조인 무결성 검증(얇게) + review_html 갤러리. opt-in: settings.visual_continuity_anchor_enabled (default False) → applicability="if_visual_continuity_anchor_enabled". OFF 시 not_applicable cp 만 — default 경로 영향 0. anchor 품질은 deterministic 테스트 비대상 — canary + 육안 gate. 설계: docs/w21b-w7-visual-continuity-anchor-production-brief-20260612/. """ from __future__ import annotations import hashlib import html import json import logging from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Set, Tuple from app.core.step_runner import StepRunner from app.modules.pipeline.visual_continuity_anchor_plan import ( ANCHOR_TYPE_IMMOBILIZED, DEFAULT_D_RESERVED_SLOTS, DEFAULT_GROUP_CAP, DEFAULT_IMMOBILIZED_CAP, ROLE_ENVIRONMENT, SCHEMA_VERSION, build_immobilized_prev_frame_plan, build_immobilized_subject_group, build_manifest, build_prop_group, detect_c_seeds, detect_d_seeds, build_identity_family_by_sid, detect_immobilized_subject_seeds, select_immobilized_seeds_with_cap, select_seeds_with_cap, validate_anchor_manifest, ) logger = logging.getLogger(__name__) _HTML_SUBDIR_NAME = "review_html" ShotKey = Tuple[int, int] # ───────────────────── W-B consumer 진입점 (module-level) ───────────────────── def load_printed_prop_anchor_context(project_id: str, episode_id: str) -> Dict[str, Any]: """printed_prop anchor consumer 공용 로더 (ref_image_gen / scene_detail / attach 라벨). flag OFF / cp 부재·미완료 / printed_prop 그룹 0 이면 빈 dict — 소비자 전원 no-op (default 경로 byte-identical). Returns (비어있지 않으면): { "by_prop": {prop_short_id: prop_anchor dict}, "members_by_shot": {(si, shi): [prop_short_id, ...]}, "keep_real_scale_by_shot": {(si, shi): {prop_short_id, ...}}, "stamp": {schema_version, config_hash, group_count, digest} — scene_detail config drift 용 (Codex 판정 ③: manifest 전체가 아닌 checkpoint stamp 수준). } """ from app.core.config import settings if not bool(getattr(settings, "visual_continuity_anchor_enabled", False)): return {} cp_path = ( Path(settings.projects_dir) / project_id / "checkpoints" / "episodes" / episode_id / "visual_continuity_anchor" / "manifest.json" ) if not cp_path.exists(): return {} try: cp = json.loads(cp_path.read_text(encoding="utf-8")) except Exception as exc: logger.warning("visual_continuity_anchor: consumer cp parse 실패: %s", exc) return {} if cp.get("status") != "completed": return {} groups = (cp.get("data") or {}).get("groups") or [] prop_groups = [ g for g in groups if g.get("anchor_type") == "printed_prop" and (g.get("prop_anchor") or {}).get("prop_short_id") # carries_printed_content gate: a prop that bears no printed/displayed # content of its own (a painted mark/symbol, tool, plant, …) must NOT # impose a printed-content anchor — that conflates it with a different # co-occurring printed prop and overrides its own canon. Missing field # (legacy checkpoints) defaults to active for backward compatibility. and (g.get("prop_anchor") or {}).get("carries_printed_content", True) is not False ] if not prop_groups: return {} by_prop: Dict[str, Dict[str, Any]] = {} members_by_shot: Dict[ShotKey, List[str]] = {} keep_by_shot: Dict[ShotKey, Set[str]] = {} for g in prop_groups: sid = g["prop_anchor"]["prop_short_id"] by_prop[sid] = dict(g["prop_anchor"]) for m in g.get("members") or []: si, shi = m.get("scene_index"), m.get("shot_index") if isinstance(si, int) and isinstance(shi, int): members_by_shot.setdefault((si, shi), []).append(sid) for key_s, hint in (g.get("per_shot_consumption_hints") or {}).items(): if not (hint or {}).get("keep_real_scale"): continue try: si_s, shi_s = key_s.split(":", 1) keep_by_shot.setdefault((int(si_s), int(shi_s)), set()).add(sid) except (ValueError, AttributeError): continue digest = hashlib.sha256( json.dumps(groups, sort_keys=True, ensure_ascii=False).encode("utf-8") ).hexdigest()[:16] return { "by_prop": by_prop, "members_by_shot": members_by_shot, "keep_real_scale_by_shot": keep_by_shot, "stamp": { "schema_version": cp.get("schema_version"), "config_hash": cp.get("config_hash"), "group_count": len(groups), "digest": digest, }, } def load_immobilized_subject_anchor_context( project_id: str, episode_id: str ) -> Dict[str, Any]: """P8 immobilized_subject anchor consumer 로더 (scene_detail 주입 전용). printed_prop loader 와 분리 — 자기 anchor_type(immobilized_subject) 그룹만 읽어 old/new manifest 호환을 깔끔히 유지한다 (Codex 합의 ⑦). 두 flag (visual_continuity_anchor_enabled AND immobilized_subject_continuity_enabled) 이 모두 ON 이고 cp 가 completed 이며 immobilized 그룹이 있을 때만 비어있지 않은 dict 를 반환 — 그 외엔 빈 dict = 소비자 no-op(default 경로 byte-identical). Returns (비어있지 않으면): { "anchors_by_group": {group_id: subject_anchor dict (+ "group_id")}, "members_by_shot": {(si, shi): [group_id, ...]}, "stamp": {schema_version, config_hash, group_count, digest}, } ★ key 는 character_short_id 가 아니라 group_id 다 (Codex 리뷰 BLOCKING1): 같은 인물이 여러 씬에서 각각 immobilized group 을 가지면 character_short_id 단일 키는 마지막 그룹으로 덮여 한 씬의 계약이 다른 씬 샷에 새는 generic 결함이 생긴다. group_id 키 + members_by_shot 이 group_id 를 가리키면 충돌 0. 빈 shared_state_contract 그룹은 소비처에서 무의미 주입이 되므로 제외한다 (Codex 리뷰 MINOR1 — group 생성 단계에서도 막지만 loader 도 방어). """ from app.core.config import settings if not bool(getattr(settings, "visual_continuity_anchor_enabled", False)): return {} if not bool(getattr(settings, "immobilized_subject_continuity_enabled", False)): return {} cp_path = ( Path(settings.projects_dir) / project_id / "checkpoints" / "episodes" / episode_id / "visual_continuity_anchor" / "manifest.json" ) if not cp_path.exists(): return {} try: cp = json.loads(cp_path.read_text(encoding="utf-8")) except Exception as exc: logger.warning("immobilized_subject: consumer cp parse 실패: %s", exc) return {} if cp.get("status") != "completed": return {} groups = (cp.get("data") or {}).get("groups") or [] imm_groups = [ g for g in groups if g.get("anchor_type") == ANCHOR_TYPE_IMMOBILIZED and g.get("group_id") and (g.get("subject_anchor") or {}).get("character_short_id") and str((g.get("subject_anchor") or {}).get("shared_state_contract") or "").strip() ] if not imm_groups: return {} anchors_by_group: Dict[str, Dict[str, Any]] = {} members_by_shot: Dict[ShotKey, List[str]] = {} # P8 I-1: group→members (role 포함) — registered pose guide attach 가 그룹의 # environment 멤버 background plate 를 underlay 로 찾는 데 쓴다. members_by_group: Dict[str, List[Dict[str, Any]]] = {} for g in imm_groups: gid = g["group_id"] sa = dict(g["subject_anchor"]) sa["group_id"] = gid anchors_by_group[gid] = sa for m in g.get("members") or []: si, shi = m.get("scene_index"), m.get("shot_index") if isinstance(si, int) and isinstance(shi, int): members_by_shot.setdefault((si, shi), []).append(gid) members_by_group.setdefault(gid, []).append({ "scene_index": si, "shot_index": shi, "role": m.get("role"), }) digest = hashlib.sha256( json.dumps(imm_groups, sort_keys=True, ensure_ascii=False).encode("utf-8") ).hexdigest()[:16] return { "anchors_by_group": anchors_by_group, "members_by_shot": members_by_shot, "members_by_group": members_by_group, "stamp": { "schema_version": cp.get("schema_version"), "config_hash": cp.get("config_hash"), "group_count": len(imm_groups), "digest": digest, }, } # P8 I-1: registered (bg-aware) immobilized pose guide ref 라벨 — pose/contact/support # surface 전용 시각 SOT. 실인간 렌더(identity=character state ref+본문, 환경=bg plate)는 # 유지하고 마네킹/관절선/wooden look 은 복사금지 — 강한 real-human 단언은 prompt_service # render branch(I-3 REAL_HUMAN_ASSERT)가 담당. 시나리오 토큰 0. IMMOBILIZED_POSE_GUIDE_LABEL = ( "POSE / SUPPORT GUIDE for the immobilized figure (a rough line-art mannequin " "registered over a faint room): use it ONLY to copy the body's pose, every " "contact point, and which support surface the body rests on — keep the figure " "supported on the same surface at the same place, never floating. Do NOT copy " "its drawing lines, flat tone, wooden look, ball joints or featureless head; the " "rendered figure is a REAL human person whose appearance comes from the character " "reference and whose environment comes from the background plate, NOT from this guide." ) def _resolve_group_env_plate( members: List[Dict[str, Any]], current_key: ShotKey, bg_map: Dict[str, Any], ) -> Tuple[Optional[bytes], str]: """그룹의 environment plate bytes + bg_key 해결. 현재 샷 → environment-role 멤버 → 임의 멤버 순으로 background_chain_bg_map 에서 image_bytes 보유 entry 를 찾는다 (location 재추론 0 — coordinator 가 쓰는 동일 SOT 재사용, subspace mismatch 방지).""" csi, cshi = current_key ordered_keys: List[str] = [] if isinstance(csi, int) and isinstance(cshi, int): ordered_keys.append(f"{csi}_{cshi}") env_members = [m for m in members if m.get("role") == ROLE_ENVIRONMENT] for m in env_members + members: k = f"{m.get('scene_index')}_{m.get('shot_index')}" if k not in ordered_keys: ordered_keys.append(k) for k in ordered_keys: e = bg_map.get(k) or {} b = e.get("image_bytes") if b: return b, str(e.get("bg_id") or k) return None, "" def attach_registered_pose_guide_ref( labeled_refs: List[Any], ref_roles: List[str], ref_role_metadata: List[Dict[str, Any]], attached_meta: List[Any], *, scene_index: Optional[int], shot_index: Optional[int], anchor_ctx: Dict[str, Any], background_chain_bg_map: Dict[str, Any], cache_dir: Path, openai_client: Any = None, force: bool = False, ) -> bool: """P8 I-1 — immobilized group 멤버 샷에 bg-aware 등록 pose guide PNG 를 ``immobilized_pose_guide`` ref 로 append (4-list parallel mutate). flag(immobilized_registered_pose_guide_enabled) OFF → no-op(byte-identical). 생성 책임은 registered_pose_guide_service 에 위임(gate/cache/underlay/edit/diag, Codex 합의: coordinator/step 은 orchestration). bg plate = 이 그룹의 environment 멤버(또는 현재 샷)의 background_chain plate bytes(location 재추론 X). 생성 실패/plate 부재/계약 부재 → no-op(★white-bg fallback 없음 = 시각 가이드 없이 text 계약+dead ref 로 degrade). worker thread 안전 — db 접근 0, list mutate + 파일 IO 만.""" from app.core.config import settings if not bool(getattr(settings, "immobilized_registered_pose_guide_enabled", False)): return False if not anchor_ctx: return False members_by_shot = anchor_ctx.get("members_by_shot") or {} anchors_by_group = anchor_ctx.get("anchors_by_group") or {} members_by_group = anchor_ctx.get("members_by_group") or {} gids = members_by_shot.get((scene_index, shot_index)) or [] if not gids: return False from app.services import registered_pose_guide_service as rpg model = str(getattr( settings, "immobilized_registered_pose_guide_model", "gpt-image-2.5-sunburst")) variant = str(getattr( settings, "immobilized_registered_pose_guide_underlay", "original")) attached = False for gid in gids: subject_anchor = anchors_by_group.get(gid) or {} env_bytes, bg_key = _resolve_group_env_plate( members_by_group.get(gid) or [], (scene_index, shot_index), background_chain_bg_map) png, diag = rpg.build_registered_pose_guide( group_id=gid, subject_anchor=subject_anchor, env_bg_bytes=env_bytes, bg_key=bg_key, cache_dir=cache_dir, openai_client=openai_client, model=model, underlay_variant=variant, force=force, ) if not png: logger.info( "registered_pose_guide: S%ssh%s no-guide degrade (group=%s, reason=%s)", scene_index, shot_index, gid, diag.get("reason")) continue visible_focus = (subject_anchor.get("per_shot_visible_focus") or {}).get( str(shot_index), "") labeled_refs.append((IMMOBILIZED_POSE_GUIDE_LABEL, png)) ref_roles.append("immobilized_pose_guide") ref_role_metadata.append({ "group_id": gid, "character_short_id": subject_anchor.get("character_short_id"), "visible_focus": visible_focus, # P8 Fix2 (Codex BLOCKING2): prompt_service 가 "lifeless" 단정을 dead 일 # 때만 쓰도록 subject_state 전달 (상태 일반화 방지). "subject_state": subject_anchor.get("subject_state"), # P0 (2026-07-01, Codex 합의): worker-thread 라 asset_id 는 여기서 못 얻지만 # (capture 가 UUID 동기반환 X), persistence 가 라벨없이 post-hoc resolve # 하도록 구조키 보존. DB lookup role 은 capture role 인 registered_pose_guide # (ref_role 은 immobilized_pose_guide 로 유지). bg_key/guide_hash 는 exact # match 용, group_id 는 fallback pair 용. "pipeline_role": "registered_pose_guide", "bg_key": bg_key, "guide_hash": diag.get("hash"), }) attached_meta.append( ("immobilized_pose_guide", f"{scene_index}:{shot_index}")) logger.info( "registered_pose_guide: S%ssh%s 등록 가이드 부착 (group=%s, status=%s)", scene_index, shot_index, gid, diag.get("status")) attached = True return attached # A5 (2026-07-02): immobilized prev 완성프레임 연속성 ref 라벨 — 같은 immobilized # 인물의 상태(포즈/접촉점/지지면/소품/방 상태) 연속성 SOT. 구도/프레이밍은 본문 # 프롬프트가 SOT — composition hint 가 아님을 명시(Codex 합의). 시나리오 토큰 0. IMMOBILIZED_PREV_FRAME_LABEL = ( "CONTINUITY REFERENCE (a completed earlier frame from this same scene showing " "the same immobilized figure): keep that figure's exact pose, every contact " "point, the surface supporting the body, its nearby props and the room state " "consistent with this frame. Use it ONLY for state continuity — the current " "shot's framing, camera angle and crop come from the text prompt, NOT from " "this reference." ) # 연속성 anchor 로 bytes 교체 가능한 prev-shot ref role — outdoor option C 와 동일 # 집합 (zoom crop 계열 previous_shot_same_frame_zoomed 는 제외: zoom provenance 충돌). _IMMOBILIZED_PREV_FRAME_REPLACE_ROLES = frozenset({ "previous_shot_same_room", "previous_shot_continuity", }) def attach_immobilized_prev_frame_ref( labeled_refs: List[Any], ref_roles: List[str], ref_role_metadata: List[Dict[str, Any]], attached_meta: List[Any], *, scene_index: Optional[int], shot_index: Optional[int], anchor_ctx: Dict[str, Any], source_bytes_resolver: Optional[Callable[[ShotKey], Optional[bytes]]], source_still_id_by_key: Optional[Dict[ShotKey, str]] = None, diag_out: Optional[Dict[str, Any]] = None, ) -> bool: """A5 (2026-07-02) — immobilized 그룹 후속 멤버 샷에 선행 environment 멤버의 **현재-run/영속 primary 완성 프레임**을 연속성 ref 로 부착 (4-list parallel mutate). star-to-environment (Codex 합의): 계획은 ``build_immobilized_prev_frame_plan`` (manifest members_by_group 의 structured role 조인만). anchor 프레임 bytes 는 호출자 resolver(현재-run path map / 단건은 resume state)로만 해석 — **stale DB fallback 없음** (outdoor option C 계약 동일). 기존 prev-shot 계열 ref 가 있으면 bytes+라벨을 in-place 교체(image index 불변), 없으면 append. zoom crop ref 존재 시 no-op (zoom provenance 충돌 회피). lineage: worker thread 라 asset_id 를 못 얻음 → 구조키 ``source_still_id`` 를 metadata 에 보존해 persistence 가 post-hoc resolve (registered_pose_guide 와 동일 패턴, P0 계약 — UUID SOT 와 bytes ref 미혼합). ``diag_out`` (호출자 제공 dict) 에 Codex 합의 진단 필드(previous_frame_required / resolved / source_selection / source_still_id / reason_if_missing)를 채운다 — 멤버 샷(계획 엔트리 보유)일 때만. flag OFF / 비멤버 → no-op byte-identical. """ from app.core.config import settings if not bool(getattr(settings, "immobilized_prev_frame_chain_enabled", False)): return False if not anchor_ctx: return False if not (isinstance(scene_index, int) and isinstance(shot_index, int)): return False plan = build_immobilized_prev_frame_plan( anchor_ctx.get("members_by_group") or {}) entry = plan.get((scene_index, shot_index)) if not entry: return False src_key: ShotKey = tuple(entry["anchor_source"]) # type: ignore[assignment] gid = entry.get("group_id") src_sid = (source_still_id_by_key or {}).get(src_key) diag: Dict[str, Any] = { "lane": "immobilized", "previous_frame_required": True, "resolved": False, "source_selection": "environment_member", "anchor_source": [src_key[0], src_key[1]], "group_id": gid, "source_still_id": src_sid, "reason_if_missing": None, } if diag_out is not None: diag_out.update(diag) def _diag(k: str, v: Any) -> None: diag[k] = v if diag_out is not None: diag_out[k] = v # zoom crop ref 존재 → zoom_continuity 와 충돌하므로 no-op (outdoor 동일 방어). if "previous_shot_same_frame_zoomed" in ref_roles: _diag("reason_if_missing", "zoom_conflict") logger.info( "immobilized_prev_frame: S%ssh%s no-op — zoom crop ref 존재 (충돌 회피)", scene_index, shot_index) return False src_bytes: Optional[bytes] = None if source_bytes_resolver is not None: try: src_bytes = source_bytes_resolver(src_key) except Exception as exc: logger.warning( "immobilized_prev_frame: S%ssh%s anchor S%dsh%d bytes resolve 실패 " "(비차단): %s", scene_index, shot_index, src_key[0], src_key[1], exc) src_bytes = None if not src_bytes: _diag("reason_if_missing", "previous_frame_required_missing") logger.warning( "immobilized_prev_frame: S%ssh%s 연속성 anchor 미부착 — env 멤버 " "S%dsh%d 완성 프레임 부재 (stale fallback 안 함, group=%s)", scene_index, shot_index, src_key[0], src_key[1], gid) return False meta_anchor: Dict[str, Any] = { "immobilized_prev_frame_anchor": True, "anchor_source": [src_key[0], src_key[1]], "group_id": gid, # persistence post-hoc resolve 용 구조키 (P0 패턴 — 라벨 파싱 금지). "pipeline_role": "immobilized_prev_frame", } if src_sid: meta_anchor["source_still_id"] = src_sid # 기존 prev-shot 계열 ref bytes 를 anchor 프레임으로 in-place 교체 (image index 불변). for i, role in enumerate(ref_roles): if role in _IMMOBILIZED_PREV_FRAME_REPLACE_ROLES: labeled_refs[i] = (IMMOBILIZED_PREV_FRAME_LABEL, src_bytes) ref_role_metadata[i] = {**ref_role_metadata[i], **meta_anchor} _diag("resolved", True) _diag("replaced_role", role) logger.info( "immobilized_prev_frame: S%ssh%s prev-shot ref(idx=%d role=%s) " "bytes 를 env 멤버 S%dsh%d 완성 프레임으로 교체 (group=%s)", scene_index, shot_index, i, role, src_key[0], src_key[1], gid) return True labeled_refs.append((IMMOBILIZED_PREV_FRAME_LABEL, src_bytes)) ref_roles.append("previous_shot_same_room") ref_role_metadata.append(meta_anchor) attached_meta.append(("background_prev_shot", "")) _diag("resolved", True) logger.info( "immobilized_prev_frame: S%ssh%s env 멤버 S%dsh%d 완성 프레임 append " "(group=%s)", scene_index, shot_index, src_key[0], src_key[1], gid) return True # P02 전수조사 fix (2026-06-12): anchored prop 이 등장하는 still 의 previous-shot # 계열 background ref 가 그 prop 의 옛 위치(예: 벽에 꽂힌 상태)를 그대로 운반해 # 한 프레임에 같은 실물이 중복 등장하는 실측 결함 (S25sh7) — 해당 role 에만 # generic 사본 무시 지시를 metadata["ignore"] 로 덧붙인다. # ※ prompt_service 의 ignore 채널을 실제로 출력하는 role 만 포함 (Codex 리뷰 # 실측: previous_shot_continuity / background_chain_ref 분기는 ignore 미소비 — # prompt_service 가 채널을 얻으면 추가). zoomed-frame role 은 의도적 제외 # (crop 관계는 source 프레임 전체가 SOT). _PREVIOUS_SHOT_BG_ROLES = frozenset({ "previous_shot_same_room", }) # generic boilerplate — 시나리오 토큰 0, 내용은 전부 anchor 데이터가 결정. _SINGLE_OBJECT_IDENTITY_SENTENCE = ( "This is the SAME single physical object in every shot it appears in — match " "this reference's printed content, crease pattern, stains and wear EXACTLY." ) # previous-shot bg role 의 프롬프트 표기는 prompt_service 가 role 기반 고정 # 문자열로 쓰므로(라벨 미표시 — 2026-06-12 실측) metadata["ignore"] 채널로 # 전달한다 ("- from image {i}: ignore {…}" 로 출력되는 기존 계약). _BG_STALE_COPY_IGNORE = ( "any copy of the separately-referenced printed/displayed object appearing " "anywhere in this background in ANY form (pinned, taped, framed, hung or " "lying around) — in the current shot exactly ONE such object exists, located " "only where the prompt text places it" ) def build_zoom_fill_prop_refs( project_id: str, episode_id: str, visible_entities: List[Dict[str, Any]], ref_image_map: Dict[str, bytes], *, scene_index: Optional[int], shot_index: Optional[int], max_refs: int = 1, ) -> Tuple[List[Tuple[str, bytes]], bool]: """zoom continuity crop/i2i-fill 경로의 object reference 목록 (단건/배치 공용). S12sh12 인쇄 내용 발명 fix (2026-06-12): crop 경로의 prop ref 라벨은 이름만 실려 anchor 계약이 fill 에 도달하지 않았다 — 이 still 이 printed_prop anchor 멤버면 라벨에 printed_content + physical_form + scale_contract + 동일 실물 identity 문장을 덧붙인다 (attach 경로의 ``apply_keep_real_scale_to_prop_labels`` 와 같은 데이터 SOT, 라벨 채널만 crop 경로 형식). anchored prop 을 cap 보다 앞에 두어 ``max_refs`` 절단으로 탈락하지 않게 한다. flag OFF / anchor 부재 still 은 기존 라벨 그대로 (byte-identical). 파일 read 만 — batch ThreadPool worker 안전 (db 비사용). Returns ``(refs, has_anchor_contract)`` — 두 번째 값은 반환 refs 중 anchor 계약이 실제 적용된 ref 존재 여부. fill 프롬프트의 object-ref content-SOT 절은 이 값이 True 일 때만 붙는다 (Codex S12_FILL_PROP_REVIEW narrow: 일반 prop ref 만 있는 still 의 fill 프롬프트는 byte-identical 유지 — 인쇄/표시 내용 SOT 예외는 printed_prop anchor 계약 ref 에만 연다). """ anchor_ctx = load_printed_prop_anchor_context(project_id, episode_id) member_sids: Set[str] = set() by_prop: Dict[str, Dict[str, Any]] = {} if anchor_ctx and isinstance(scene_index, int) and isinstance(shot_index, int): member_sids = set( anchor_ctx.get("members_by_shot", {}).get((scene_index, shot_index)) or []) by_prop = anchor_ctx.get("by_prop", {}) candidates: List[Tuple[bool, str, bytes]] = [] for ve in visible_entities: if ve.get("entity_type") != "prop" or ve.get("id") not in ref_image_map: continue label = f"Reference image 2 (object reference): {ve.get('name', '')}" sid = str(ve.get("short_id") or "") anchored = bool(sid and sid in member_sids and sid in by_prop) if anchored: anchor = by_prop.get(sid) or {} parts = [ str(anchor.get("printed_content") or "").strip(), str(anchor.get("physical_form") or "").strip(), str(anchor.get("scale_contract") or "").strip(), _SINGLE_OBJECT_IDENTITY_SENTENCE, ] extra = " ".join(p for p in parts if p) label = f"{label} — {extra}" candidates.append((anchored, label, ref_image_map[ve["id"]])) candidates.sort(key=lambda c: not c[0]) # anchored 먼저, 그 외 원래 순서 유지(stable) selected = candidates[:max_refs] return ( [(label, img) for _, label, img in selected], any(anchored for anchored, _, _ in selected), ) def apply_keep_real_scale_to_prop_labels( labeled_refs: List[Tuple[str, bytes]], ref_roles: List[str], ref_role_metadata: List[Dict[str, Any]], *, scene_index: Optional[int], shot_index: Optional[int], anchor_ctx: Dict[str, Any], ) -> List[Tuple[str, bytes]]: """attach anchor 계약 — keep_real_scale hint 가 있는 still 에서 1. prop_ref 라벨에 scale_contract + physical_form + 동일 실물 identity 문장 (스파이크2 라벨 패턴 + P02 전수조사 마모 편차 fix), 2. previous-shot 계열 background ref 의 metadata["ignore"] 에 stale 사본 무시 지시 (P02 전수조사 S25sh7 중복 fix — prompt_service 가 bg role 표기를 고정 문자열로 쓰므로 ignore 채널이 유일한 프롬프트 도달 경로) 를 적용한다. labeled_refs 는 새 list 반환, ref_role_metadata 의 해당 bg entry 는 in-place 갱신(기존 ignore 와 병합). anchor_ctx 빈 dict / hint 없는 still / 해당 role 아닌 ref 는 원본 그대로 (byte-identical). batch path 와 단건 regen path 가 같은 helper 를 호출한다 (post-pass 금지 — Codex 판정 ②와 동일 원칙). """ if not anchor_ctx or not isinstance(scene_index, int) or not isinstance(shot_index, int): return labeled_refs keep_sids = anchor_ctx.get("keep_real_scale_by_shot", {}).get((scene_index, shot_index)) if not keep_sids: return labeled_refs by_prop = anchor_ctx.get("by_prop", {}) out: List[Tuple[str, bytes]] = [] for i, (label, img) in enumerate(labeled_refs): role = ref_roles[i] if i < len(ref_roles) else "" sid = (ref_role_metadata[i] or {}).get("sid") if i < len(ref_role_metadata) else None if role == "prop_ref" and sid in keep_sids: anchor = by_prop.get(sid) or {} parts = [ str(anchor.get("scale_contract") or "").strip(), str(anchor.get("physical_form") or "").strip(), _SINGLE_OBJECT_IDENTITY_SENTENCE, ] extra = " ".join(p for p in parts if p) label = f"{label} — {extra}" elif role in _PREVIOUS_SHOT_BG_ROLES and i < len(ref_role_metadata): meta = ref_role_metadata[i] if isinstance(meta, dict): existing = str(meta.get("ignore") or "").strip() meta["ignore"] = ( f"{existing}; {_BG_STALE_COPY_IGNORE}" if existing else _BG_STALE_COPY_IGNORE ) out.append((label, img)) return out class VisualContinuityAnchorStep(StepRunner): """Step 21.66: 시각 연속성 anchor manifest (W21B-W7).""" # Test-only injection slots (production resolves provider functions). _zoom_anchor_override: Optional[Callable[..., Dict[str, Any]]] = None _prop_anchor_override: Optional[Callable[..., Dict[str, Any]]] = None _immobilized_anchor_override: Optional[Callable[..., Dict[str, Any]]] = None def set_overrides_for_testing( self, *, zoom=None, prop=None, immobilized=None, ) -> None: self._zoom_anchor_override = zoom self._prop_anchor_override = prop self._immobilized_anchor_override = immobilized # ───────────────────── checkpoint / config plumbing ───────────────────── def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict[str, Any]]: 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(): try: return json.loads(cp.read_text(encoding="utf-8")) except Exception as exc: logger.warning("visual_continuity_anchor: %s parse failed: %s", step_id, exc) return None def _checkpoint_dir(self) -> Path: from app.core.config import settings return ( Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / "visual_continuity_anchor" ) def _config_hash(self) -> str: from app.core.config import settings from app.modules.pipeline.visual_continuity_anchor_provider import ( PROMPT_VERSION, PROVIDER_VERSION, ) payload = { "enabled": bool(getattr(settings, "visual_continuity_anchor_enabled", False)), "group_cap": int(getattr( settings, "visual_continuity_anchor_group_cap", DEFAULT_GROUP_CAP)), "d_reserved_slots": int(getattr( settings, "visual_continuity_anchor_d_reserved_slots", DEFAULT_D_RESERVED_SLOTS)), # P8: immobilized flag/cap fold into config_hash so resume invalidates # when the new anchor type toggles (additive — OFF keeps prior hash modulo # the prompt_version bump which is a one-time invalidation). "immobilized_enabled": bool(getattr( settings, "immobilized_subject_continuity_enabled", False)), "immobilized_cap": int(getattr( settings, "immobilized_subject_continuity_cap", DEFAULT_IMMOBILIZED_CAP)), "schema_version": SCHEMA_VERSION, "prompt_version": PROMPT_VERSION, "provider_version": PROVIDER_VERSION, } return hashlib.sha256( json.dumps(payload, sort_keys=True).encode("utf-8") ).hexdigest()[:16] def _not_applicable(self) -> Dict[str, Any]: return { "applicable_count": 0, "completed_count": 0, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {}, } # ───────────────────── input loaders ───────────────────── @staticmethod def _refined_dependencies(t2i_cp: Optional[Dict[str, Any]]) -> Optional[List[Dict[str, Any]]]: """shot_dependency_t2i cp soft 역참조 — refined ref_usage 의 유일 source. scene_context_loader 의 legacy-cp fail-fast 와 달리 여기선 soft 신호라 legacy schema 는 '부재' 취급 + 진단만 (비차단).""" if not t2i_cp: return None if t2i_cp.get("schema_version") != 4: return None deps = t2i_cp.get("data", {}).get("dependencies") return deps if isinstance(deps, list) else None @staticmethod def _ve_by_shot(shot_director_cp: Optional[Dict[str, Any]]) -> Dict[ShotKey, List[str]]: out: Dict[ShotKey, List[str]] = {} for sc in (shot_director_cp or {}).get("data", {}).get("scenes", []) or []: si = sc.get("scene_index") for sh in sc.get("shots", []) or []: shi = sh.get("shot_index") if isinstance(si, int) and isinstance(shi, int): out[(si, shi)] = list(sh.get("visible_entity_ids") or []) return out @staticmethod def _staging_by_shot(staging_cp: Optional[Dict[str, Any]]) -> Dict[ShotKey, Dict[str, Any]]: out: Dict[ShotKey, Dict[str, Any]] = {} for sh in (staging_cp or {}).get("data", {}).get("shots", []) or []: si, shi = sh.get("scene_index"), sh.get("shot_index") if isinstance(si, int) and isinstance(shi, int): out[(si, shi)] = sh return out @staticmethod def _scene_texts(save_cp: Optional[Dict[str, Any]]) -> Dict[int, str]: out: Dict[int, str] = {} for seg in (save_cp or {}).get("data", {}).get("segments", []) or []: si = seg.get("scene_index") if isinstance(si, int): out[si] = seg.get("text") or "" return out @staticmethod def _shot_descriptions(shot_cp: Optional[Dict[str, Any]]) -> Dict[ShotKey, str]: out: Dict[ShotKey, str] = {} for sc in (shot_cp or {}).get("data", {}).get("scenes", []) or []: si = sc.get("scene_index") for sh in sc.get("shots", []) or []: shi = sh.get("shot_index") if isinstance(si, int) and isinstance(shi, int): out[(si, shi)] = sh.get("description") or "" return out @staticmethod def _required_prop_refs_by_shot( scene_detail_cp: Optional[Dict[str, Any]], ) -> Optional[Dict[ShotKey, Set[str]]]: """scene_detail cp soft 역참조 — required_refs(kind=prop) priority 신호. cp 부재(fresh run) 시 None — seed 차단 아님 (Codex 판정 ④).""" if not scene_detail_cp: return None out: Dict[ShotKey, Set[str]] = {} for sc in scene_detail_cp.get("data", {}).get("scenes", []) or []: si, shi = sc.get("scene_index"), sc.get("_shot_index") if not isinstance(si, int) or not isinstance(shi, int): continue card = sc.get("render_prompt_card") or {} refs = (card.get("asset_requirements") or {}).get("required_refs") or [] sids = { r.get("id") for r in refs if r.get("kind") == "prop" and isinstance(r.get("id"), str) } if sids: out[(si, shi)] = sids return out @staticmethod def _scene_detail_t2i_by_shot( scene_detail_cp: Optional[Dict[str, Any]], ) -> Dict[ShotKey, str]: """현 per-shot 대표 t2i prompt (soft — LLM 입력 보조 컨텍스트).""" out: Dict[ShotKey, str] = {} for sc in (scene_detail_cp or {}).get("data", {}).get("scenes", []) or []: si, shi = sc.get("scene_index"), sc.get("_shot_index") if not isinstance(si, int) or not isinstance(shi, int): continue variations = sc.get("t2i_variations") or [] if variations and isinstance(variations[0], dict): prompt = variations[0].get("t2i_prompt") if isinstance(prompt, str) and prompt.strip(): out[(si, shi)] = prompt return out def _source_still_exists(self, scene_index: int, shot_index: int) -> bool: from app.models.project import SceneStill return ( self.db.query(SceneStill.id) .filter( SceneStill.project_id == self.project_id, SceneStill.episode_id == self.episode_id, SceneStill.scene_index == scene_index, SceneStill.shot_index == shot_index, ) .first() is not None ) @staticmethod def _immobilized_body_pose( staging: Optional[Dict[str, Any]], subject_name: Optional[str] ) -> str: """staging.character_angles 에서 subject_name(exact) 의 immobilized body_pose. 샷별로 독립 생성되는 body_pose 가 드리프트의 원천 — provider 에 reconcile 대상으로 넘긴다. name exact 일치 + is_immobilized_state 만 (글자 판정 0).""" from app.core.subject_state import is_immobilized_state if not staging or not subject_name: return "" for ca in staging.get("character_angles") or []: if (ca.get("character") == subject_name and is_immobilized_state(ca.get("subject_state") or "")): return ca.get("body_pose") or "" return "" @staticmethod def _related_zoom_evidence( zoom_groups: List[Dict[str, Any]], scene_index: int, member_shis: Set[int], ) -> Tuple[List[str], List[Dict[str, Any]]]: """member 샷과 겹치는 zoom_continuity 그룹의 (group_ids, locked_elements). SOT 아님 — provider 입력 참고자료 + manifest related_zoom_group_ids 진단 (Codex Q1). 겹침은 같은 scene_index 안에서 shot_index 교집합으로만 판정.""" related_ids: List[str] = [] prior_locked: List[Dict[str, Any]] = [] for g in zoom_groups or []: g_shis = { m.get("shot_index") for m in g.get("members") or [] if m.get("scene_index") == scene_index } if g_shis & member_shis: gid = g.get("group_id") if gid: related_ids.append(gid) prior_locked.extend(g.get("locked_elements") or []) return related_ids, prior_locked # ───────────────────── execute ───────────────────── def _execute(self, mode: str = "resume") -> Dict[str, Any]: from app.core.config import settings from app.core.errors import AppError if not bool(getattr(settings, "visual_continuity_anchor_enabled", False)): return self._not_applicable() shot_cp = self._load_prev_checkpoint("shot_validator") if not shot_cp or not shot_cp.get("data", {}).get("scenes"): raise AppError(code="step.no_shots", message="shot_validator 결과 없음", status_code=400) from app.core.steps.shot_validator_step import assert_no_failed_scenes assert_no_failed_scenes(shot_cp, self.project_config, consumer_step="visual_continuity_anchor") from app.core.steps.episode_reference_policy_step import build_selected_map_or_raise selected_map = build_selected_map_or_raise(self._load_prev_checkpoint("shot_selection")) refined_deps = self._refined_dependencies( self._load_prev_checkpoint("shot_dependency_t2i")) ve_by_shot = self._ve_by_shot(self._load_prev_checkpoint("shot_director")) staging_by_shot = self._staging_by_shot(self._load_prev_checkpoint("shot_staging")) scene_texts = self._scene_texts(self._load_prev_checkpoint("scene_save")) shot_descriptions = self._shot_descriptions(shot_cp) scene_detail_cp = self._load_prev_checkpoint("scene_detail") required_prop_refs = self._required_prop_refs_by_shot(scene_detail_cp) t2i_data = (self._load_prev_checkpoint("entity_t2i") or {}).get("data", {}) prop_canons = { p.get("short_id"): p for p in t2i_data.get("props", []) or [] if isinstance(p.get("short_id"), str) } character_canons = [ c for c in t2i_data.get("characters", []) or [] if isinstance(c.get("short_id"), str) ] framing_by_shot = { key: sh.get("framing_scale") for key, sh in staging_by_shot.items() } # ── seed 탐지 (deterministic) ── c_seeds: List[Dict[str, Any]] = [] c_skipped: List[Dict[str, Any]] = [] exact_candidates: List[Dict[str, Any]] = [] if refined_deps is not None: c_seeds, c_skipped, exact_candidates = detect_c_seeds( refined_deps, selected_map, source_still_exists=self._source_still_exists, ) ref_usage_source = "shot_dependency_t2i" else: # ref_usage 는 refined cp 에만 실재 (18.1 은 None — 실측) — # fresh run 첫 pass 는 C seed 0 + 진단 (2-pass 시 재실행으로 채움). c_skipped.append({"seed": "*", "reason": "refined_ref_usage_unavailable"}) ref_usage_source = "unavailable" d_seeds = detect_d_seeds( ve_by_shot, selected_map, set(prop_canons.keys()), framing_by_shot=framing_by_shot, required_prop_refs_by_shot=required_prop_refs, ) # ── P8 immobilized_subject seed (deterministic) ── # name→sids exact 조인 source = entity_t2i character canons (display name). # 동명이인은 set 으로 모아 detect 가 ambiguous 로 skip (substring 매칭 0). immobilized_enabled = bool(getattr( settings, "immobilized_subject_continuity_enabled", False)) name_to_sids: Dict[str, Set[str]] = {} for c in character_canons: nm = c.get("name") sid = c.get("short_id") if isinstance(nm, str) and nm and isinstance(sid, str) and sid: name_to_sids.setdefault(nm, set()).add(sid) imm_seeds: List[Dict[str, Any]] = [] imm_skipped: List[Dict[str, Any]] = [] if immobilized_enabled: # ★identity-variant aware (2026-07-02): entity_relation cp 의 # base↔variant(character) 관계로 subject↔VE 를 family 매칭 — 같은 인물의 # variant EntityCanon(환영/시신 외형 등)이 VE 에 있을 때 seed 소실 방지. _relations_cp = ((self._load_prev_checkpoint("entity_relation") or {}) .get("data") or {}).get("relations") or [] identity_family_by_sid = build_identity_family_by_sid(_relations_cp) imm_seeds, imm_skipped = detect_immobilized_subject_seeds( staging_by_shot, ve_by_shot, selected_map, name_to_sids, framing_by_shot=framing_by_shot, identity_family_by_sid=identity_family_by_sid, ) cap = int(getattr(settings, "visual_continuity_anchor_group_cap", DEFAULT_GROUP_CAP)) d_reserved = int(getattr( settings, "visual_continuity_anchor_d_reserved_slots", DEFAULT_D_RESERVED_SLOTS)) # W-C1 (Codex C3 변형): v1 production scope = printed_prop(D) 전용 — # zoom(C) seed 는 diagnostics 로만 기록하고 zoom_continuity_anchor # (21.73, refined cp 가 hard dep = fresh-run proof) 가 SOT. selected_seeds, cap_skipped = select_seeds_with_cap( [], d_seeds, cap, d_reserved=d_reserved) # P8: immobilized seed 는 printed_prop cap 과 독립 cap (Codex 합의 ⑧) — # prop anchor 에 밀려 사라지지 않게 한다. imm_cap = int(getattr( settings, "immobilized_subject_continuity_cap", DEFAULT_IMMOBILIZED_CAP)) imm_selected, imm_cap_skipped = select_immobilized_seeds_with_cap( imm_seeds, imm_cap) # ── LLM anchor 추출 (그룹당 1콜, 실패 비차단) ── prop_fn = self._prop_anchor_override if prop_fn is None: from app.modules.pipeline import visual_continuity_anchor_provider as vca_provider prop_fn = vca_provider.extract_prop_anchor groups: List[Dict[str, Any]] = [] llm_skipped: List[Dict[str, Any]] = [] for anchor_type, seed in selected_seeds: try: sid = seed["prop_short_id"] member_keys = [tuple(m) for m in seed["member_shots"]] member_scene_indices = sorted({si for si, _ in member_keys}) result = prop_fn( [(si, scene_texts.get(si, "")) for si in member_scene_indices], prop_canons.get(sid, {"short_id": sid}), character_canons, [(si, shi, shot_descriptions.get((si, shi), "")) for si, shi in member_keys], project_config=self.project_config, opik_metadata=self.build_opik_metadata(), ) groups.append(build_prop_group( seed, result, framing_by_shot=framing_by_shot)) except Exception as exc: logger.warning( "visual_continuity_anchor: %s anchor 추출 실패 (%s): %s", anchor_type, seed, exc, ) llm_skipped.append({ "seed": json.dumps(seed, ensure_ascii=False), "reason": "llm_error", }) # ── P8 immobilized_subject LLM 추출 (그룹당 1콜, 실패 비차단) ── imm_groups: List[Dict[str, Any]] = [] imm_llm_skipped: List[Dict[str, Any]] = [] if imm_selected: imm_fn = self._immobilized_anchor_override if imm_fn is None: from app.modules.pipeline import visual_continuity_anchor_provider as vca_provider imm_fn = vca_provider.extract_immobilized_subject_anchor sid_to_canon = { c.get("short_id"): c for c in character_canons if isinstance(c.get("short_id"), str) } # zoom_continuity_anchor cp = prior continuity evidence (참고자료, SOT 아님). zoom_groups = ( (self._load_prev_checkpoint("zoom_continuity_anchor") or {}) .get("data", {}).get("groups") or [] ) for seed in imm_selected: try: csid = seed["character_short_id"] si = seed["scene_index"] member_keys = [tuple(m) for m in seed["member_shots"]] member_shis = {shi for _, shi in member_keys} subject_canon = sid_to_canon.get(csid, {"short_id": csid}) subject_name = subject_canon.get("name") member_shots = [ (msi, mshi, shot_descriptions.get((msi, mshi), ""), self._immobilized_body_pose( staging_by_shot.get((msi, mshi)), subject_name), framing_by_shot.get((msi, mshi)) or "") for msi, mshi in member_keys ] related_ids, prior_locked = self._related_zoom_evidence( zoom_groups, si, member_shis) result = imm_fn( scene_texts.get(si, ""), {"short_id": csid, "name": subject_name}, member_shots, prior_locked, project_config=self.project_config, opik_metadata=self.build_opik_metadata(), ) # MINOR1 (Codex): 빈 shared_state_contract 는 소비처에서 # "C04 [dead]: " 무의미 주입을 만든다 — group 생성 자체를 skip. if not str(result.get("shared_state_contract") or "").strip(): imm_llm_skipped.append({ "seed": json.dumps(seed, ensure_ascii=False), "reason": "empty_shared_state_contract", }) continue imm_groups.append(build_immobilized_subject_group( seed, result, framing_by_shot=framing_by_shot, related_zoom_group_ids=related_ids)) except Exception as exc: logger.warning( "visual_continuity_anchor: immobilized anchor 추출 실패 (%s): %s", seed, exc, ) imm_llm_skipped.append({ "seed": json.dumps(seed, ensure_ascii=False), "reason": "llm_error", }) groups.extend(imm_groups) diagnostics = { "seeds_detected": len(c_seeds) + len(d_seeds) + len(imm_seeds), "c_seeds": len(c_seeds), "d_seeds": len(d_seeds), "immobilized_seeds": len(imm_seeds), "groups_extracted": len(groups), "skipped": (c_skipped + cap_skipped + llm_skipped + imm_skipped + imm_cap_skipped + imm_llm_skipped), # W-C1: zoom(C) 후보는 진단만 — SOT 는 zoom_continuity_anchor(21.73). "zoom_seed_candidates": c_seeds, "zoom_anchor_moved_to": "zoom_continuity_anchor", "candidate_same_moment_exact_background": exact_candidates, "ref_usage_source": ref_usage_source, "immobilized_enabled": immobilized_enabled, } manifest = build_manifest(groups, diagnostics) # C seed 의 source 는 selected 가 아니어도 기존 still 로 성립 가능 — # validator 에 알려준다 (조인 무결성용 known shot). extra_known = { tuple(s["source"]) for s in c_seeds } | set(ve_by_shot.keys()) violations = validate_anchor_manifest( manifest, selected_map, set(prop_canons.keys()), extra_known_shots=extra_known, known_char_short_ids={ c.get("short_id") for c in character_canons if isinstance(c.get("short_id"), str) }, ) if violations: raise AppError( code="visual_continuity_anchor.manifest_invalid", message="anchor manifest 조인 무결성 위반: " + "; ".join(violations), status_code=500, ) gallery_rel = self._write_review_html(manifest) # BLOCKING2 (Codex): P8 immobilized anchor 추출 실패는 "비차단 fallback" # 이어야 한다 — failed_count 에 넣으면 StepRunner 가 (groups==0 일 때) # status=failed 로 승격해 scene_detail(depends_on) 을 막는다. imm_llm_skipped # 는 diagnostics.skipped 에만 surface 하고 completed status 를 유지한다 # (printed_prop llm_skipped 의 기존 hard-fail 동작은 이번 범위 밖 — 그대로). return { "applicable_count": len(selected_seeds) + len(imm_selected), "completed_count": len(groups), "failed_count": len(llm_skipped), "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {**manifest, "review_html": gallery_rel}, } # ───────────────────── review gallery (W-A 필수 — Codex) ───────────────────── def _write_review_html(self, manifest: Dict[str, Any]) -> Optional[str]: try: out_dir = self._checkpoint_dir() / _HTML_SUBDIR_NAME out_dir.mkdir(parents=True, exist_ok=True) path = out_dir / "index.html" path.write_text(render_anchor_review_html(manifest), encoding="utf-8") return f"{_HTML_SUBDIR_NAME}/index.html" except Exception as exc: logger.warning("visual_continuity_anchor: review html 생성 실패: %s", exc) return None def render_anchor_review_html(manifest: Dict[str, Any]) -> str: """anchor manifest 의 사람 검토용 HTML — generic 렌더링 (수치/필드 그대로).""" def esc(v: Any) -> str: return html.escape(str(v if v is not None else "")) rows: List[str] = [] for g in manifest.get("groups", []) or []: members = "
".join( f"S{m.get('scene_index')}sh{m.get('shot_index')} {esc(m.get('role'))}" for m in g.get("members", []) ) locked = "".join( f"
  • [{esc(le.get('kind'))}/{esc(le.get('confidence'))}] {esc(le.get('description'))}" f"
    “{esc(le.get('evidence_quote'))}”
    • " for le in g.get("locked_elements", []) ) extra = "" if g.get("wide_shot_contract"): extra += "

    wide_shot_contract

    " if g.get("prop_anchor"): pa = g["prop_anchor"] extra += ( f"

    prop_anchor {esc(pa.get('prop_short_id'))}

    " f"

    printed_content: {esc(pa.get('printed_content'))}

    " f"

    physical_form: {esc(pa.get('physical_form'))}

    " f"

    scale_contract: {esc(pa.get('scale_contract'))}

    " ) if g.get("subject_anchor"): sa = g["subject_anchor"] focus = sa.get("per_shot_visible_focus") or {} focus_html = "".join( f"
  • sh{esc(k)}: {esc(v)}
    • " for k, v in sorted(focus.items()) ) extra += ( f"

    subject_anchor {esc(sa.get('character_short_id'))} " f"{esc(sa.get('subject_state'))}

    " f"

    shared_state_contract: {esc(sa.get('shared_state_contract'))}

    " + (f"

    per_shot_visible_focus

    " if focus_html else "") ) if g.get("related_zoom_group_ids"): extra += ( "

    related_zoom_group_ids: " + esc(", ".join(g["related_zoom_group_ids"])) + "

    " ) hints = json.dumps(g.get("per_shot_consumption_hints", {}), ensure_ascii=False) rows.append( f"

    {esc(g.get('group_id'))} {esc(g.get('anchor_type'))}

    " f"

    {members}

    {extra}" f"

    {esc(hints)}

    " ) diag = json.dumps(manifest.get("diagnostics", {}), ensure_ascii=False, indent=1) return ( "" "visual_continuity_anchor review" "" f"

    visual_continuity_anchor — groups {len(manifest.get('groups', []) or [])}

    " + "".join(rows) + f"

    diagnostics

    {html.escape(diag)}
    " )