"""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. # ★v3 로 올렸다가 **되돌렸다** (2026-09-05). 이 판의 변경은 스텝을 **더 # 너그럽게** 만들 뿐 산출 모양을 안 바꾼다 — 곁가지 칸을 떼어 내도 남는 것은 # 종전과 같은 세 칸이고, 재시도가 도는 것도 산출이 아니라 도달 경로다. # 그러니 옛 계약에서 만들어진 완료 CP 는 **새 계약에서도 그대로 유효하다.** # 올렸더니 이미 완주한 1화의 CP 가 drift 로 걸려 이미지 99장을 통째로 다시 # 사게 됐다 — 얻는 것 없이 유료 하류를 다시 사는 부류다. VALIDATION_VERSION = 2 PROMPT_VERSION_MAP = { "1": "1.202607190212", # v2 (2026-09-05 실측 2회): 모델이 `group_key` 를 **`shot_plans` 항목 안**에 # 넣어 스키마 검증이 통째로 버렸다(`shot_plans/S9sh5`, `shot_plans/S7sh4`). # 샷과 그룹의 연결은 `share_groups[].shot_tags` 로 이미 표현되는데, v1 은 # `group_key` 를 이름 규칙으로만 언급하고 **어디에 속하는 칸인지**는 말하지 # 않았다. v2 가 그 자리를 명시하고, 항목이 세 칸뿐임을 못박는다. "2": "2.202609051240", } #: `shot_plans` 항목이 실제로 갖는 칸 — **한 곳에서만** 적는다. SHOT_PLAN_FIELDS = ("ref_plan", "prev_anchor_tag", "rationale_ko") def normalize_shot_plans(result: Dict[str, Any]) -> Dict[str, List[str]]: """`shot_plans` 항목에서 **모르는 칸을 떼어 낸다**. 뗀 것을 돌려준다. ★조용히 버리지 않는다 — 무엇을 뗐는지 호출자가 받아 기록한다. 그래야 「모델이 자꾸 무엇을 더 넣는가」를 나중에 되짚을 수 있다. ★샷과 그룹의 연결은 `share_groups[].shot_tags` 가 이미 표현한다. 그래서 샷마다 `group_key` 를 또 적는 것은 **중복이지 정보가 아니다** — 떼어도 잃는 것이 없다. """ dropped: Dict[str, List[str]] = {} plans = (result or {}).get("shot_plans") if not isinstance(plans, dict): return dropped for tag, entry in plans.items(): if not isinstance(entry, dict): continue extra = [k for k in entry if k not in SHOT_PLAN_FIELDS] if not extra: continue for k in extra: entry.pop(k, None) dropped[str(tag)] = sorted(extra) return dropped 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": { # ★★★`additionalProperties: False` 를 **뺐다** # (2026-09-05, 실측 3회). 모델이 `group_key` 하나를 # 더 넣었다는 이유로 **멀쩡한 계획 전체가 버려졌고** # 스텝이 죽었다(`shot_plans/S9sh5`·`S7sh4` 두 화에서). # 이 항목의 뜻은 아래 세 칸으로 온전히 정해진다 — # 곁가지 칸은 버리면 그만이지 계획을 버릴 이유가 아니다. # ★대신 **코드가 결정적으로 막는다**: 아래 # `normalize_shot_plans` 가 모르는 칸을 떼어 내고 # 무엇을 뗐는지 기록한다. 조용히 통과시키지 않는다. "type": "object", "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) ) # ★★★스키마 오류를 **이 루프 안으로 들인다** (2026-09-05 실측). # 이 함수는 3번 재시도하게 돼 있는데, 스키마 위반은 # `call_structured` 안에서 예외로 터져 루프를 **통째로 건너뛰었다** — # 그래서 곁가지 칸 하나에 한 번 만에 죽었다(실측 3회). 위반을 문자로 # 받아 다음 시도의 입력에 넣으면 다른 위반과 똑같이 고쳐 쓸 수 있다. from app.modules.llm.safety import SchemaValidationError try: result = call_structured_fn( _MODULE, system, content, schema, project_config=project_config, schema_name=_MODULE, opik_metadata=opik_metadata, ) except SchemaValidationError as exc: violations = [f"schema: {exc}"] logger.warning( "background_share_plan: 스키마 위반(attempt %d) — 다시 쓴다: %s", attempt, str(exc)[:200]) if attempt >= max_attempts: raise continue # ★모르는 칸은 **떼어 내고 기록한다**. 계획을 통째로 버리지 않는다. _dropped = normalize_shot_plans(result) if _dropped: logger.warning( "background_share_plan: shot_plans 곁가지 칸 제거 %d건 — %s", len(_dropped), dict(list(_dropped.items())[:5])) result.setdefault("_dropped_shot_plan_fields", _dropped) 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, )