"""G3.2 owned validation helpers — single source. producer-side normalize + consumer-side hash/sentinel build + shape validator + fail-fast guard (옛 v4 / partial v5 cp 차단). post-parse 가 cp 에 삽입하는 sentinel: {schema_version, t2i_prompt_hash, owned_hash, camera_direction_hash, validator, violations} LLM 응답 schema 에는 들어가지 않음 (CP-only). """ from __future__ import annotations import hashlib import json import logging import re from typing import Any, Dict, Iterable, List, Optional from app.core.errors import AppError logger = logging.getLogger(__name__) # Wave 6 BLOCKING fix: hash field 는 sha256[:16] → 16자리 lowercase hex. _HASH_HEX_RE = re.compile(r"[0-9a-f]{16}") OWNED_SENTINEL_SCHEMA_VERSION = 2 # C2 v1: owned_usage_hash field 추가 (sentinel v2). OWNED_VALIDATOR_FULL = "scene_detail_owned_objects.v1" OWNED_VALIDATOR_CLOSE_SKIP = "scene_detail_owned_objects.v1.close_skip" OWNED_MAX_LEN = 80 # background_prompt schema items maxLength 와 일치 _OWNED_USAGE_KINDS = ("redraw", "anchor", "absent") def normalize_owned_list(items: Iterable[Any]) -> List[str]: """strip / dedupe (case-sensitive) / drop empty whitespace / 정렬. 저장 직전에 호출 — 단순 검증이 아니라 정규화 책임. case 보존 이유: 영어 canonical 안에서도 "TV" vs "tv" 같은 분리가 의도일 수 있음. Wave 6 IMPORTANT fix (feedback_no_silent_fallback.md): - 비-string entry → AppError(step.contract_violation) raise (silent drop 차단 — upstream LLM 이 schema 위반 가능). - overlength entry (>OWNED_MAX_LEN) → AppError raise (silent truncation 으로 의미 잘림 차단). - non-ASCII (Hangul / CJK / Kana 등) → AppError raise (영어 canonical 위반 surface — 옛 silent skip 패턴 폐기). - 진정한 빈/whitespace-only entry 만 silent drop (정규화 의도). """ seen: set = set() out: List[str] = [] for it in items or []: # Wave 6 fix: 비-string fail-fast (silent drop 차단). if not isinstance(it, str): raise AppError( code="step.contract_violation", message=( f"normalize_owned_list got non-string item {it!r} " f"(type={type(it).__name__})" ), ) s = it.strip() if not s: # 빈 / whitespace-only 만 silent drop (정규화 의도). continue # Wave 6 fix: ASCII 위반 fail-fast (영어 canonical 강제, 옛 skip 폐기). try: s.encode("ascii") except UnicodeEncodeError: raise AppError( code="step.contract_violation", message=( f"normalize_owned_list got non-ASCII item {s!r} — " "owned MUST be English canonical common nouns " "(Hangul / CJK / Kana 등 차단)." ), ) # Wave 6 fix: overlength fail-fast (silent truncation 차단). if len(s) > OWNED_MAX_LEN: raise AppError( code="step.contract_violation", message=( f"normalize_owned_list got overlength item " f"({len(s)} > {OWNED_MAX_LEN}): {s!r}" ), ) if s in seen: continue seen.add(s) out.append(s) out.sort() return out def compute_owned_hash(owned: List[str]) -> str: """sorted+joined sha256[:16]. caller 가 normalize 통과한 list 를 넘긴다고 가정. 빈 list 도 결정론적 hash — sentinel hash 가 항상 존재 보장. Wave 6 IMPORTANT fix: defense-in-depth — caller order 와 무관하게 내부에서 sorted 적용. docstring 의 "sorted+joined" 약속을 구현이 보장. """ payload = "|".join(sorted(owned)).encode("utf-8") return hashlib.sha256(payload).hexdigest()[:16] def compute_camera_direction_hash(camera_direction: str) -> str: """camera_direction 텍스트 sha256[:16]. close framing 판정의 input 이라 text 변경 시 sentinel drift 감지에 사용. """ payload = (camera_direction or "").encode("utf-8") return hashlib.sha256(payload).hexdigest()[:16] def compute_t2i_prompt_hash(t2i_prompt: str) -> str: """variation['t2i_prompt'] or "" exact string 의 sha256[:16] (round 5 BLOCKING 3). t2i_review 가 t2i_prompt 를 수정 후 sentinel 미갱신 하면 verify_completion 의 drift 검증이 stale 로 partial 마킹. case / whitespace / punctuation 변화 모두 다른 hash → drift 검출 보장. """ payload = (t2i_prompt or "").encode("utf-8") return hashlib.sha256(payload).hexdigest()[:16] def refresh_t2i_prompt_hash( sentinel: Dict[str, Any], t2i_prompt: str, where: str = "" ) -> bool: """sentinel 의 t2i_prompt_hash 만 갱신. 다른 hash 필드 보존. sentinel shape 검증 후 갱신 — shape 위반 시 raise. Returns: True (갱신 발생) / False (이미 일치, no-op). AC-A1 (spec §3.4): t2i_review 가 mutation 후 의무 호출. owned_hash / camera_direction_hash / validator / violations 는 보존. """ assert_owned_sentinel_shape(sentinel, where=where) new_hash = compute_t2i_prompt_hash(t2i_prompt) if sentinel["t2i_prompt_hash"] == new_hash: return False sentinel["t2i_prompt_hash"] = new_hash return True _REQUIRED_SENTINEL_FIELDS = ( "schema_version", "t2i_prompt_hash", "owned_hash", "camera_direction_hash", "validator", "violations", "owned_usage_hash", ) _VALID_VALIDATORS = (OWNED_VALIDATOR_FULL, OWNED_VALIDATOR_CLOSE_SKIP) def validate_owned_object_usage_coverage( owned_object_usage: Any, normalized_owned_list: List[str], where: str = "", ) -> None: """C2 v1 §4.1 — owned_object_usage[] 가 normalized owned list 를 정확히 1:1 echo 검증. 위반 시 AppError(step.contract_violation). silent miss 차단. 검증: - owned_object_usage 는 list. - 각 entry 는 dict + {owned_token, usage_kind, source_phrase}. - usage_kind ∈ {redraw, anchor, absent}. - source_phrase 는 usage_kind == absent 일 때만 빈 문자열 허용. - exact cardinality: len(owned_object_usage) == len(normalized_owned_list). - owned_token unique (duplicate 금지). - set(owned_token) == set(normalized_owned_list) (no extra / no missing). - empty owned list → owned_object_usage 는 [] (required field 존재). """ if not isinstance(owned_object_usage, list): raise AppError( code="step.contract_violation", message=( f"owned_object_usage must be list " f"(got {type(owned_object_usage).__name__}) {where}" ), ) tokens: List[str] = [] for idx, entry in enumerate(owned_object_usage): if not isinstance(entry, dict): raise AppError( code="step.contract_violation", message=( f"owned_object_usage[{idx}] must be dict " f"(got {type(entry).__name__}) {where}" ), ) for k in ("owned_token", "usage_kind", "source_phrase"): if k not in entry or not isinstance(entry[k], str): raise AppError( code="step.contract_violation", message=( f"owned_object_usage[{idx}].{k} must be str " f"(got {entry.get(k)!r}) {where}" ), ) if entry["usage_kind"] not in _OWNED_USAGE_KINDS: raise AppError( code="step.contract_violation", message=( f"owned_object_usage[{idx}].usage_kind {entry['usage_kind']!r} " f"not in {_OWNED_USAGE_KINDS} {where}" ), ) if entry["usage_kind"] != "absent" and not entry["source_phrase"]: raise AppError( code="step.contract_violation", message=( f"owned_object_usage[{idx}].source_phrase empty but " f"usage_kind={entry['usage_kind']!r} (empty allowed only for absent) {where}" ), ) tokens.append(entry["owned_token"]) # exact cardinality + uniqueness + set equality. if len(tokens) != len(normalized_owned_list): raise AppError( code="step.contract_violation", message=( f"owned_object_usage cardinality {len(tokens)} != " f"normalized owned_list {len(normalized_owned_list)} {where}" ), ) if len(set(tokens)) != len(tokens): raise AppError( code="step.contract_violation", message=f"owned_object_usage owned_token has duplicates: {tokens} {where}", ) if set(tokens) != set(normalized_owned_list): raise AppError( code="step.contract_violation", message=( f"owned_object_usage owned_token set {sorted(set(tokens))} != " f"normalized owned_list {sorted(set(normalized_owned_list))} " f"(no extra / no missing) {where}" ), ) def merge_owned_object_usage( llm_owned_object_usage: Any, normalized_owned_list: List[str], *, is_close_framing: bool, where: str = "", diagnostics: Optional[Dict[str, Any]] = None, ) -> List[Dict[str, str]]: """FINDING 5 (e2e-bughunt-v1 W3) — deterministic owned_object_usage skeleton/merge. code 가 owned_object_usage 의 cardinality skeleton 을 소유한다 — `normalized_owned_list` 의 모든 owned token 마다 정확히 1 개 entry. LLM 은 실제로 redraw / anchor 한 owned 객체만 부분 declare 하고, code 가 stable owned-token identity 로 skeleton 에 merge 한다. LLM 이 빠뜨린 token → deterministic default (`usage_kind="absent"` + empty `source_phrase`). 이 helper 는 옛 1:1 exact-cardinality echo contract 를 대체한다. 옛 contract 는 LLM 이 12~23 개 owned token 을 모두 echo 하도록 강제했고, gemini-pro 가 이를 체계적으로 무시 (`owned_object_usage=[]`) 해 scene_detail 29/30 shot 이 실패했다. coverage cardinality 는 이제 code 가 보장 — LLM 누락은 더이상 위반이 아니다. close-framing (`is_close_framing=True`): close-framing prompt 은 의도적으로 owned block 을 LLM 입력에서 제외하므로 (`_build_phase2_prepend_blocks` 가 skip), LLM echo 는 무의미하다. code 가 모든 owned token 을 all-absent 로 synthesize 하고 LLM 입력은 완전히 무시한다. LLM 이 close-framing 에서 non-empty echo 를 내도 deterministic all-absent 가 우선 — shot 을 fail 시키지 않는다 (debug log 만). non-close: 각 LLM entry 를 fail-fast validate 후 owned_token 으로 index, skeleton 을 `normalized_owned_list` 순서로 build (LLM entry 있으면 그것, 없으면 absent default). fail-fast (`AppError` step.contract_violation) — malformed shape/type 만 차단한다 (silent 수용 금지): - `llm_owned_object_usage` 가 list 아님. - entry 가 dict 아님. - `owned_token` / `usage_kind` / `source_phrase` 누락 또는 non-str. - `usage_kind` 가 enum {redraw, anchor, absent} 밖. - 같은 `owned_token` 중복 declare. - `usage_kind != "absent"` 인데 `source_phrase` 빈 문자열 (모순 — 근거 없음). - `usage_kind == "absent"` 인데 `source_phrase` 비어있지 않음 (모순). W20F3 (Codex 2026-05-28) — `owned_token` 이 `normalized_owned_list` 밖 (unknown label) 은 더이상 hard fail 이 아니다. 해당 entry 를 drop 하고 `diagnostics["rejected_owned_tokens"]` 에 push (없으면 logger.warning). 그 자리는 skeleton 의 absent default 로 채워진다. stable owned-id 구조화는 별도 wave. LLM 누락 token 도 invalid 가 아니다 — skeleton 에서 absent default 로 채운다. Returns: merged owned_object_usage — `len == len(normalized_owned_list)`, owned_token set 은 `normalized_owned_list` 와 정확히 일치, 출력 순서도 `normalized_owned_list` 순서 (test 안정성). 직후 `build_owned_sentinel` 의 `validate_owned_object_usage_coverage` 가 통과하도록 보장한다 (coverage 함수 자체는 미약화 — 병합된 결과를 검증). caller 는 `normalized_owned_list` 가 `normalize_owned_list` 통과 list (정렬·dedupe·ASCII) 라고 가정한다 — `chain_bg_owned_by_shot` loader 가 보장. """ if is_close_framing: if isinstance(llm_owned_object_usage, list) and llm_owned_object_usage: logger.debug( "merge_owned_object_usage: close-framing path ignoring %d " "LLM owned_object_usage entries — deterministic all-absent " "synthesized %s", len(llm_owned_object_usage), where, ) return [ {"owned_token": t, "usage_kind": "absent", "source_phrase": ""} for t in normalized_owned_list ] if not isinstance(llm_owned_object_usage, list): raise AppError( code="step.contract_violation", message=( f"owned_object_usage must be list " f"(got {type(llm_owned_object_usage).__name__}) {where}" ), ) owned_set = set(normalized_owned_list) llm_by_token: Dict[str, Dict[str, str]] = {} for idx, entry in enumerate(llm_owned_object_usage): if not isinstance(entry, dict): raise AppError( code="step.contract_violation", message=( f"owned_object_usage[{idx}] must be dict " f"(got {type(entry).__name__}) {where}" ), ) for k in ("owned_token", "usage_kind", "source_phrase"): if k not in entry or not isinstance(entry[k], str): raise AppError( code="step.contract_violation", message=( f"owned_object_usage[{idx}].{k} must be str " f"(got {entry.get(k)!r}) {where}" ), ) token = entry["owned_token"] usage_kind = entry["usage_kind"] source_phrase = entry["source_phrase"] if usage_kind not in _OWNED_USAGE_KINDS: raise AppError( code="step.contract_violation", message=( f"owned_object_usage[{idx}].usage_kind {usage_kind!r} " f"not in {_OWNED_USAGE_KINDS} {where}" ), ) if token not in owned_set: # W20F3 — unknown owned_token 은 hard fail 대신 drop + diagnostic. # LLM 이 owned_list 밖 자유어 token 을 만들었을 때 step 전체를 막지 # 않고 해당 entry 만 skip 하고 normalized_owned_list 의 나머지는 # 정상 merge / 누락분은 absent skeleton 으로 채운다 (Codex # 2026-05-28 narrow wave directive). 누락 surfaces 는 diagnostics # 가 주어졌으면 거기에, 없으면 logger.warning 으로. if diagnostics is not None: diagnostics.setdefault("rejected_owned_tokens", []).append({ "index": idx, "owned_token": token, "usage_kind": usage_kind, "source_phrase": source_phrase, "where": where, }) else: logger.warning( "merge_owned_object_usage: drop unknown owned_token " "[%d]=%r (usage_kind=%r) %s", idx, token, usage_kind, where, ) continue if token in llm_by_token: raise AppError( code="step.contract_violation", message=( f"owned_object_usage[{idx}].owned_token {token!r} declared " f"more than once {where}" ), ) if usage_kind != "absent" and not source_phrase: raise AppError( code="step.contract_violation", message=( f"owned_object_usage[{idx}].source_phrase empty but " f"usage_kind={usage_kind!r} (non-absent needs evidence) {where}" ), ) if usage_kind == "absent" and source_phrase: raise AppError( code="step.contract_violation", message=( f"owned_object_usage[{idx}].source_phrase {source_phrase!r} " f"non-empty but usage_kind='absent' (contradiction) {where}" ), ) llm_by_token[token] = { "owned_token": token, "usage_kind": usage_kind, "source_phrase": source_phrase, } return [ llm_by_token.get( t, {"owned_token": t, "usage_kind": "absent", "source_phrase": ""} ) for t in normalized_owned_list ] def compute_owned_usage_hash(owned_object_usage: List[Dict[str, str]]) -> str: """C2 v1 §3.2 — owned_object_usage[] canonical sha256[:16]. Canonicalization: 1. sorted by owned_token (deterministic order). 2. payload per entry = {owned_token, usage_kind, source_phrase} exact (source_phrase = exact post-schema string, no strip / no normalization). 3. canonical JSON (sort_keys, no whitespace variance). caller 가 validate_owned_object_usage_coverage 를 먼저 호출했다고 가정 (build_owned_sentinel). """ ordered = sorted( ( { "owned_token": e["owned_token"], "usage_kind": e["usage_kind"], "source_phrase": e["source_phrase"], } for e in owned_object_usage ), key=lambda e: e["owned_token"], ) payload = json.dumps(ordered, sort_keys=True, separators=(",", ":"), ensure_ascii=False) return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16] def build_owned_sentinel( *, owned: List[str], camera_direction: str, t2i_prompt: str, is_close_framing: bool, violations: List[Dict[str, str]], owned_object_usage: List[Dict[str, str]], ) -> Dict[str, Any]: """post-parse 가 cp 에 삽입할 sentinel 생성 (C2 v1 sentinel v2). close framing 도 모든 hash 포함 (round 3 #4 + round 5 BLOCKING 3). drift 잡기 위함: - t2i_prompt 변경 → t2i_prompt_hash drift (t2i_review 수정 검출). - owned 변경 → owned_hash drift. - camera_direction 변경 → camera_direction_hash drift. - owned_object_usage 변경 → owned_usage_hash drift (C2 v1 — tampered usage detect). C2 v1 §3.2: owned_object_usage coverage 를 먼저 validate (validate_owned_object_usage_coverage against `owned`) 한 뒤 owned_usage_hash 계산. coverage 위반 시 AppError. """ validate_owned_object_usage_coverage( owned_object_usage, owned, where="build_owned_sentinel", ) return { "schema_version": OWNED_SENTINEL_SCHEMA_VERSION, "t2i_prompt_hash": compute_t2i_prompt_hash(t2i_prompt), "owned_hash": compute_owned_hash(owned), "camera_direction_hash": compute_camera_direction_hash(camera_direction), "owned_usage_hash": compute_owned_usage_hash(owned_object_usage), "validator": ( OWNED_VALIDATOR_CLOSE_SKIP if is_close_framing else OWNED_VALIDATOR_FULL ), "violations": list(violations or []), } def assert_owned_usage_hash_matches( sentinel: Dict[str, Any], owned_object_usage: List[Dict[str, str]], where: str = "", ) -> None: """C2 v1 N-1 — sentinel owned_usage_hash vs current owned_object_usage hash 비교. mismatch → AppError(step.contract_violation) — tampered usage array detect. sentinel shape 검증 후 비교 (shape 위반 시 raise). """ assert_owned_sentinel_shape(sentinel, where=where) current_hash = compute_owned_usage_hash(owned_object_usage) if sentinel["owned_usage_hash"] != current_hash: raise AppError( code="step.contract_violation", message=( f"owned_validation.owned_usage_hash mismatch — sentinel " f"{sentinel['owned_usage_hash']!r} != current {current_hash!r} " f"(tampered owned_object_usage array) {where}" ), ) def assert_close_framing_absent_echo( owned_object_usage: List[Dict[str, str]], where: str = "", ) -> None: """C2 v1 §4.1 — close-framing path 의 owned_object_usage invariant. close framing 은 owned 환경객체 redraw 불가 → 모든 entry 가 usage_kind="absent" + source_phrase="" 여야. 위반 시 AppError. caller 는 build_owned_sentinel / validate_owned_object_usage_coverage 로 entry shape 를 먼저 검증했다고 가정 (owned_token/usage_kind/source_phrase str). """ non_absent = [ e["owned_token"] for e in owned_object_usage if e["usage_kind"] != "absent" ] non_empty_phrase = [ e["owned_token"] for e in owned_object_usage if e["usage_kind"] == "absent" and e["source_phrase"] != "" ] if non_absent or non_empty_phrase: raise AppError( code="step.contract_violation", message=( f"close-framing owned_object_usage must be all usage_kind=" f"'absent' with empty source_phrase (non-absent: {non_absent}, " f"non-empty source_phrase: {non_empty_phrase}) {where}" ), ) def assert_owned_sentinel_shape(sentinel: Dict[str, Any], where: str = "") -> None: """sentinel field/type/enum 검증. 위반 시 AppError(step.contract_violation). post-parse 직후 + verify_completion 둘 다에서 호출 — silent fallback 차단. Wave 6 BLOCKING fix (Codex partB): 추가 검증 항목. - schema_version: int + 현재 OWNED_SENTINEL_SCHEMA_VERSION 일치. - hash 3 종 (t2i_prompt_hash / owned_hash / camera_direction_hash): 16자리 lowercase hex (sha256[:16]) 형식. - violations 항목: dict + owned_object/violating_phrase/reason str. """ if not isinstance(sentinel, dict): raise AppError( code="step.contract_violation", message=f"owned_validation must be dict (got {type(sentinel).__name__}) {where}", ) missing = [f for f in _REQUIRED_SENTINEL_FIELDS if f not in sentinel] if missing: raise AppError( code="step.contract_violation", message=f"owned_validation missing fields {missing} {where}", ) if sentinel["validator"] not in _VALID_VALIDATORS: raise AppError( code="step.contract_violation", message=( f"owned_validation.validator {sentinel['validator']!r} " f"not in {_VALID_VALIDATORS} {where}" ), ) if not isinstance(sentinel["violations"], list): raise AppError( code="step.contract_violation", message=( f"owned_validation.violations must be list " f"(got {type(sentinel['violations']).__name__}) {where}" ), ) # Wave 6 BLOCKING fix: schema_version 정수 + 현재 버전 일치. # bool 도 int 서브클래스라 isinstance(True, int)==True → 별도 reject # (sv=True 가 OWNED_SENTINEL_SCHEMA_VERSION==1 과 silent pass 차단). sv = sentinel["schema_version"] if isinstance(sv, bool) or not isinstance(sv, int) or sv != OWNED_SENTINEL_SCHEMA_VERSION: raise AppError( code="step.contract_violation", message=( f"owned_validation.schema_version must be int " f"={OWNED_SENTINEL_SCHEMA_VERSION} (got {sv!r}) {where}" ), ) # Wave 6 BLOCKING fix + C2 v1: hash 4 종 모두 16-char lowercase hex # (owned_usage_hash 는 sentinel v2 신규). for hk in ("t2i_prompt_hash", "owned_hash", "camera_direction_hash", "owned_usage_hash"): hv = sentinel[hk] if not isinstance(hv, str) or not _HASH_HEX_RE.fullmatch(hv): raise AppError( code="step.contract_violation", message=( f"owned_validation.{hk} must be 16-char lowercase hex " f"(got {hv!r}) {where}" ), ) # Wave 6 BLOCKING fix: violations 항목 dict + 3 키 str. # owned-judge prompt v2 (2.202605051641) carry: optional `verdict` 필드. # v2 schema 는 verdict enum {"redraw_violation","anchor_reference"} required — # 새로 생성되는 sentinel 은 항상 verdict present. optional 통과 path 는 오직 # 옛 v1 cp resume 시점 (pre-fix DB 잔존) — 한번 verify 통과 후 다음 force run # 에서 v2 schema 로 재기록. helper 도달 path: LLM 응답 (1차 _validate_local_schema # 가 reject) / 수동 cp 편집 / test fixture / future migration. for idx, v in enumerate(sentinel["violations"]): if not isinstance(v, dict): raise AppError( code="step.contract_violation", message=( f"owned_validation.violations[{idx}] must be dict " f"(got {type(v).__name__}) {where}" ), ) for vk in ("owned_object", "violating_phrase", "reason"): vv = v.get(vk) if not isinstance(vv, str): raise AppError( code="step.contract_violation", message=( f"owned_validation.violations[{idx}].{vk} must be str " f"(got {type(vv).__name__}) {where}" ), ) # v2 prompt carry — verdict 가 present 면 enum 검증. # judge v5 (2026-07-02): narrow exception 2종 추가 — # allowed_visual_state_change (shot-intent 가 명시 요구하는 owned 객체 # 시각 상태/내용 변형) / allowed_prop_contact (쥔/사용 중 객체의 물리적 # 상호작용 기하 서술). 둘 다 has_redraw_violation 카운트 대상 아님. if "verdict" in v: verdict = v["verdict"] if verdict not in ( "redraw_violation", "anchor_reference", "allowed_visual_state_change", "allowed_prop_contact", ): raise AppError( code="step.contract_violation", message=( f"owned_validation.violations[{idx}].verdict must be " f"one of 'redraw_violation'/'anchor_reference'/" f"'allowed_visual_state_change'/'allowed_prop_contact' " f"(got {verdict!r}) {where}" ), ) def has_redraw_violation(violations: List[Dict[str, Any]]) -> bool: """violations array 가 진짜 redraw violation 을 포함하는지 판정. Schema 자동 감지 (owned-judge prompt v1 vs v2): - v2 schema (모든 entry 에 verdict present): `verdict == "redraw_violation"` 만 카운트. anchor_reference 는 LLM 이 "발견했지만 위반 아님" 으로 분류한 것 — 통과. - v1 legacy schema (모든 entry 에 verdict 없음): 옛 동작 — 비어있지 않으면 violation. - mixed schema (일부만 verdict): **strict raise** (review I1) — production 도달 불가 (LiteLLM 1차 schema 검증이 차단). 만약 도달했다면 LLM bug 또는 cp 손상 → silent fallback 대신 fail-fast 로 debug 명확성 보장. cascade fix 2026-05-05: v1 prompt 는 LLM 이 anchor reference 도 violations 배열에 포함시키고 reason 으로 self-clearing 하는 false positive 패턴 발생. v2 prompt 는 verdict enum 으로 강제 분류 → 진짜 redraw 만 contract_violation marking. """ if not violations: return False verdict_count = sum( 1 for v in violations if isinstance(v, dict) and "verdict" in v ) if verdict_count == 0: # legacy v1: 빈 배열 아니면 violation (옛 동작 유지). return True if verdict_count == len(violations): # v2: 모든 entry 분류됨 → redraw_violation 만 카운트. return any(v.get("verdict") == "redraw_violation" for v in violations) # mixed schema — production 도달 불가, 도달 시 LLM/cp 손상. raise AppError( code="step.contract_violation", message=( f"owned_validation.violations mixed verdict schema: " f"{verdict_count}/{len(violations)} entries have verdict — LLM " "schema bug 또는 cp 손상. silent fallback 차단 (cascade fix I1)." ), ) def assert_background_prompt_owned_contract( bp_cp: Dict[str, Any] | None, *, background_mode_on: bool, where: str = "", ) -> None: """Spec 8.7 critical gap (round 4 BLOCKING 1 강화 + round 7 BLOCKING 1) — 옛 v4 / cp 부재 / partial v5 cp / 수동 편집 silent drop fail-fast. 허용 path: - bg off → cp None / 옛 cp 모두 통과 (caller 가 빈 dict 리턴). - bg on + cp schema=2 + 모든 ok background entry 에 owned ASCII 1+ entries (normalize 후) → OK. 차단 path (모두 raise): - bg on + cp None → block (round 4 BLOCKING 1 — silent {} 통과 차단). - bg on + cp schema<2 → block (옛 v4 cp 잔존). - bg on + cp ok 인데 어느 ok background entry 라도 owned 부재/빈 → block. - bg on + cp ok 인데 owned entry 에 non-ASCII 포함 → block (round 7 BLOCKING 1 — 수동 편집 / 부분 산출 stale ["문"] silent drop 차단). - bg on + cp ok 인데 normalize 후 owned 가 빈 list → block (entry 가 모두 whitespace 등으로 silent drop 되는 케이스 차단). """ if not background_mode_on: return if bp_cp is None: # round 4 BLOCKING 1: bg-on + None 도 fail-fast. raise AppError( code="step.contract_violation", message=( f"{where}: background_mode is on but background_prompt cp is " "missing. force background_prompt 먼저 실행 필요." ), ) schema_v = bp_cp.get("schema_version") or 0 if schema_v < 2: raise AppError( code="step.contract_violation", message=( f"{where}: background_prompt cp schema_version={schema_v} (<2) — " f"옛 v4 cp 잔존. force background_prompt 먼저 실행 필요." ), ) backgrounds = (bp_cp.get("data", {}) or {}).get("backgrounds", {}) or {} for bid, entry in backgrounds.items(): if not isinstance(entry, dict): continue if entry.get("status") != "ok": continue owned = entry.get("objects_owned_by_background") if not owned or not isinstance(owned, list): raise AppError( code="step.contract_violation", message=( f"{where}: background_prompt[{bid}] missing or empty " "objects_owned_by_background — partial v5 cp." ), ) # round 7 BLOCKING 1: raw item ASCII 재검증 — 수동 편집 / 부분 산출 # ["문"] silent drop 차단. for idx, item in enumerate(owned): if not isinstance(item, str): continue try: item.encode("ascii") except UnicodeEncodeError: raise AppError( code="step.contract_violation", message=( f"{where}: background_prompt[{bid}].objects_owned_by_" f"background[{idx}] {item!r} contains non-ASCII. owned " "MUST be English canonical common nouns " "(round 4 Q2=B / round 7 BLOCKING 1)." ), ) # round 7 BLOCKING 1: normalize 후 non-empty 검증 — entry 가 모두 # whitespace / non-string 으로 silent drop 되는 케이스 차단. normalized = normalize_owned_list(owned) if not normalized: raise AppError( code="step.contract_violation", message=( f"{where}: background_prompt[{bid}].objects_owned_by_" "background empty after normalize — silent drop 차단." ), ) def format_redraw_violations_block(redraw_violations: List[Dict[str, Any]]) -> str: """owned judge redraw_violation evidence 를 repair prompt 용 텍스트 블록으로. 각 violation = {owned_object, violating_phrase, reason, verdict}. caller 가 이미 verdict == "redraw_violation" 로 필터링한 list 를 넘긴다. FINDING 7 (e2e-bughunt-v1): `scene_detail_owned_repair` user_template 의 {redraw_violations_block} placeholder 치환에 사용. pure 함수 — LLM 호출 없음. """ lines = [] for v in redraw_violations: owned_object = v.get("owned_object", "") violating_phrase = v.get("violating_phrase", "") reason = v.get("reason", "") lines.append( f'- owned 객체: "{owned_object}" | 위반 구절: "{violating_phrase}" ' f"| 사유: {reason}" ) return "\n".join(lines) # --------------------------------------------------------------------------- # Task 14 Wave 2 — visible-prop / background-owned namespace overlap # reconciliation. # --------------------------------------------------------------------------- _PROP_TERM_NORMALIZE_RE = re.compile(r"\s+") _RECONCILE_NOTE_FMT = ( "Deterministic prop-owned namespace reconciliation: visible prop {pp} " "owns token {tok}; prompt uses {pp} in the source phrase." ) # Wave 2 fixup (Task 14) — fixed stopword set excluded from ASCII content # token expansion of prop terms. Function-word only; no domain-specific # vocabulary (e.g. nouns like "map", "lantern", "tv" are NOT stopwords). _PROP_TERM_STOPWORDS = frozenset({ "a", "an", "the", "of", "at", "in", "on", "to", "with", "and", "or", "for", "from", "by", "into", "over", "under", "across", "between", }) # ASCII content token regex — runs against the already-normalized # (strip+lowercase+collapse-whitespace) prop term. Matches word-boundary # runs of [a-z0-9]; non-ASCII glyphs (e.g. Korean Hangul) produce no # matches and are skipped by tokenization. _PROP_TERM_ASCII_TOKEN_RE = re.compile(r"[a-z0-9]+") def _normalize_prop_term(s: str) -> str: """prop term normalize — strip + lowercase + collapse internal whitespace. producer (SceneContextLoader.prop_term_map prebuild) 과 consumer (`reconcile_owned_prop_namespace_overlap` 의 owned_token 비교) 양쪽에서 동일 normalize 를 사용한다. caller 가 빈 / whitespace-only 입력을 이미 거른 후 호출하는 것을 가정. """ return _PROP_TERM_NORMALIZE_RE.sub(" ", s.strip().lower()) def _expand_prop_term_variants(src: str) -> set[str]: """Wave 2 fixup (Task 14) — expand ONE prop source string into the set of normalized terms that should be added to a prop's owned-match term set. For an ASCII string: full normalized phrase plus each ASCII content token found by ``re.findall(r"[a-z0-9]+", normalized)``, minus the fixed stopword set (``_PROP_TERM_STOPWORDS``). Tokens are word- boundary exact, so ``"mapped territory"`` yields ``{"mapped territory", "mapped", "territory"}`` and NOT ``"map"``. For a non-ASCII (Korean / mixed) string: only the full normalized phrase. ``re.findall(r"[a-z0-9]+", ...)`` returns ``[]`` against a pure Korean phrase, and we do not tokenize Korean. Mixed-script strings (e.g. ``"Korean 지도 paper map"``) yield the full normalized phrase plus the ASCII tokens only; Hangul glyphs are not split into sub-tokens. Empty / whitespace-only input → empty set. No length cutoff — 2-letter ASCII tokens like ``"tv"`` survive (unless they appear in the stopword set). This util is pure (no DB / I/O / global state). Producer (``SceneContextLoader._load_entity_canon_prop_term_map``) owns token expansion; the consumer helper (``reconcile_owned_prop_namespace_overlap``) stays membership-only. Args: src: a single prop source string (entity_canon.name / description / t2i_prompt or an entity_alias.alias row). Returns: set of normalized term strings to union into the prop's ``prop_term_map`` entry. Empty set if the input normalizes to empty. """ normalized = _normalize_prop_term(src) if not normalized: return set() out: set[str] = {normalized} for tok in _PROP_TERM_ASCII_TOKEN_RE.findall(normalized): if tok in _PROP_TERM_STOPWORDS: continue out.add(tok) return out def reconcile_owned_prop_namespace_overlap( violations: List[Dict[str, Any]], owned_object_usage: List[Dict[str, Any]], t2i_prompt: str, visible_entities: List[str], prop_term_map: Dict[str, frozenset] | None, ) -> List[Dict[str, Any]]: """Task 14 Wave 2 — visible prop ↔ background-owned namespace overlap fix. background_prompt 가 어떤 location 의 owned 환경 객체 (e.g. ``"map"``) 를 선언했고, 같은 shot 의 visible_entities 에 같은 의미의 visible prop (e.g. ``P06 = 종이 지도``) 이 함께 있을 때, scene_detail T2I prompt 가 prop short_id (``P06``) 로 prop 을 명시적으로 anchor 하면 owned-judge 가 overlap 토큰을 redraw 로 오인하는 false-positive 가 발생한다. 이 helper 는 consumer boundary 의 결정론적 reclass — judge 가 ``redraw_violation`` 으로 찍은 entry 중, 같은 shot 의 visible prop short_id 와 source phrase 가 문법적으로 묶인 경우만 ``anchor_reference`` 로 다운그레이드한다. 재분류 4개 gate (모두 통과시에만 reclass): Gate A (visible-prop gate) ``visible_entities`` 안에 ``P##`` 접두 short_id 가 존재 AND 그 short_id 가 ``prop_term_map`` 의 key 에 존재. Gate B (prompt-literal gate) ``t2i_prompt`` 의 literal substring 에 ``P##`` 가 포함. Gate C (term-match gate) ``v["owned_object"]`` 를 ``_normalize_prop_term`` 으로 normalize 한 뒤 ``prop_term_map[P##]`` set 에 full-string membership 검사. multi-word / single-word 모두 동일 규칙 — helper 는 prop term 을 sub-word 로 split 하지 않는다 (예: owned_object ``"map"`` 은 prop term set ``{"map", "paper map"}`` 에는 matched 되지만 ``{"oil lantern"}`` (bare ``"lantern"`` 미포함) / ``{"mapped territory"}`` 에는 not matched). 캐논 prop 표현에 bare common-noun 토큰이 명시적으로 등록되어 있을 때만 reclass — 옛 background owned ASCII canonical 과 prop canon term 양쪽이 같은 어휘로 등재되어야 한다는 의도된 boundary. Gate D (violating-phrase linkage gate, W4b-tightened) ``v["violating_phrase"]`` 가 ``P##`` 와 ``v["owned_object"]`` 둘 다 literal substring 으로 포함 (case-insensitive — 양쪽 모두 ``lower()`` 후 substring 검사). After W4b: gate 4 requires ``violating_phrase`` to contain both ``P##`` and ``owned_object`` as case-insensitive substrings. ``owned_object_usage`` source_phrase echo alone is **insufficient** — the prior 4(a) branch was dropped because the LLM can declare a source_phrase containing the prop's ``P##`` even when the actual ``violating_phrase`` describes a real redraw with no ``P##`` reference. The ``owned_object_usage`` parameter is retained in the signature for input-immutability assertions and future audit; it is not consulted for the linkage gate. 여러 ``P##`` 가 후보면 ``sorted(visible_entities)`` 순서에서 처음 4 gate 를 통과하는 것 선택 (결정론). 매치된 entry 는 새 dict (shallow copy of ``v``) 로 생성 — 입력 dict 는 절대 mutate 안 함 (G9 invariant). 새 dict 의 ``verdict`` 를 ``"anchor_reference"`` 로 바꾸고, ``reason`` 끝에 정확히 한 공백 + 결정론적 note 를 append. 이미 ``verdict == "anchor_reference"`` 인 entry 는 그대로 pass-through — idempotent. Defensive short-circuit (G10) — 모두 input list 의 reference identity 보존: - ``prop_term_map is None`` 또는 ``{}`` → ``violations`` 그대로 반환. - ``violations`` 가 empty → 그대로 반환. - ``violations`` 안에 visible-prop gate 통과 후보 없음 → 그대로 반환. Args: violations: owned-judge ``violations`` list. 각 entry 는 dict (owned_object/violating_phrase/reason/[verdict]). verdict 누락 또는 ``"anchor_reference"`` 는 pass-through. owned_object_usage: 같은 shot 의 merged owned_object_usage. **After W4b: not consulted for reclassification** — gate D dropped the source_phrase echo branch. Parameter is retained in the signature for input-immutability assertions (G9 invariant) and future audit / debug visibility; the helper does not mutate it and does not index into it for gate decisions. t2i_prompt: 현재 variation 의 t2i_prompt 문자열. Gate B literal substring 검사 입력. visible_entities: shot 의 visible_entities list (C##/L##/P##/C##O## mixed). helper 가 ``P`` 접두만 추출. prop_term_map: short_id (``P##``) → 정규화된 prop term frozenset. ``SceneContextLoader._load_entity_canon_prop_term_map`` 가 main thread 에서 미리 build 한 dict. helper 는 read-only. Returns: 새 list. 재분류 발생 entry 만 새 dict (shallow copy + verdict / reason 갱신). 나머지 entry 는 원본 reference 그대로 reuse. defensive short-circuit 시엔 ``violations`` 자체 reference 를 그대로 반환. Pure 함수 — DB / I/O / 글로벌 상태 접근 없음. """ # G10a / G10b — short-circuit: no prop_term_map → input reference. if not prop_term_map: return violations if not violations: return violations # Gate A precompute — visible_entities 안 P## ∩ prop_term_map keys (sorted). visible_props_sorted = sorted( sid for sid in visible_entities if isinstance(sid, str) and sid.startswith("P") and sid in prop_term_map ) if not visible_props_sorted: # 어떤 visible prop 도 prop_term_map 에 매칭 안 됨 → reclass 후보 없음. return violations # Gate B precompute — prompt literal substring 검사 결과 캐시 (case- # insensitive: prompt 를 lower 한 사본 1회 비교, P## 도 lower 비교). _t2i_lower = (t2i_prompt or "").lower() prop_in_prompt = { pp: (pp.lower() in _t2i_lower) for pp in visible_props_sorted } # W4b: owned_object_usage source_phrase echo path (formerly Gate D(a)) # was dropped — gate D now consults only violating_phrase literal # substring containment. owned_object_usage is still passed in for # signature stability / input-immutability assertions / future audit. out: List[Dict[str, Any]] = [] any_reclass = False for v in violations: # G10e — entry 가 dict 아님 / verdict 누락 → 그대로 pass-through. if not isinstance(v, dict) or "verdict" not in v: out.append(v) continue # G10d — 이미 anchor_reference → idempotent pass-through. if v.get("verdict") != "redraw_violation": out.append(v) continue owned_object = v.get("owned_object", "") if not isinstance(owned_object, str) or not owned_object: out.append(v) continue norm_tok = _normalize_prop_term(owned_object) if not norm_tok: out.append(v) continue violating_phrase = v.get("violating_phrase", "") or "" violating_phrase_lower = violating_phrase.lower() owned_object_lower = owned_object.lower() matched_pp: str | None = None for pp in visible_props_sorted: # Gate B — prompt literal substring (case-insensitive). if not prop_in_prompt.get(pp, False): continue # Gate C — full-string membership in prop_term_map[pp] (no # sub-word splitting). multi-word vs single-word 동일 규칙. prop_terms = prop_term_map.get(pp) or frozenset() if norm_tok not in prop_terms: continue # Gate D — W4b tightened: violating_phrase MUST contain BOTH the # P## and the owned_object as case-insensitive substrings. The # prior 4(a) source_phrase-echo branch is dropped — a # source_phrase containing P## alone is no longer sufficient. # This protects against false-pass on shots where the LLM truly # intends a redraw but happens to echo the prop's short_id in # source_phrase (e.g. S18_Shot12 photo-frame interior redraw). if ( pp.lower() not in violating_phrase_lower or owned_object_lower not in violating_phrase_lower ): continue matched_pp = pp break if matched_pp is None: out.append(v) continue # All 4 gates passed — build NEW dict (shallow copy) with # verdict=anchor_reference and reason augmented by deterministic note. new_entry = dict(v) new_entry["verdict"] = "anchor_reference" note = _RECONCILE_NOTE_FMT.format(pp=matched_pp, tok=owned_object) prev_reason = v.get("reason", "") if isinstance(prev_reason, str) and prev_reason: new_entry["reason"] = prev_reason + " " + note else: new_entry["reason"] = note out.append(new_entry) any_reclass = True if not any_reclass: # 어떤 entry 도 4 gate 통과 못 함 — caller 가 가능하면 원본 reference # 그대로 사용할 수 있도록 input 을 반환. return violations return out