"""frame_spatial_contract — opt-in shot 의 화면 좌표/방향 SOT helper. Spec: docs/superpowers/specs/2026-05-14-frame-spatial-contract-design.md 5 함수 (caller 분리): - find_frame_spatial_contract_violations(batch_shots): shot_staging producer 가 retry hint 형성용 violation list 반환. raise X. - validate_and_prepare(contract, visible_entities): render_prompt_card 가 constraint_id 부여 + cross-check. invalid 시 AppError raise. - validate_echoes(card, result): scene_detail post-validation 이 per- variation echo set 검증. violations list 반환. raise X (caller retry 판단). - phrase_diagnostic(t2i_prompt, constraints): soft warning data 반환. raise X. - _format_retry_hint(fsc_violations): retry hint formatter (orientation pattern mirror). """ from __future__ import annotations import logging import re from typing import Any, Dict, List, Optional from app.core.errors import AppError logger = logging.getLogger(__name__) # ── Constants ────────────────────────────────────────────────────────────── ZONE_PHRASES: Dict[str, List[str]] = { "upper_left": ["upper-left", "top-left", "upper left"], "upper_center": ["upper-center", "top-center", "upper center"], "upper_right": ["upper-right", "top-right", "upper right"], "middle_left": ["middle-left", "center-left", "middle left", "left side"], "middle_center": ["middle-center", "center", "middle"], "middle_right": ["middle-right", "center-right", "middle right", "right side"], "lower_left": ["lower-left", "bottom-left", "lower left"], "lower_center": ["lower-center", "bottom-center", "lower center"], "lower_right": ["lower-right", "bottom-right", "lower right"], } DEPTH_PHRASES: Dict[str, List[str]] = { "foreground": ["foreground", "front"], "midground": ["midground", "middle ground"], "background": ["background", "back"], } _FSC_ID_PREFIX = "fsc_" _FSC_MAX_CONSTRAINTS = 3 _VALID_REASON = frozenset([ "movement_direction", "points_to_anchor", "looks_to_anchor", "shared_space_relation", "required_background_position", "primary_subject_isolation", ]) _VALID_TARGET_KIND = frozenset(["character", "prop", "background"]) _VALID_GESTURE_ACTION = frozenset(["none", "points_to", "reaches_for", "looks_toward", "moves_toward"]) _CHARACTER_ID_RE = re.compile(r"^C\d{2,3}$") _PROP_ID_RE = re.compile(r"^P\d{2,3}$") def find_frame_spatial_contract_violations(batch_shots: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """shot_staging producer: violation list 반환 (raise X).""" out: List[Dict[str, Any]] = [] for shot in batch_shots or []: si = shot.get("scene_index") shi = shot.get("shot_index") contract = shot.get("frame_spatial_contract") if contract is None: continue # opt-in null OK if not isinstance(contract, dict): out.append({"scene_index": si, "shot_index": shi, "reason": "contract_not_dict"}) continue reason = contract.get("reason") if reason not in _VALID_REASON: out.append({"scene_index": si, "shot_index": shi, "reason": "invalid_reason_enum", "value": reason}) constraints = contract.get("constraints") if not isinstance(constraints, list) or len(constraints) == 0: out.append({"scene_index": si, "shot_index": shi, "reason": "constraints_empty_or_not_list"}) continue if len(constraints) > _FSC_MAX_CONSTRAINTS: out.append({"scene_index": si, "shot_index": shi, "reason": "constraints_max_3_exceeded", "count": len(constraints)}) for idx, c in enumerate(constraints): if not isinstance(c, dict): out.append({"scene_index": si, "shot_index": shi, "constraint_idx": idx, "reason": "constraint_not_dict"}) continue tk = c.get("target_kind") tid = c.get("target_id", "") label = c.get("label", "") zone = c.get("screen_zone") depth = c.get("depth_plane") ga = c.get("gesture_action") gtl = c.get("gesture_target_label", "") if tk not in _VALID_TARGET_KIND: out.append({"scene_index": si, "shot_index": shi, "constraint_idx": idx, "reason": "invalid_target_kind", "value": tk}) elif tk == "character" and not _CHARACTER_ID_RE.match(tid or ""): out.append({"scene_index": si, "shot_index": shi, "constraint_idx": idx, "reason": "target_kind_character_id_shape_mismatch", "target_id": tid}) elif tk == "prop" and not _PROP_ID_RE.match(tid or ""): out.append({"scene_index": si, "shot_index": shi, "constraint_idx": idx, "reason": "target_kind_prop_id_shape_mismatch", "target_id": tid}) elif tk == "background" and tid != "": out.append({"scene_index": si, "shot_index": shi, "constraint_idx": idx, "reason": "target_kind_background_id_nonempty", "target_id": tid}) if not (isinstance(label, str) and label.strip()): out.append({"scene_index": si, "shot_index": shi, "constraint_idx": idx, "reason": "label_empty"}) if zone not in ZONE_PHRASES: out.append({"scene_index": si, "shot_index": shi, "constraint_idx": idx, "reason": "invalid_screen_zone", "value": zone}) if depth not in DEPTH_PHRASES: out.append({"scene_index": si, "shot_index": shi, "constraint_idx": idx, "reason": "invalid_depth_plane", "value": depth}) if ga not in _VALID_GESTURE_ACTION: out.append({"scene_index": si, "shot_index": shi, "constraint_idx": idx, "reason": "invalid_gesture_action", "value": ga}) else: if ga != "none" and not (isinstance(gtl, str) and gtl.strip()): out.append({"scene_index": si, "shot_index": shi, "constraint_idx": idx, "reason": "gesture_target_label_empty_for_non_none_action"}) if ga == "none" and (gtl or "").strip(): out.append({"scene_index": si, "shot_index": shi, "constraint_idx": idx, "reason": "gesture_target_label_nonempty_for_none_action"}) return out def filter_fsc_constraints_to_visible( contract: Optional[Dict[str, Any]], visible_entities: Optional[List[str]], ) -> Optional[Dict[str, Any]]: """FINDING 6 W4a — consumer-boundary normalization (render_prompt_card path). shot_staging 이 emit 한 frame_spatial_contract constraints 중 character / prop target 이 shot-level visible set 에 없는 constraint 를 deterministic drop 한다. `shot_director.visible_entity_ids` 가 SOT — depicted-but-not- present target (침입손 / 유리 반사상) + shot-level visibility narrowing 으로 target 이 visible 밖이라 `validate_and_prepare` 가 `fsc_cross_check_failed` 로 raise 하던 것을 차단한다. drop 대상 = target_kind ∈ {character, prop} 이고 target_id 가 well-formed C##/P## 이면서 visible set 에 없는 constraint 만. background-kind constraint + visible character/prop constraint 는 **보존**. target_kind 미상 / target_id shape 위반 등 malformed constraint 도 보존 — 직후 `validate_and_prepare` 가 fail-fast 하도록. 모든 constraint 가 drop 되면 None 반환 (no-contract — card path 의 nullable frame_spatial_contract 값과 정합). contract 가 None / non-dict / constraints non-list 면 그대로 pass-through (`validate_and_prepare` 가 처리). """ if contract is None or not isinstance(contract, dict): return contract constraints = contract.get("constraints") if not isinstance(constraints, list): return contract visible_set = set(visible_entities or []) kept: List[Dict[str, Any]] = [] for c in constraints: if isinstance(c, dict): tk = c.get("target_kind") tid = c.get("target_id", "") or "" if tk == "character" and _CHARACTER_ID_RE.match(tid) and tid not in visible_set: logger.warning( "frame_spatial_contract: dropping constraint for " "non-shot-visible character target %s (visible=%s)", tid, sorted(visible_set), ) continue if tk == "prop" and _PROP_ID_RE.match(tid) and tid not in visible_set: logger.warning( "frame_spatial_contract: dropping constraint for " "non-shot-visible prop target %s (visible=%s)", tid, sorted(visible_set), ) continue kept.append(c) if not kept: return None return {**contract, "constraints": kept} def validate_and_prepare( contract: Optional[Dict[str, Any]], visible_entities: List[str], ) -> Optional[Dict[str, Any]]: """render_prompt_card consumer: shape validate + constraint_id assign + visible_entities cross-check. invalid 시 raise AppError. visible_entities 는 SID list (C##, P##, P###...). 기존 build_render_prompt_card signature 와 일관 (render_prompt_card.py:1020/3518). """ if contract is None: return None # No Silent Fallback gate — visible_entities=None 은 producer-side bug # (build_id_policy convention at render_prompt_card.py:1061-1068 일관). if visible_entities is None: raise AppError( code="render_prompt_card.fsc_invalid", message=( "validate_and_prepare: visible_entities is None — caller must " "pass explicit [] for intentionally empty (matches build_id_policy " "convention at render_prompt_card.py:1061-1068)." ), ) if not isinstance(contract, dict): raise AppError( code="render_prompt_card.fsc_invalid", message=f"frame_spatial_contract is {type(contract).__name__}, expected dict or None", ) reason = contract.get("reason") constraints = contract.get("constraints") or [] if reason not in _VALID_REASON: raise AppError(code="render_prompt_card.fsc_invalid", message=f"invalid reason {reason!r}") if not isinstance(constraints, list) or not (1 <= len(constraints) <= _FSC_MAX_CONSTRAINTS): raise AppError(code="render_prompt_card.fsc_invalid", message=f"constraints must be list of 1..{_FSC_MAX_CONSTRAINTS}") # sort by 7-tuple — LLM emit 순서 무관 deterministic id 부여 def _sort_key(c: Dict[str, Any]): return ( c.get("target_kind", ""), c.get("target_id", "") or "", c.get("label", "") or "", c.get("screen_zone", "") or "", c.get("depth_plane", "") or "", c.get("gesture_action", "") or "", c.get("gesture_target_label", "") or "", ) sorted_constraints = sorted(constraints, key=_sort_key) # duplicate detection seen_keys: set = set() for c in sorted_constraints: k = _sort_key(c) if k in seen_keys: raise AppError( code="render_prompt_card.fsc_invalid", message=f"duplicate_constraint (same 7-tuple sort key): {k}", ) seen_keys.add(k) visible_set = set(visible_entities) prepared_constraints = [] for idx, c in enumerate(sorted_constraints): tk = c.get("target_kind") tid = c.get("target_id", "") or "" label = (c.get("label") or "").strip() zone = c.get("screen_zone") depth = c.get("depth_plane") ga = c.get("gesture_action") gtl = c.get("gesture_target_label", "") # shape validation (defensive consumer-side — stale/malformed cp 가 # provider schema 우회해서 들어왔을 때도 fail-fast). if tk not in _VALID_TARGET_KIND: raise AppError(code="render_prompt_card.fsc_invalid", message=f"invalid target_kind {tk!r}") if not label: raise AppError(code="render_prompt_card.fsc_invalid", message="label is empty") if zone not in ZONE_PHRASES: raise AppError(code="render_prompt_card.fsc_invalid", message=f"invalid screen_zone {zone!r}") if depth not in DEPTH_PHRASES: raise AppError(code="render_prompt_card.fsc_invalid", message=f"invalid depth_plane {depth!r}") if ga not in _VALID_GESTURE_ACTION: raise AppError(code="render_prompt_card.fsc_invalid", message=f"invalid gesture_action {ga!r}") if ga != "none" and not (isinstance(gtl, str) and gtl.strip()): raise AppError(code="render_prompt_card.fsc_invalid", message="gesture_target_label is empty but gesture_action != 'none'") if ga == "none" and (gtl or "").strip(): raise AppError(code="render_prompt_card.fsc_invalid", message="gesture_target_label is non-empty but gesture_action == 'none'") # cross-check: visible_entities 는 SID list (C##/P## 모두 한 set). # target_id 형식 자체로 character/prop 구분되므로 set membership 만 검사. if tk == "character": if not _CHARACTER_ID_RE.match(tid): raise AppError( code="render_prompt_card.fsc_invalid", message=f"target_kind=character requires C## target_id (got {tid!r})", ) if tid not in visible_set: raise AppError( code="render_prompt_card.fsc_cross_check_failed", message=( f"target_id {tid!r} not in visible_entities ({sorted(visible_set)})" ), ) elif tk == "prop": if not _PROP_ID_RE.match(tid): raise AppError( code="render_prompt_card.fsc_invalid", message=f"target_kind=prop requires P## target_id (got {tid!r})", ) if tid not in visible_set: raise AppError( code="render_prompt_card.fsc_cross_check_failed", message=( f"target_id {tid!r} not in visible_entities ({sorted(visible_set)})" ), ) elif tk == "background": if tid != "": raise AppError( code="render_prompt_card.fsc_invalid", message=f"target_kind=background requires empty target_id (got {tid!r})", ) prepared_constraints.append({ "constraint_id": f"{_FSC_ID_PREFIX}{idx + 1:03d}", **c, }) return {"reason": reason, "constraints": prepared_constraints} def validate_echoes( card: Dict[str, Any], result: Dict[str, Any], ) -> List[Dict[str, Any]]: """scene_detail post-validation: per-variation echo set 검증. violation list 반환 (raise X). caller 가 retry 판단.""" rs = (card or {}).get("render_strategy") or {} contract = rs.get("frame_spatial_contract") if contract is None: injected_ids: set = set() else: injected_ids = {c["constraint_id"] for c in (contract.get("constraints") or [])} out: List[Dict[str, Any]] = [] for v in (result or {}).get("t2i_variations", []) or []: echoed = set(v.get("applied_frame_spatial_constraint_ids") or []) if echoed != injected_ids: out.append({ "variant_label": v.get("variant_label"), "injected_ids": sorted(injected_ids), "echoed_ids": sorted(echoed), "missing": sorted(injected_ids - echoed), "extra": sorted(echoed - injected_ids), }) return out def phrase_diagnostic( t2i_prompt: str, constraints: List[Dict[str, Any]], ) -> List[Dict[str, Any]]: """soft warning data 반환 (raise X). v1 soft, v2 hard 승격 가능.""" out: List[Dict[str, Any]] = [] prompt_lower = (t2i_prompt or "").lower() for c in constraints or []: label = (c.get("label") or "").lower() zone = c.get("screen_zone") depth = c.get("depth_plane") zone_variants = ZONE_PHRASES.get(zone, []) depth_variants = DEPTH_PHRASES.get(depth, []) label_present = bool(label) and label in prompt_lower zone_present = any(v in prompt_lower for v in zone_variants) depth_present = any(v in prompt_lower for v in depth_variants) if not (label_present and zone_present and depth_present): out.append({ "constraint_id": c.get("constraint_id"), "label_missing": not label_present, "zone_missing": not zone_present, "depth_missing": not depth_present, }) return out def _format_retry_hint(fsc_violations: List[Dict[str, Any]]) -> str: """retry hint formatter — shot_staging.py:_format_retry_hint orientation pattern mirror.""" lines = [ "", "", "[재시도 — 직전 응답의 frame_spatial_contract 가 다음 위반을", " 포함했습니다. 수정해서 재출력하세요:]", ] for v in fsc_violations: si = v.get("scene_index") shi = v.get("shot_index") idx = v.get("constraint_idx") rsn = v.get("reason") loc = f"S{si} Shot{shi}" if idx is not None: loc += f" constraint[{idx}]" # LLM retry context — enum violation 이면 invalid value / # target_id shape violation 이면 target_id / max_3_exceeded 이면 count # 노출 (find_frame_spatial_contract_violations payload 와 일관). suffix = "" if "value" in v: suffix = f" (got {v['value']!r})" elif "target_id" in v: suffix = f" (target_id={v['target_id']!r})" elif "count" in v: suffix = f" (count={v['count']})" lines.append(f" - {loc}: {rsn}{suffix}") return "\n".join(lines)