"""G4.6 Wave A3 RC-E+RC-H — visible_entities ⊃ C##/C##O## contract validator. ID coverage primary + dynamic entity_canon.name secondary (RO-2 강화): - Source 1 Primary (ID coverage): - Forward: visible_entities 안 character base C## 모두 t2i_prompt 에 C## 또는 C##O## 로 등장해야 함 (RC-E 핵심 — descriptor only 차단). - Reverse: t2i_prompt 안 사용된 ID 가 visible_entities 안에 있어야 함. - Source 2 Secondary (dynamic entity_canon.name diagnostic): - prebuild name map (main thread 에서 entity_canon project_id + entity_type='character' scoped query) 사용 — entity_canon.name token 이 t2i_prompt 에 등장하면 그 specific entity 의 ID candidates (base + 모든 allowed outlook composite) 가 same sentence + ±60 char window 안에 있어야 pass. - 다른 visible C## 가 window 안에 있어도 specific 매칭 안 되면 reject (RO-23 false-positive 차단). - Source 3a (Phase 4 fix iter 3): render_prompt_card.asset_requirements .required_refs[] kind-dispatched validation: - kind ∈ {character, character_outlook}: format check + visible_entities base membership. - kind = background: id format is chain_bg snake_case (entity regex 금지) — cross-field equality vs render_prompt_card.background_binding (mode == background_ref_attached + bg_id == id). reverse direction also enforced: ref_attached binding without a matching background entry raises (producer-drift fail-fast). - kind ∈ {prop, location}: id format only (P##/L##); membership against visible_entities is intentionally NOT enforced today (producer does not emit these kinds — see in-line comment in dispatch). - unknown kind: AppError fail-fast. - Source 3b: render_prompt_card.id_policy.allowed_outlook_pairs[]. character_id (legacy fallback: base_id / composite_id) → visible_entities. Codex iter 1 B1 carry — ThreadPool worker 안 SQLAlchemy session race 회피. caller 가 main thread 에서 prebuild 한 `name_by_short_id` map 을 인자로 전달. validator 안에서 db query 0. A-prime binding: - module 안에 nationality / ethnicity / 작품 고유명사 token (frozenset/tuple/ list/regex literal) 0건 — `feedback_no_scenario_keywords.md` (RC-C carry). - Source 1 의 ID regex `\\bC\\d{2,3}(?:O\\d{2,3})?\\b` 만 사용 (semantic keyword frozenset 금지와 충돌하지 않음 — universal ID structure parsing). - Source 2 의 entity_canon.name 은 prebuild map (project_id scoped) — 다른 nationality / 시나리오 자동 적용. PRO-4: required field 부재 / 타입 mismatch 시 fail-fast — silent `or []`/`or {}` 패턴 금지 (`feedback_no_silent_fallback.md`). optional field (description / shot_index 등 non-contract) 만 nullable 허용. """ import logging import re from typing import Any, Dict, List, Optional, Tuple from app.core.errors import AppError from app.core.subject_reference_policy import ( derive_visible_subject_ids, get_subject_reference_policy_or_default, normalize_subject_reference_policy_items, policy_to_id_usage_rule, ) logger = logging.getLogger(__name__) _MISSING = object() _ENTITY_ID_PATTERN = re.compile(r"\b(C\d{2,3}(?:O\d{2,3})?)\b") _OUTLOOK_ID_SHORT = re.compile(r"^O\d{2,3}$") _COMPOSITE_ID = re.compile(r"^C\d{2,3}O\d{2,3}$") _ANCHOR_WINDOW = 60 _SENTENCE_BOUNDARY_CHARS = ".!?\n。" # Phase 4 fix iter 3 — required_refs[].kind dispatch. # Phase 4 originally collapsed all kinds onto a single C##/L##/P## regex, # which mis-rejected legitimate chain_bg ids like "bg_rooftop_day_normal" # emitted by build_asset_requirements when background_mode_on + bg_id set. # kind=character / character_outlook / prop / location keep entity-shaped # patterns; kind=background validates against background_binding cross-field. _CHAR_REF_ID_PATTERN = re.compile(r"^C\d{2,3}$") _OUTLOOK_REF_ID_PATTERN = re.compile(r"^C\d{2,3}O\d{2,3}$") _PROP_REF_ID_PATTERN = re.compile(r"^P\d{2,3}$") _LOCATION_REF_ID_PATTERN = re.compile(r"^L\d{2,3}$") _BG_MODE_REF_ATTACHED = "background_ref_attached" def _require_field(d: dict, key: str, expected_type: type, shot_label: str): """PRO-4 — required field 부재 / 타입 mismatch fail-fast helper.""" value = d.get(key, _MISSING) if value is _MISSING: raise AppError( code="step.scene_detail.contract_violation_missing_field", message=( f"{shot_label} (PRO-4): required field '{key}' missing in shot. " f"Fix: scene_detail step output must include this field." ), status_code=400, ) if not isinstance(value, expected_type): raise AppError( code="step.scene_detail.contract_violation_field_type", message=( f"{shot_label} (PRO-4): field '{key}' must be " f"{expected_type.__name__}, got {type(value).__name__}." ), status_code=400, ) return value def _entity_specific_id_in_window( prompt: str, name_pos: int, name_len: int, entity_id_candidates: set, ) -> bool: """RO-23 + RO-27 — name_pos 기준 ±60 char window + sentence boundary + specific entity ID 매칭. window 안에 검사 중인 name 의 specific entity 의 ID (C## 또는 C##O##) 가 있어야만 True. 핵심 (RO-23 false-positive 차단): 다른 visible C## 가 window 안에 있어도, 검사 중인 name 의 specific entity 의 ID 가 아니면 reject. """ start = max(0, name_pos - _ANCHOR_WINDOW) end = min(len(prompt), name_pos + name_len + _ANCHOR_WINDOW) window_text = prompt[start:end] for m in _ENTITY_ID_PATTERN.finditer(window_text): sid = m.group(0) if sid not in entity_id_candidates: continue id_abs_pos = start + m.start() a, b = sorted([id_abs_pos, name_pos]) between = prompt[a:b] if any(ch in _SENTENCE_BOUNDARY_CHARS for ch in between): continue return True return False # ────────────────────────────────────────────────────────────────────── # Area #1 (2026-05-16) — body-part / face close-up substring matching 폐기. # 대체 SOT = id_policy.subject_reference_policy (per-subject array; # helper module: app.core.subject_reference_policy). # # 본 module 안 structured exemption 보존 (Area B/C 회귀 차단): # - render_strategy.mode == "partial_focus" → Area B/G4 forward exempt # - id_policy.reproduction_surface_rule.applies == True → Area C forward exempt # substring/noun-list 분기 모두 폐기 — Area #1 에서 helper SOT 로 대체. # ────────────────────────────────────────────────────────────────────── def _is_forward_exempt_by_render_strategy( rpc: Dict[str, Any], ) -> Tuple[bool, str]: """Area B/G4 보존 — render_strategy.mode == 'partial_focus' 면 forward 면제.""" rs = rpc.get("render_strategy") if isinstance(rs, dict): mode = rs.get("mode") if isinstance(mode, str) and mode == "partial_focus": return True, "render_strategy.mode == 'partial_focus'" return False, "" def _is_forward_exempt_by_reproduction_surface_rule( rpc: Dict[str, Any], ) -> Tuple[bool, str]: """Area C 보존 — id_policy.reproduction_surface_rule.applies == True 면 면제. Critical 2 (review fix-up 2026-05-12) — sentinel 기반 분기: 1) field absent → fall-through (legacy/stale card compat). 2) field present + dict + applies bool → 4-branch logic. 3) field present + non-dict → fail-fast (Gate 4 정합). """ id_policy = rpc.get("id_policy") if not isinstance(id_policy, dict): return False, "" repro = id_policy.get("reproduction_surface_rule", _MISSING) if repro is _MISSING: return False, "" if isinstance(repro, dict): applies = repro.get("applies", _MISSING) if applies is True: return True, "reproduction_surface_rule.applies == True" if applies is False or applies is _MISSING: return False, "" raise AppError( code="visible_entities_validator.reproduction_surface_rule_malformed", message=( f"reproduction_surface_rule.applies must be bool, " f"got {type(applies).__name__} ({applies!r})." ), ) raise AppError( code="visible_entities_validator.reproduction_surface_rule_malformed", message=( f"reproduction_surface_rule must be dict or absent, " f"got {type(repro).__name__} ({repro!r})." ), ) def _outlook_id_candidates_for_base( base_id: str, allowed_outlook_pairs: list, shot_label: str, ) -> set: """base C## + 그 base 의 모든 allowed C##O## composite ID set. Pair 에서 candidate 도출 우선순위: 1. `composite_id` (C##O## 형식) 있으면 직접 추가 (가장 신뢰). 2. 없으면 `character_id` (또는 `base_id`) + `outlook_id` 합성. 단 `outlook_id` 는 O## short id (e.g., "O11", "O123") 형태 강제. Why: production 데이터에서 outlook_id 가 short id (O##) 또는 UUID 둘 다 가능. composite_id 가 있으면 그쪽이 truth (UUID outlook 도 정확 식별). composite_id 가 없을 때만 outlook_id 가 O## short id 라는 가정 — UUID / 다른 형태면 fail-fast (silent absorb 금지). Raises: AppError: composite_id 가 C##O## 형식 위반 또는 (composite_id 부재 시) outlook_id 가 O## 형식 위반. """ candidates = {base_id} for pair in allowed_outlook_pairs: if not isinstance(pair, dict): continue # source 3b 의 PRO-4 가 fail-fast 잡음 char_id = ( pair.get("character_id") or pair.get("base_id") or (pair.get("composite_id") or "").split("O")[0] or "" ) if char_id != base_id: continue composite_id = pair.get("composite_id") if composite_id: if not _COMPOSITE_ID.match(composite_id): raise AppError( code="step.scene_detail.contract_violation_composite_id_format", message=( f"{shot_label}: allowed_outlook_pairs.composite_id " f"'{composite_id}' must match C##O## (2-3 digits each). " f"entry={pair!r}" ), status_code=400, ) candidates.add(composite_id) continue outlook_id = pair.get("outlook_id") if not outlook_id: continue # source 3b PRO-13 fail-fast 가 잡음 if not _OUTLOOK_ID_SHORT.match(outlook_id): raise AppError( code="step.scene_detail.contract_violation_outlook_id_format", message=( f"{shot_label}: allowed_outlook_pairs.outlook_id " f"'{outlook_id}' must match O## short id (O followed by " f"2-3 digits). composite_id 부재 시 outlook_id 만으로 " f"candidate 합성 — UUID / 다른 형태는 명시적으로 " f"composite_id 필드로 표현하세요. entry={pair!r}" ), status_code=400, ) candidates.add(f"{base_id}{outlook_id}") return candidates def validate_visible_entities_contract( shot: dict, name_by_short_id: dict, *, audit_warnings: Optional[List[Dict[str, Any]]] = None, ) -> None: """ID coverage primary + dynamic entity_canon.name secondary. silent bypass 없음 (RO-15 + PRO-4) — required field 모두 _MISSING fail-fast. Codex iter 1 B1 carry — db / project_id 인자 제거. caller (detail_steps `_post_process` 또는 ctx-aware test) 가 main thread 에서 prebuild 한 `name_by_short_id` map 을 전달. ThreadPool worker 안 SQLAlchemy session race 회피. Args: shot: scene_detail step 결과의 단일 shot dict (visible_entities, t2i_variations, render_prompt_card 포함). name_by_short_id: short_id (e.g., "C08") → entity_canon.name. ctx level 에서 prebuild — entity_type='character' 만 포함. """ scene_idx = shot.get("scene_index") shot_idx = shot.get("_shot_index") or shot.get("shot_index") shot_label = f"S{scene_idx}_Shot{shot_idx}" visible_list = _require_field(shot, "visible_entities", list, shot_label) visible = set(visible_list) visible_bases = derive_visible_subject_ids(visible_list) t2i_variations = _require_field(shot, "t2i_variations", list, shot_label) if not t2i_variations: raise AppError( code="step.scene_detail.contract_violation_empty_t2i", message=( f"{shot_label} (PRO-4): t2i_variations is empty. scene_detail " f"must produce at least one variation per shot." ), status_code=400, ) rpc = _require_field(shot, "render_prompt_card", dict, shot_label) asset_req = _require_field( {"asset_requirements": rpc.get("asset_requirements", _MISSING)}, "asset_requirements", dict, shot_label, ) id_policy = _require_field( {"id_policy": rpc.get("id_policy", _MISSING)}, "id_policy", dict, shot_label, ) allowed_outlook_pairs = id_policy.get("allowed_outlook_pairs", _MISSING) if allowed_outlook_pairs is _MISSING: raise AppError( code="step.scene_detail.contract_violation_missing_field", message=( f"{shot_label} (PRO-10): id_policy.allowed_outlook_pairs missing — " f"required field per RO-16 contract." ), status_code=400, ) if not isinstance(allowed_outlook_pairs, list): raise AppError( code="step.scene_detail.contract_violation_outlook_pairs_type", message=f"{shot_label} (PRO-4): id_policy.allowed_outlook_pairs must be list.", status_code=400, ) # Codex iter 1 I3 carry — allowed_outlook_pairs entry shape validation 을 # Source 2 helper 호출 전 선행. Source 3b 가 마지막에 잡으면 Source 2 가 # malformed pair 를 silent skip → "specific ID 없음" 같은 잘못된 에러로 # 바뀜. Shape 검증 (dict / character_id-or-base_id-or-composite_id / # outlook_id) 을 먼저 통과시킨 후 Source 2 진입. for pair in allowed_outlook_pairs: if not isinstance(pair, dict): raise AppError( code="step.scene_detail.contract_violation_outlook_pair_type", message=f"{shot_label} (PRO-4): allowed_outlook_pairs entry must be dict.", status_code=400, ) _shape_base = ( pair.get("character_id") or pair.get("base_id") or (pair.get("composite_id") or "").split("O")[0] or "" ) if not _shape_base: raise AppError( code="step.scene_detail.contract_violation_outlook_pair_no_id", message=( f"{shot_label} (Codex I3 + PRO-13): allowed_outlook_pairs " f"entry missing all of character_id / base_id / composite_id. " f"entry={pair!r}" ), status_code=400, ) _shape_outlook = pair.get("outlook_id", _MISSING) if _shape_outlook is _MISSING: raise AppError( code="step.scene_detail.contract_violation_outlook_pair_no_outlook_id", message=( f"{shot_label} (Codex I3 + PRO-13): allowed_outlook_pairs " f"entry missing 'outlook_id' field. entry={pair!r}" ), status_code=400, ) if not isinstance(_shape_outlook, str): raise AppError( code="step.scene_detail.contract_violation_outlook_pair_outlook_id_type", message=( f"{shot_label} (Codex I3 + PRO-13): outlook_id must be str, " f"got {type(_shape_outlook).__name__}. entry={pair!r}" ), status_code=400, ) # Source 2 prep — prebuild name map (Codex iter 1 B1). # caller 가 main thread 에서 entity_canon project_id + entity_type='character' # scoped query 한 결과를 인자로 전달. validator 는 db query 0. base_to_name: dict = { sid: name_by_short_id[sid] for sid in visible_bases if sid in name_by_short_id and name_by_short_id[sid] } for vidx, variation in enumerate(t2i_variations): if not isinstance(variation, dict): raise AppError( code="step.scene_detail.contract_violation_variation_type", message=f"{shot_label} (PRO-4): t2i_variations entry must be dict.", status_code=400, ) prompt = variation.get("t2i_prompt", _MISSING) if prompt is _MISSING or not isinstance(prompt, str): raise AppError( code="step.scene_detail.contract_violation_variation_prompt", message=( f"{shot_label} (PRO-4): t2i_variations[].t2i_prompt missing " f"or non-string." ), status_code=400, ) used_ids = set(_ENTITY_ID_PATTERN.findall(prompt)) used_bases = {sid.split("O")[0] for sid in used_ids} for sid in used_ids: base = sid.split("O")[0] if base not in visible_bases: raise AppError( code="step.scene_detail.contract_violation_id_not_visible", message=( f"{shot_label} (Source 1, RO-2): t2i_variations[{vidx}]." f"t2i_prompt uses '{sid}' (base={base}) but " f"visible_entities={sorted(visible)} does not include it." ), status_code=400, ) # Area #1 (2026-05-16) — per-subject policy loop + structured exemption. # body-part substring branch 폐기. structured exemption (Area B/G4 + # Area C) 보존. card.id_policy.subject_reference_policy field 필수 # (Gate 4 — silent fallback 차단). srp_array = id_policy.get("subject_reference_policy", _MISSING) if srp_array is _MISSING: raise AppError( code="step.contract_violation.id_policy.field_missing", message=( f"{shot_label} (Area #1): card.id_policy missing required " f"field 'subject_reference_policy'." ), status_code=400, ) policy_map = normalize_subject_reference_policy_items( srp_array, visible_subject_ids=visible_bases, where=f"validator:{shot_label}.variation[{vidx}]", ) # Used outlook detect: f"{base}O" prefix. used_outlook_for_base = { base for base in used_bases if any(sid.startswith(f"{base}O") for sid in used_ids) } missing_bases = visible_bases - used_bases if missing_bases: exempt_rs, reason_rs = _is_forward_exempt_by_render_strategy(rpc) exempt_rep, reason_rep = ( _is_forward_exempt_by_reproduction_surface_rule(rpc) ) if exempt_rs: logger.debug( "%s (Source 1 forward exempt, variation %d): %s " "— missing %s allowed.", shot_label, vidx, reason_rs, sorted(missing_bases), ) elif exempt_rep: logger.debug( "%s (Source 1 forward exempt, variation %d): %s " "— missing %s allowed.", shot_label, vidx, reason_rep, sorted(missing_bases), ) else: # Per-subject policy loop — base_required check. for base in sorted(missing_bases): policy = get_subject_reference_policy_or_default( policy_map, base, where=f"validator:{shot_label}.variation[{vidx}]", ) rule = policy_to_id_usage_rule(policy.policy) if rule.base_required: raise AppError( code=( "step.scene_detail.contract_violation." "subject_reference_policy.base_id_missing" ), message=( f"{shot_label} (variation {vidx}): subject " f"{base!r} policy={policy.policy!r} requires " f"base C## in t2i_prompt, but missing. " f"visible_entities={sorted(visible)}." ), status_code=400, ) # Per-subject outlook required + outlook forbidden 검사 (used_bases 기준). for base in sorted(used_bases): policy = get_subject_reference_policy_or_default( policy_map, base, where=f"validator:{shot_label}.variation[{vidx}]", ) rule = policy_to_id_usage_rule(policy.policy) has_outlook = base in used_outlook_for_base if rule.outlook_required and not has_outlook: raise AppError( code=( "step.scene_detail.contract_violation." "subject_reference_policy.outlook_id_missing" ), message=( f"{shot_label} (variation {vidx}): subject {base!r} " f"policy={policy.policy!r} requires outlook C##O## " f"but only base form present." ), status_code=400, ) if rule.outlook_forbidden and has_outlook: raise AppError( code=( "step.scene_detail.contract_violation." "subject_reference_policy.outlook_forbidden" ), message=( f"{shot_label} (variation {vidx}): subject {base!r} " f"policy={policy.policy!r} forbids outlook C##O## " f"but prompt uses outlook form." ), status_code=400, ) # Phase 3 W1 — base_forbidden: generic_descriptor_allowed subjects # must use NO id at all (no C##, no C##O##). outlook_forbidden check # above already fires for C##O## form; this branch catches bare C##. if rule.base_forbidden and not has_outlook: raise AppError( code=( "step.scene_detail.contract_violation." "subject_reference_policy.base_id_forbidden" ), message=( f"{shot_label} (variation {vidx}): subject {base!r} " f"policy={policy.policy!r} " f"forbids bare C## but prompt uses bare id form." ), status_code=400, ) # W20F2 — entity_canon.name ±60 char window check 는 자연어 substring # + character-window 기반 의미 판단 이라 SOT 로 부적합. structured SOT # (id_policy.subject_reference_policy + visible_entities + asset_req # required_refs) 가 진짜 enforcement 이고, 본 분기는 audit warning 만 # carry 한다 (Codex 2026-05-28 narrow wave directive). 옛 raise 경로는 # retry 후에도 LLM 이 동일 자유어를 다시 뱉어 stuck 시키는 원인이었다. for vidx, variation in enumerate(t2i_variations): prompt = variation["t2i_prompt"] prompt_lower = prompt.lower() for base_id, name in base_to_name.items(): if not name or len(name.strip()) < 2: continue entity_id_cands = _outlook_id_candidates_for_base( base_id, allowed_outlook_pairs, shot_label, ) name_lower = name.lower() offset = 0 while True: pos = prompt_lower.find(name_lower, offset) if pos == -1: break if not _entity_specific_id_in_window( prompt, pos, len(name_lower), entity_id_cands, ): warning_payload = { "code": ( "step.scene_detail.audit_warning_" "entity_name_no_specific_id" ), "shot_label": shot_label, "variation_index": vidx, "base_id": base_id, "entity_name": name, "entity_id_candidates": sorted(entity_id_cands), "match_pos": pos, "anchor_window": _ANCHOR_WINDOW, } if audit_warnings is not None: audit_warnings.append(warning_payload) else: import logging as _logging _logging.getLogger(__name__).warning( "validate_visible_entities_contract audit: %s " "entity_canon.name=%r base=%s — no specific ID " "(candidates=%r) within ±%d window (Source 2 " "downgraded to audit, see W20F2).", shot_label, name, base_id, sorted(entity_id_cands), _ANCHOR_WINDOW, ) offset = pos + len(name_lower) required_refs = asset_req.get("required_refs", _MISSING) if required_refs is _MISSING: raise AppError( code="step.scene_detail.contract_violation_missing_field", message=( f"{shot_label} (PRO-12): asset_requirements.required_refs missing — " f"required field. 빈 list 가능, missing 불가." ), status_code=400, ) if not isinstance(required_refs, list): raise AppError( code="step.scene_detail.contract_violation_required_refs_type", message=f"{shot_label} (PRO-12): asset_requirements.required_refs must be list.", status_code=400, ) # Source 3a — required_refs entry validation (Codex iter 1 I1 carry + # Phase 4 fix iter 3 kind-dispatch + iter 4 bidirectional cross-field). # # build_asset_requirements emits two ID shapes today: entity short_id for # character_outlook (C##O##) and chain_bg snake_case for background. The # original collapsed regex `^[CLP]\d{2,3}(?:O\d{2,3})?$` mis-rejected bg_* # ids as a false-positive. kind dispatch keeps each kind on its own # pattern, and kind=background defers ID validation to background_binding # cross-field equality (mode == background_ref_attached + id == bb.bg_id). # # Order discipline (Codex iter 4 I2): all entry-level schema errors raise # inside the loop FIRST; cross-field errors (binding mode/id, reverse # missing) raise AFTER the loop using the accumulated background_ref_ids. # This keeps malformed-entry diagnostics from mixing with binding errors. # # Producer-emitted kinds today: character_outlook + background. The # character / prop / location branches are defense-in-depth for future # producer additions and are exercised by authored tests. # # rid 가 truthy non-string 이면 .split() AttributeError → except Exception # 가 잡아 silent return None. PRO-4 fail-fast 정책 위반. 따라서 id/kind # type 도 명시 검증 + AppError keyword args. background_ref_ids: list[str] = [] for ref in required_refs: if not isinstance(ref, dict): raise AppError( code="step.scene_detail.contract_violation_required_ref_type", message=f"{shot_label} (PRO-4): required_refs entry must be dict.", status_code=400, ) # M2 (Codex iter 4): id type-then-truthiness ordering keeps the # diagnostic bucket clean — non-string falsy ids surface as id_type # rather than no_id. if "id" not in ref: raise AppError( code="step.scene_detail.contract_violation_required_ref_no_id", message=f"{shot_label} (PRO-4): required_refs entry missing 'id'.", status_code=400, ) rid = ref["id"] kind = ref.get("kind", "") if not isinstance(rid, str): raise AppError( code="step.scene_detail.contract_violation_required_ref_id_type", message=( f"{shot_label} (Codex I1): required_refs entry 'id' must be " f"str, got {type(rid).__name__}. entry={ref!r}" ), status_code=400, ) if not rid: raise AppError( code="step.scene_detail.contract_violation_required_ref_no_id", message=( f"{shot_label} (PRO-4): required_refs entry has empty 'id'. " f"entry={ref!r}" ), status_code=400, ) if not isinstance(kind, str): raise AppError( code="step.scene_detail.contract_violation_required_ref_kind_type", message=( f"{shot_label} (Codex I1): required_refs entry 'kind' must be " f"str, got {type(kind).__name__}. entry={ref!r}" ), status_code=400, ) if kind == "character": if not _CHAR_REF_ID_PATTERN.match(rid): raise AppError( code="step.scene_detail.contract_violation_required_ref_id_format", message=( f"{shot_label} (Phase 4 fix iter 3): kind='character' " f"id={rid!r} must match C## (2-3 digits). entry={ref!r}" ), status_code=400, ) if rid not in visible_bases: raise AppError( code="step.scene_detail.contract_violation_required_ref_not_visible", message=( f"{shot_label} (Source 3a, RO-2): " f"asset_requirements.required_refs has character id={rid} " f"but visible_entities={sorted(visible)} does not include it." ), status_code=400, ) elif kind == "character_outlook": if not _OUTLOOK_REF_ID_PATTERN.match(rid): raise AppError( code="step.scene_detail.contract_violation_required_ref_id_format", message=( f"{shot_label} (Phase 4 fix iter 3): kind='character_outlook' " f"id={rid!r} must match C##O## (2-3 digits each). " f"entry={ref!r}" ), status_code=400, ) base = rid.split("O")[0] if base not in visible_bases: raise AppError( code="step.scene_detail.contract_violation_required_ref_not_visible", message=( f"{shot_label} (Source 3a, RO-2): " f"asset_requirements.required_refs has character_outlook " f"id={rid} but visible_entities={sorted(visible)} does not " f"include base={base}." ), status_code=400, ) elif kind == "prop": # NOTE (Patch A 2026-05-11 update): build_asset_requirements 가 # 이제 story-critical prop 을 kind='prop' required_refs 로 emit # 한다. visible_entities 멤버십은 여전히 안 강제 — visible_entities # 는 character base 만 set (id_policy.allowed_outlook_pairs 검사용). # Prop binding 은 PRO-13 (아래 t2i_prompt 검사) + Tier 3 # validate_attached_refs 가 image stage 진입 직전 fail-fast. if not _PROP_REF_ID_PATTERN.match(rid): raise AppError( code="step.scene_detail.contract_violation_required_ref_id_format", message=( f"{shot_label} (Phase 4 fix iter 3): kind='prop' " f"id={rid!r} must match P## (2-3 digits). entry={ref!r}" ), status_code=400, ) elif kind == "location": # NOTE: location membership likewise unenforced — producer does # not emit kind='location'. Mirror the prop note when introduced. if not _LOCATION_REF_ID_PATTERN.match(rid): raise AppError( code="step.scene_detail.contract_violation_required_ref_id_format", message=( f"{shot_label} (Phase 4 fix iter 3): kind='location' " f"id={rid!r} must match L## (2-3 digits). entry={ref!r}" ), status_code=400, ) elif kind == "background": # M1 (Codex iter 4): policy="required" 강제 — producer emits # background entries only with policy='required'; anything else # is producer drift. policy = ref.get("policy") if policy != "required": raise AppError( code="step.scene_detail.contract_violation_required_ref_background_policy", message=( f"{shot_label} (Phase 4 fix iter 4 M1): kind='background' " f"required_refs entry must have policy='required', got " f"{policy!r}. entry={ref!r}" ), status_code=400, ) background_ref_ids.append(rid) else: raise AppError( code="step.scene_detail.contract_violation_required_ref_unknown_kind", message=( f"{shot_label} (Phase 4 fix iter 3): required_refs entry " f"kind={kind!r} not in supported kinds (character, " f"character_outlook, prop, location, background). " f"entry={ref!r}" ), status_code=400, ) # Patch A — PRO-13 (Tier 1): required_refs(kind='prop') 의 P## 가 적어도 # 한 t2i_variations[*].t2i_prompt 본문에 word-boundary 매치로 등장해야 한다. # attached_meta 검사 없음 (Tier 3 = validate_attached_refs 책임 분리). # 미등장 시 scene_detail step retry trigger. _patch_a_required_props = [ (entry.get("id") or "") for entry in required_refs if isinstance(entry, dict) and entry.get("kind") == "prop" and (entry.get("id") or "") ] if _patch_a_required_props: _patch_a_all_prompts = "\n".join( (v.get("t2i_prompt") or "") for v in t2i_variations if isinstance(v, dict) ) _patch_a_missing = [ pid for pid in _patch_a_required_props if not re.search(rf'\b{re.escape(pid)}\b', _patch_a_all_prompts) ] if _patch_a_missing: raise AppError( code="step.scene_detail.contract_violation_prop_p_id_missing_in_prompt", message=( f"{shot_label} (PRO-13): required prop P## not in any " f"t2i_variation prompt — missing={_patch_a_missing}. " f"Tier 1 fail-fast — scene_detail step retry." ), status_code=400, ) # forbidden_refs is intentionally not validated here — producer emits a # deterministic close-framing entry without an id (kind='background', # reason=...). Extend dispatch only when future producers add id-bearing # forbidden refs. # # B1 + I2 (Codex iter 4): top-level cross-field check. After all # entry-level schema errors have cleared, enforce both directions of the # required_refs ↔ background_binding contract: # forward — kind='background' entry present → binding must be REF_ATTACHED # with bg_id == ref.id; # reverse — binding REF_ATTACHED + bg_id non-empty → required_refs must # contain a matching kind='background' entry (silent hole if # producer drops the entry while keeping the binding). bb_top = rpc.get("background_binding", _MISSING) def _bb_dict_or_raise(label_kind: str) -> dict: if bb_top is _MISSING: raise AppError( code="step.scene_detail.contract_violation_missing_field", message=( f"{shot_label} (Phase 4 fix iter 3 — {label_kind}): " f"required_refs has a kind='background' entry but " f"render_prompt_card.background_binding is missing. The " f"card producer must emit background_binding alongside " f"any background required_ref so cross-field equality " f"can be checked." ), status_code=400, ) if not isinstance(bb_top, dict): raise AppError( code="step.scene_detail.contract_violation_field_type", message=( f"{shot_label} (PRO-4 — {label_kind}): " f"render_prompt_card.background_binding must be dict, got " f"{type(bb_top).__name__}." ), status_code=400, ) return bb_top if background_ref_ids: bb = _bb_dict_or_raise("forward") bb_mode = bb.get("mode") bb_bg_id = bb.get("bg_id") if bb_mode != _BG_MODE_REF_ATTACHED: raise AppError( code="step.scene_detail.contract_violation_required_ref_background_mode_mismatch", message=( f"{shot_label} (Phase 4 fix iter 3): asset_requirements." f"required_refs has kind='background' entries " f"{background_ref_ids} but background_binding.mode=" f"{bb_mode!r} ≠ {_BG_MODE_REF_ATTACHED!r}. background ref " f"is only valid when the binding is in ref-attached mode." ), status_code=400, ) # M3 (Codex iter 4): producer-missing bg_id is a distinct diagnostic # bucket from actual id-mismatch. if not (isinstance(bb_bg_id, str) and bb_bg_id): raise AppError( code="step.scene_detail.contract_violation_background_binding_bg_id_invalid", message=( f"{shot_label} (Phase 4 fix iter 4 M3): " f"background_binding.mode={_BG_MODE_REF_ATTACHED!r} but " f"bg_id={bb_bg_id!r} is missing / empty / non-string. " f"Producer must populate bg_id whenever required_refs " f"contains a kind='background' entry." ), status_code=400, ) for rid in background_ref_ids: if rid != bb_bg_id: raise AppError( code="step.scene_detail.contract_violation_required_ref_background_id_mismatch", message=( f"{shot_label} (Phase 4 fix iter 3): asset_requirements." f"required_refs has kind='background' id={rid!r} but " f"background_binding.bg_id={bb_bg_id!r}. The two fields " f"must be identical (chain_bg id is the single source " f"of truth)." ), status_code=400, ) elif ( isinstance(bb_top, dict) and bb_top.get("mode") == _BG_MODE_REF_ATTACHED ): # B1 (Codex iter 4 BLOCKING): reverse direction — binding claims a # ref is attached but required_refs does not have the matching entry. bb_bg_id = bb_top.get("bg_id") if not (isinstance(bb_bg_id, str) and bb_bg_id): raise AppError( code="step.scene_detail.contract_violation_background_binding_bg_id_invalid", message=( f"{shot_label} (Phase 4 fix iter 4 M3 — reverse): " f"background_binding.mode={_BG_MODE_REF_ATTACHED!r} but " f"bg_id={bb_bg_id!r} is missing / empty / non-string." ), status_code=400, ) raise AppError( code="step.scene_detail.contract_violation_required_ref_background_missing", message=( f"{shot_label} (Phase 4 fix iter 4 B1): " f"background_binding.mode={_BG_MODE_REF_ATTACHED!r} with " f"bg_id={bb_bg_id!r} but asset_requirements.required_refs " f"contains no kind='background' entry. Producer drift — when " f"the binding is ref-attached, the matching background ref " f"must be present in required_refs (one source of truth)." ), status_code=400, ) for pair in allowed_outlook_pairs: if not isinstance(pair, dict): raise AppError( code="step.scene_detail.contract_violation_outlook_pair_type", message=f"{shot_label} (PRO-4): allowed_outlook_pairs entry must be dict.", status_code=400, ) base = ( pair.get("character_id") or pair.get("base_id") or (pair.get("composite_id") or "").split("O")[0] or "" ) if not base: raise AppError( code="step.scene_detail.contract_violation_outlook_pair_no_id", message=( f"{shot_label} (PRO-13): allowed_outlook_pairs entry missing " f"all of character_id / base_id / composite_id. entry={pair!r}" ), status_code=400, ) outlook_id = pair.get("outlook_id", _MISSING) if outlook_id is _MISSING: raise AppError( code="step.scene_detail.contract_violation_outlook_pair_no_outlook_id", message=( f"{shot_label} (PRO-13): allowed_outlook_pairs entry missing " f"'outlook_id' field. entry={pair!r}" ), status_code=400, ) if not isinstance(outlook_id, str): raise AppError( code="step.scene_detail.contract_violation_outlook_pair_outlook_id_type", message=( f"{shot_label} (PRO-13): outlook_id must be str, " f"got {type(outlook_id).__name__}." ), status_code=400, ) if base not in visible_bases: raise AppError( code="step.scene_detail.contract_violation_outlook_pair_not_visible", message=( f"{shot_label} (Source 3b, RO-16): " f"id_policy.allowed_outlook_pairs has character_id={base} " f"but visible_entities={sorted(visible)} does not include it." ), status_code=400, )