"""background_share_plan — 에피소드 전체 배경 공유·참조 계획 (2026-07-19). 재설계 B-2 (사용자 확정): '비슷한 배경이 있는 곳은 배경 하나(seed 기반)를 만들고, 그 뒤 샷들은 전부 이전 샷 참조' — 샷별 개별 판정이 아니라 에피소드 전체를 조망하는 LLM 계획으로 '배경 참조 vs 앞쪽 샷 참조'를 구분 지휘한다. 씬 경계 무관 재방문 포함, 장소 동일성 판단=LLM(인용 근거 의무). 스틸 소비 시 이 계획이 shot_ref_classify prev 판정의 상위 권위(계획 부재=기존 판정 fail-safe). 결정론 검증(fail-closed): 전 선택 샷 커버·중복 0, 그룹 첫 샷=background, prev 앵커=같은 그룹의 스토리상 앞 샷, 인용 실재(whitespace 정규화). """ from __future__ import annotations import logging import re from typing import Any, Dict, List, Mapping, Sequence logger = logging.getLogger(__name__) _MODULE = "background_share_plan" BG_AUTHORITIES = ("seed", "plate", "interior") REF_PLANS = ("background", "prev") # 검증 계약 버전 — step config_hash 스탬프(기존 CP 감지). v2 (2026-07-21 # E2E10 Codex 재리뷰): group_key 유일성 fail-closed. VALIDATION_VERSION = 2 PROMPT_VERSION_MAP = { "1": "1.202607190212", } def resolve_prompt_version(version: str) -> str: if version not in PROMPT_VERSION_MAP: raise ValueError(f"background_share_plan 프롬프트 버전 없음: {version}") return PROMPT_VERSION_MAP[version] def build_share_schema() -> Dict[str, Any]: evidence = { "type": "array", "minItems": 1, "items": { "type": "object", "additionalProperties": False, "properties": { "scene_index": {"type": "integer"}, "quote_ko": {"type": "string", "minLength": 1}, }, "required": ["scene_index", "quote_ko"], }, } return { "type": "object", "additionalProperties": False, "properties": { "share_groups": { "type": "array", "minItems": 1, "items": { "type": "object", "additionalProperties": False, "properties": { "group_key": {"type": "string", "minLength": 1}, "bg_authority": {"type": "string", "enum": list(BG_AUTHORITIES)}, "shot_tags": {"type": "array", "minItems": 1, "items": {"type": "string"}}, "rationale_ko": {"type": "string"}, "evidence": evidence, }, "required": ["group_key", "bg_authority", "shot_tags", "evidence"], }, }, "shot_plans": { "type": "object", "additionalProperties": { "type": "object", "additionalProperties": False, "properties": { "ref_plan": {"type": "string", "enum": list(REF_PLANS)}, "prev_anchor_tag": {"type": "string"}, "rationale_ko": {"type": "string"}, }, "required": ["ref_plan"], }, }, }, "required": ["share_groups", "shot_plans"], } def _tag_key(tag: str): m = re.match(r"S(\d+)sh(\d+)$", str(tag)) return (int(m.group(1)), int(m.group(2))) if m else (10**9, 10**9) def _norm_ws(text: str) -> str: return " ".join(str(text).split()) def validate_share_plan( result: Mapping[str, Any], shot_tags: Sequence[str], scene_texts: Mapping[int, str], ) -> List[str]: """결정론 완전성 검증 — 위반 리스트 반환 (fail-closed 재시도용). v2 (2026-07-21 E2E10 Codex 재리뷰): group_key 유일성 fail-closed — 소비부(스틸 groupbg record key·group_sig·group_of 색인, first_of_group 검증)가 group_key 를 실질 PK 로 쓰므로 중복 키는 서로 다른 장소 그룹을 한 groupbg/sidecar 로 합치고 첫 샷/prev 검증을 약화시킨다. """ violations: List[str] = [] groups = result.get("share_groups") or [] plans = result.get("shot_plans") or {} expected = list(shot_tags) seen: Dict[str, str] = {} group_of: Dict[str, str] = {} group_keys_seen: set = set() for g in groups: gk = str((g or {}).get("group_key") or "") if gk: if gk in group_keys_seen: violations.append( f"group_key '{gk}' 중복 — 그룹 키는 유일해야 함" ) group_keys_seen.add(gk) if (g or {}).get("bg_authority") not in BG_AUTHORITIES: violations.append(f"그룹 '{gk}' bg_authority 무효") tags = list((g or {}).get("shot_tags") or []) for t in tags: if t in seen: violations.append(f"샷 {t} 그룹 중복 배정({seen[t]}·{gk})") seen[t] = gk group_of[t] = gk # 그룹 내 스토리 순서 검증 if tags != sorted(tags, key=_tag_key): violations.append(f"그룹 '{gk}' shot_tags 가 스토리 순이 아님") # 인용 실재 (whitespace 정규화 — lane_plan 관례) for ev in (g or {}).get("evidence") or []: si = (ev or {}).get("scene_index") quote = _norm_ws((ev or {}).get("quote_ko") or "") src = _norm_ws(scene_texts.get(si) or "") if not quote or quote not in src: violations.append( f"그룹 '{gk}' 인용 원문 부재 (scene {si})") missing = [t for t in expected if t not in seen] if missing: violations.append(f"샷 커버리지 누락: {missing[:8]}") extra = [t for t in seen if t not in set(expected)] if extra: violations.append(f"미선택 샷 배정: {extra[:8]}") first_of_group: Dict[str, str] = {} for g in groups: tags = list((g or {}).get("shot_tags") or []) if tags: first_of_group[str(g.get("group_key") or "")] = tags[0] for t in expected: p = plans.get(t) if not isinstance(p, dict): violations.append(f"샷 {t} shot_plan 누락") continue rp = p.get("ref_plan") if rp not in REF_PLANS: violations.append(f"샷 {t} ref_plan 무효: {rp!r}") continue gk = group_of.get(t) if rp == "background": continue # prev — 앵커 검증 anchor = p.get("prev_anchor_tag") if not anchor: violations.append(f"샷 {t} prev 앵커 누락") continue if group_of.get(anchor) != gk: violations.append(f"샷 {t} prev 앵커 {anchor} 가 다른 그룹") elif _tag_key(anchor) >= _tag_key(t): violations.append(f"샷 {t} prev 앵커 {anchor} 가 앞 샷이 아님") for gk, first in first_of_group.items(): p = plans.get(first) or {} if p.get("ref_plan") != "background": violations.append( f"그룹 '{gk}' 첫 샷 {first} 은 background 여야 함") return violations def run_background_share_plan( *, shots_block: str, scenes_block: str, shot_tags: Sequence[str], scene_texts: Mapping[int, str], project_config: Dict[str, Any] | None = None, prompt_version: str = "1", max_attempts: int = 3, call_structured_fn=None, opik_metadata: Dict[str, Any] | None = None, ) -> Dict[str, Any]: """계획 저작+검증 재시도 — 소진=AppError (fail-closed, lane_plan 관례). 반환 {"plan": result, "attempts": n}. """ from app.core.errors import AppError from app.modules.prompt_loader import load_prompt if call_structured_fn is None: from app.modules.llm.llm_client import call_structured call_structured_fn = call_structured resolved = resolve_prompt_version(prompt_version) system = load_prompt(_MODULE, "system", version=resolved) user = load_prompt( _MODULE, "user_template", version=resolved, scenes_block=scenes_block, shots_block=shots_block, ) schema = build_share_schema() violations: List[str] = [] for attempt in range(1, max_attempts + 1): content = user if not violations else ( user + "\n\nPRIOR ATTEMPT VIOLATIONS (fix all):\n" + "\n".join(f"- {v}" for v in violations) ) result = call_structured_fn( _MODULE, system, content, schema, project_config=project_config, schema_name=_MODULE, opik_metadata=opik_metadata, ) violations = validate_share_plan(result, shot_tags, scene_texts) if not violations: return {"plan": result, "attempts": attempt} logger.warning( "background_share_plan: 검증 위반(attempt %d): %s", attempt, violations[:6], ) # 슬라이스 E 실측(2026-07-23): 이 경로가 positional 호출로 TypeError 를 # 내며 실제 위반 목록을 가리던 잠복 버그 — AppError 계약(code/message) # 으로 교정 (재시도 소진 사유가 로그·응답에 그대로 드러나야 한다). raise AppError( code="step.contract_violation.background_share_plan", message=( f"background_share_plan 검증 실패(재시도 소진): {violations[:6]}" ), status_code=422, )