"""W20B: real-LLM planner provider for shot_aware_bg_render_plan. Production-adjacent helper that, **when explicitly resolved by the step wrapper**, performs a single ``litellm.completion`` call to drive the dwelling-scoped reference-graph DAG planner. The output is the parsed graph dict ready for ``build_render_plan_for_fp`` consumption. Why ``litellm.completion`` directly rather than the project's ``call_structured`` router: the router applies retry / fallback / temperature defaults and re-shapes ``response_format`` in ways that trip OpenAI's strict ``json_schema`` mode (notably forwarding ``temperature`` to a gpt-5.5 route that rejects unknown tunable params and forwarding the prompt-pack schema with constraint keys OpenAI strict mode rejects). This module is the single, lock-stable entry point that: - reads the prompt pack itself (system / user_template / schema), - deep-copies + recursive-strips the schema for OpenAI strict mode, - performs every preflight (arg shape, env, prompt pack, litellm capability) before any network call, - issues exactly one ``litellm.completion`` call with ``num_retries=0`` and no temperature, - applies fail-closed response guards, - validates against the ORIGINAL pack schema (local jsonschema is the SOT — sanitized copy is only the request shape), - hands off to the production ``validate_llm_output`` (DAG / refs / camera / etc.) and fails-closed on validator diagnostics. LLM / image / VLM API call 0 in tests (every test monkeypatches ``litellm.completion``). The default production code path keeps ``settings.shot_aware_bg_render_plan_real_provider_enabled = False``, so the step never resolves to this helper. The module's only external dependency is ``litellm``, lazy-imported inside the function body — callers that never flip the selector pay zero import cost. """ from __future__ import annotations import json import logging import os from pathlib import Path from typing import Any, Dict, FrozenSet, List, Optional, Tuple logger = logging.getLogger(__name__) # ───────────────────────────────────────────────────────────────────── # Lock constants — tests pin every one of these. # ───────────────────────────────────────────────────────────────────── PLANNER_MODEL_DEFAULT: str = "openai/gpt-6-astra" MAX_COMPLETION_TOKENS_DEFAULT: int = 16000 # W20F7-A (Codex 2026-05-28): 6000→16000. gpt-5.5 reasoning+output 합산 budget. 큰 dossier (BG 6+, long scene text) 의 reasoning 폭주로 finish_reason=stop + content="" 빈 응답이 fp_l05/l09 에서 재현됨. retry 도입 없이 budget 확보로 가설 검증. TIMEOUT_SECONDS_DEFAULT: int = 180 RESPONSE_SCHEMA_NAME: str = "shot_aware_bg_render_plan_v2" # v3 (2026-07-23 W20F10, Codex 합의): 2.202605291800 전체 사본 + # graph_anchor_repair.md 스템 신설 — graph/anchor-only validator 실패에 # validator diagnostics·기대 renderable bg 목록·허용 anchor 후보·이전 응답 # 전문을 교정 블록으로 병기한 targeted retry 1회 허용(슬라이스 E 실측: # 단일 bg 퇴화 fp 에서 무피드백 독립 샘플 4회 전부 빈 그래프 — 2026-05-28 # 'graph/anchor retry 금지' 계약의 개정이라 신규 팩 버전+config_hash 자연 # drift). system/user_template/schema=byte 사본, SCHEMA_VERSION 유지. PROMPT_VERSION: str = "3.202607231435" # Prompt pack discovery — TheRoad-I1/prompts/_base///. # The file lives at backend/app/modules/pipeline/, so the repo root is # five parents up. _PROMPT_PACK_DIR: Path = ( Path(__file__).resolve().parent.parent.parent.parent.parent / "prompts" / "_base" / "shot_aware_bg_render_plan" / PROMPT_VERSION ) # OpenAI's strict structured-outputs mode rejects these keywords. Code # walks the prompt-pack schema recursively and strips every occurrence # before passing the sanitized copy to ``response_format``. The # ORIGINAL pack schema retains them — it is the local jsonschema SOT. _OPENAI_UNSUPPORTED_SCHEMA_KEYS: FrozenSet[str] = frozenset({ "minLength", "maxLength", "minItems", "maxItems", "uniqueItems", "minimum", "maximum", "exclusiveMinimum", "exclusiveMaximum", "multipleOf", "pattern", "format", "anyOf", "oneOf", "allOf", "not", "patternProperties", "contains", "minContains", "maxContains", }) class ShotAwareBgRenderPlanProviderError(Exception): """Fail-closed signal for the W20B planner provider path. W20F10 (Codex 조건 5): retry 경로에서 실제 completion 호출이 2회였던 실패는 ``completion_call_count`` 속성으로 실비용을 보존한다 — caller (build_render_plan_for_fp 실패 감사)가 real_api_call_counts.llm 에 반영. 미설정=1 (기존 단일 호출 semantics). """ completion_call_count: int = 1 def _provider_error( msg: str, *, completion_calls: int = 1 ) -> ShotAwareBgRenderPlanProviderError: err = ShotAwareBgRenderPlanProviderError(msg) err.completion_call_count = completion_calls return err # ───────────────────────────────────────────────────────────────────── # Pure helpers (no I/O, no network) # ───────────────────────────────────────────────────────────────────── def _sanitize_schema_for_openai_strict(schema: Any) -> Any: """Deep-copy + recursive strip of OpenAI strict-mode unsupported keys. Returns a fresh tree; the input is not mutated. Additionally enforces the OpenAI Structured Outputs strict-mode contract that **every object property must be listed in ``required``**. The original prompt-pack schema marks some fields optional (e.g. ``selected_refs.items.space_description``, ``camera_decision.framing_notes``) so local jsonschema validation can accept LLM responses that omit them. The request-side sanitized copy must close that gap so strict mode does not reject the ``response_format`` before the model is even invoked. The fill keeps the original ``properties`` insertion order (Python dict ordering is the language guarantee since 3.7) so the request payload is deterministic across runs. """ if isinstance(schema, dict): out: Dict[str, Any] = {} for key, value in schema.items(): if key in _OPENAI_UNSUPPORTED_SCHEMA_KEYS: continue out[key] = _sanitize_schema_for_openai_strict(value) if out.get("type") == "object" and isinstance( out.get("properties"), dict ): out["required"] = list(out["properties"].keys()) return out if isinstance(schema, list): return [_sanitize_schema_for_openai_strict(item) for item in schema] return schema def _load_prompt_pack() -> Dict[str, Any]: """Load system / user_template / schema from the prompt pack dir. Fail-closed if any artifact is missing, unreadable, empty, or — for ``schema.json`` — not valid JSON. Caller treats this as a preflight failure (completion call count stays 0). """ pack_dir = _PROMPT_PACK_DIR if not pack_dir.exists() or not pack_dir.is_dir(): raise ShotAwareBgRenderPlanProviderError( f"prompt pack dir missing: {pack_dir!s}" ) sys_path = pack_dir / "system.md" usr_path = pack_dir / "user_template.md" schema_path = pack_dir / "schema.json" repair_path = pack_dir / "graph_anchor_repair.md" for p in (sys_path, usr_path, schema_path, repair_path): if not p.exists() or not p.is_file(): raise ShotAwareBgRenderPlanProviderError( f"prompt pack file missing: {p!s}" ) try: system_prompt = sys_path.read_text(encoding="utf-8") user_template = usr_path.read_text(encoding="utf-8") schema_raw = schema_path.read_text(encoding="utf-8") graph_anchor_repair_template = repair_path.read_text( encoding="utf-8" ) except Exception as exc: raise ShotAwareBgRenderPlanProviderError( f"prompt pack read failed: {type(exc).__name__}: {exc}" ) from exc if not system_prompt.strip(): raise ShotAwareBgRenderPlanProviderError( "prompt pack system.md is empty" ) if not user_template.strip(): raise ShotAwareBgRenderPlanProviderError( "prompt pack user_template.md is empty" ) if not graph_anchor_repair_template.strip(): raise ShotAwareBgRenderPlanProviderError( "prompt pack graph_anchor_repair.md is empty" ) try: schema = json.loads(schema_raw) except (TypeError, ValueError) as exc: raise ShotAwareBgRenderPlanProviderError( f"prompt pack schema.json not valid JSON: {exc}" ) from exc if not isinstance(schema, dict): raise ShotAwareBgRenderPlanProviderError( f"prompt pack schema.json must be a JSON object " f"(got {type(schema).__name__})" ) return { "system_prompt": system_prompt, "user_template": user_template, "schema": schema, "graph_anchor_repair_template": graph_anchor_repair_template, } def _build_user_prompt( *, user_template: str, fp_id: str, readback_status: str, dossier: Dict[str, Any], geometry: Dict[str, Any], shot_readiness: Dict[str, Any], candidate_catalog: List[Any], ) -> str: """Substitute the user_template placeholders with structured JSON blocks. The substitutions carry no scenario-specific tokens — every block is a verbatim ``json.dumps(..., sort_keys=True)`` of upstream data. W20F8: ``camera_candidates_block`` 별도 섹션에 per-unit camera/look_at candidate set 만 verbatim 추출. LLM 이 exact-copy 위반으로 coordinate 를 invent 하지 않도록 prompt 안에서 후보 set 을 두 번 노출 (geometry_block 안 + 별도 W20F8 섹션). **이 중복은 의도된 것이라 prompt-diet 대상이 아니다.** 2026-08-03 prompt-diet 2단계: ``per_bg_render_facts_by_bg_id`` 는 바로 아래 ``per_bg_facts_block`` 으로 따로 실리는데 ``dossier_block`` 안에도 그대로 들어 있어 같은 JSON 이 한 콜에 두 번 나갔다. 템플릿의 두 섹션(``## Base location dossier`` / ``## Per-BG render facts``)이 이미 분리돼 있고 dossier 섹션이 facts 포함을 말하지 않으므로, dossier 사본에서만 그 키를 빼도 전달 내용의 손실이 없다. """ per_bg_facts = (dossier or {}).get("per_bg_render_facts_by_bg_id") or {} # dossier 가 dict 가 아니면(None 등) 기존 직렬화 결과를 그대로 유지한다. dossier_for_block = ( { k: v for k, v in dossier.items() if k != "per_bg_render_facts_by_bg_id" } if isinstance(dossier, dict) else dossier ) per_bg_staging = (shot_readiness or {}).get("per_bg") or {} camera_candidates = { "camera_cell_candidates_per_unit": ( (geometry or {}).get("camera_cell_candidates_per_unit") or {} ), "look_at_cell_candidates_per_unit": ( (geometry or {}).get("look_at_cell_candidates_per_unit") or {} ), } return user_template.format( fp_id=fp_id, readback_status=readback_status, dossier_block=json.dumps(dossier_for_block, sort_keys=True), geometry_block=json.dumps(geometry, sort_keys=True), camera_candidates_block=json.dumps(camera_candidates, sort_keys=True), per_bg_facts_block=json.dumps(per_bg_facts, sort_keys=True), shot_staging_block=json.dumps(per_bg_staging, sort_keys=True), candidate_catalog_block=json.dumps( list(candidate_catalog), sort_keys=True ), ) def _build_messages( *, system_prompt: str, user_prompt: str ) -> List[Dict[str, Any]]: return [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ] def _build_camera_repair_user_prompt( *, previous_parsed: Dict[str, Any], camera_diagnostics: List[str], geometry: Dict[str, Any], ) -> str: """W20F9 — camera validator-fail targeted repair prompt. Codex 명시 (2026-05-28): retry 입력 = validator diagnostic + camera candidate block + 기존 graph/reference 보존, camera_decision 의 camera_cell/look_at_cell/lens/fov 만 후보 set 에서 verbatim copy 로 고쳐라. empty/parse/shape/graph/anchor/ref/dedup retry 금지 — 본 helper 는 caller 가 camera-only failure 분류한 후에만 호출된다. Output 은 schema 동일 (전체 graph 다시 emit) — LLM 이 camera_decision 만 바꿔서 동일 graph 를 다시 emit 한다. 2026-08-03 prompt-diet 1단계: 말미에 붙이던 ``## Original input bundle (for context)`` + 최초 user prompt 전문을 제거했다. 이 분기는 ``_is_camera_only_failure`` 가 camera 축을 뺀 11개 검증축 전부 통과를 요구하므로 graph/anchor/reference 는 이미 확정돼 있고, 고쳐야 할 카메라 좌표는 아래 candidate 배열 안에서만 고른다 — 원문 번들이 기여할 자리가 없다. 실측 절감 = 그 콜의 user 대부분 (최대 콜 기준 114k 중 원문 재첨부분). """ cam_cands = (geometry or {}).get("camera_cell_candidates_per_unit") or {} look_cands = (geometry or {}).get("look_at_cell_candidates_per_unit") or {} camera_candidates_block = json.dumps( { "camera_cell_candidates_per_unit": cam_cands, "look_at_cell_candidates_per_unit": look_cands, }, sort_keys=True, ) prev_graph_block = json.dumps( previous_parsed.get("graph") or {}, sort_keys=True ) diag_block = "\n".join(f"- {d}" for d in (camera_diagnostics or [])) or "- (none)" return ( "# W20F9 — camera validator repair\n\n" "Your previous response was structurally correct (DAG / anchor / " "references / graph completeness / rationale all passed), but the " "code-side validator rejected one or more " "`camera_decision.camera_cell` / `look_at_cell` entries because " "the coordinate is NOT a member of the supplied candidate set.\n\n" "## Validator diagnostics (camera_in_candidates only)\n\n" f"{diag_block}\n\n" "## Camera coordinate candidates (verbatim copy required)\n\n" f"{camera_candidates_block}\n\n" "## Previous graph you emitted (PRESERVE everything except " "camera_decision.camera_cell / look_at_cell / lens_enum / fov_deg)\n\n" f"{prev_graph_block}\n\n" "## What to emit\n\n" "Re-emit the SAME JSON object matching the schema, with the SAME " "graph node ordering, the SAME `is_dwelling_identity_anchor`, the " "SAME `mode`, the SAME `reference_decision`, the SAME " "`render_guidance`, and the SAME `rationale`. ONLY fix the " "`camera_decision` fields (camera_unit / camera_cell / " "look_at_unit / look_at_cell / lens_enum / fov_deg) so that:\n" "- `camera_cell` is a verbatim integer-pair copy of one entry in " "`camera_cell_candidates_per_unit[camera_unit]`.\n" "- `look_at_cell` is a verbatim integer-pair copy of one entry " "in `look_at_cell_candidates_per_unit[camera_unit]`.\n" "- `lens_enum` and `fov_deg` follow the system prompt rule 6.\n" "Do NOT change any other field. Do NOT invent any new node. Do " "NOT compute or interpolate a coordinate — use ONLY values from " "the candidate arrays above.\n" ) _W20F9_REPAIR_KEYS_MUST_PASS: Tuple[str, ...] = ( "required_shape_ok", "graph_completeness_ok", "node_index_order_ok", "dag_ok", "max_refs_per_bg_ok", "two_refs_distinct_spaces_ok", "same_fp_only_ok", "anchor_exactly_one_ok", "anchor_in_clean_candidate_set_ok", "rationale_and_mode_ok", "synthetic_readback_production_clear", ) def _is_camera_only_failure(validators: Dict[str, Any]) -> bool: """W20F9: retry 가능 조건 = camera_in_candidates 만 fail + 다른 모두 ok.""" if validators.get("camera_in_candidates_ok") is not False: return False for k in _W20F9_REPAIR_KEYS_MUST_PASS: if validators.get(k) is not True: return False return True # W20F10 (2026-07-23 Codex 합의) — graph/anchor-only 실패 분류. # 빈 그래프(nodes==[])는 required_shape/graph_completeness/anchor_exactly_ # one 의 파생 실패만, 비어 있지 않은 그래프는 graph_completeness/anchor_ # exactly_one/anchor_clean 축만 허용 — DAG/ref/same-fp/camera/rationale/ # schema 결함이 섞이면 기존 계약대로 재시도 금지. _W20F10_EXEMPT_EMPTY_NODES: FrozenSet[str] = frozenset({ "required_shape_ok", "graph_completeness_ok", "anchor_exactly_one_ok", }) _W20F10_EXEMPT_NONEMPTY_NODES: FrozenSet[str] = frozenset({ "graph_completeness_ok", "anchor_exactly_one_ok", "anchor_in_clean_candidate_set_ok", }) def _is_graph_anchor_only_failure( validators: Dict[str, Any], *, nodes: List[Any], renderable_bg_ids: FrozenSet[str], clean_anchor_candidate_bg_ids: FrozenSet[str], ) -> bool: """W20F10: 교정 retry 가능 조건 — validator boolean+노드 구조로 판정. diagnostic 문자열 검색 금지(Codex 조건 1). 상류 renderable/clean-anchor 후보가 비어 있거나 **교집합이 0**이면(그래프=renderable 전수 ∧ anchor ∈ clean set 을 동시에 만족할 노드가 존재 불가) LLM 이 복구할 수 없으므로 즉시 fail-closed (Codex TEST-GAP-3 경계). camera_in_ candidates_ok 는 면제 집합에 없으므로 camera 실패가 섞이면 False — W20F9 와 상호 배타. """ if not (renderable_bg_ids & clean_anchor_candidate_bg_ids): return False exempt = ( _W20F10_EXEMPT_EMPTY_NODES if not nodes else _W20F10_EXEMPT_NONEMPTY_NODES ) any_exempt_failed = False for key, value in validators.items(): if not isinstance(value, bool): continue # diagnostics 등 비판정 필드 if key == "all_validators_passed": continue if key in exempt: if value is False: any_exempt_failed = True continue if value is not True: return False return any_exempt_failed def _build_graph_anchor_repair_user_prompt( *, repair_template: str, original_user_prompt: str, previous_content: str, diagnostics: List[str], renderable_bg_ids: FrozenSet[str], clean_anchor_candidate_bg_ids: FrozenSet[str], ) -> str: """W20F10 교정 입력 조립 — 팩 스템({{token}} 치환, 하드 프롬프팅 금지). 치환은 **템플릿 원문 단일 패스** 정규식 callback (Codex 리뷰 NARROW-2: 순차 str.replace 는 먼저 삽입된 데이터(이전 LLM 응답 등) 속의 후속 토큰을 재치환해 '이전 응답 전문' 을 변조한다 — 삽입 값은 다시 스캔되지 않아야 byte-preserved). patch 가 아니라 전체 schema 문서 재반환 계약은 스템이 소유. """ import re diag_block = "\n".join( f"- {d}" for d in (diagnostics or []) ) or "- (none)" mapping = { "diagnostics_block": diag_block, "renderable_bg_ids_block": json.dumps( sorted(renderable_bg_ids), sort_keys=True), "clean_anchor_candidates_block": json.dumps( sorted(clean_anchor_candidate_bg_ids), sort_keys=True), "previous_response_block": previous_content, "original_user_prompt": original_user_prompt, } return re.sub( r"\{\{(" + "|".join(map(re.escape, mapping)) + r")\}\}", lambda m: mapping[m.group(1)], repair_template, ) def _build_response_format( *, sanitized_schema: Dict[str, Any] ) -> Dict[str, Any]: return { "type": "json_schema", "json_schema": { "name": RESPONSE_SCHEMA_NAME, "strict": True, "schema": sanitized_schema, }, } # ───────────────────────────────────────────────────────────────────── # Production-adjacent helper — wired in W20B # ───────────────────────────────────────────────────────────────────── def litellm_shot_aware_bg_render_plan_provider( *, fp_id: str, dossier: Dict[str, Any], geometry: Dict[str, Any], shot_readiness: Dict[str, Any], candidate_catalog: Optional[List[Any]] = None, model: str = PLANNER_MODEL_DEFAULT, timeout_seconds: int = TIMEOUT_SECONDS_DEFAULT, max_completion_tokens: int = MAX_COMPLETION_TOKENS_DEFAULT, ) -> Dict[str, Any]: """Real-LLM planner via a single ``litellm.completion`` call. Returns the parsed graph/output dict ready for ``build_render_plan_for_fp`` consumption. Fail-closed at every preflight joint — the completion call counter stays 0 whenever this function fails before issuing the network call. Flow: 1. Arg-shape guards (no env / network access). 2. Env preflight: ``OPENAI_API_KEY`` non-empty. 3. Prompt pack preflight: load system / user_template / schema; fail-closed on missing / empty / malformed. 4. Lazy ``import litellm``. 5. Litellm capability preflight: ``supports_response_schema(model)`` True AND ``response_format`` listed in ``get_supported_openai_params(model, custom_llm_provider="openai")``. 6. Assemble messages (system + structured user prompt with the JSON-dumped upstream blocks). 7. Single ``litellm.completion`` call. Strict ``json_schema`` response_format using the SANITIZED schema, ``num_retries=0``, no temperature, ``max_completion_tokens`` + ``timeout`` locked to module-level defaults. 8. Response guards: non-empty choices, message present, no truthy refusal, non-empty content, ``finish_reason == "stop"`` exactly. 9. Parse JSON content; reject non-object roots. 10. Local ``jsonschema.Draft202012Validator`` against the ORIGINAL pack schema (keeps minLength / minItems / etc. enforcement). 11. Production ``validate_llm_output`` — DAG / refs / camera / anchor / same-fp / rationale; fail-closed on diagnostics. Returns the parsed dict on full success; raises ``ShotAwareBgRenderPlanProviderError`` on any failure. """ # 1. Arg-shape guards (pre-env, pre-network). if not isinstance(fp_id, str) or not fp_id: raise ShotAwareBgRenderPlanProviderError( f"fp_id must be non-empty string (got {fp_id!r})" ) if not isinstance(dossier, dict): raise ShotAwareBgRenderPlanProviderError( f"dossier must be dict (got {type(dossier).__name__})" ) if not isinstance(geometry, dict): raise ShotAwareBgRenderPlanProviderError( f"geometry must be dict (got {type(geometry).__name__})" ) if not isinstance(shot_readiness, dict): raise ShotAwareBgRenderPlanProviderError( f"shot_readiness must be dict " f"(got {type(shot_readiness).__name__})" ) if candidate_catalog is None: candidate_catalog = [] if not isinstance(candidate_catalog, list): raise ShotAwareBgRenderPlanProviderError( f"candidate_catalog must be list " f"(got {type(candidate_catalog).__name__})" ) # 2. Env preflight (pre-network). from app.core.openai_keys import has_openai_key if not has_openai_key(): raise ShotAwareBgRenderPlanProviderError( "OPENAI_API_KEY missing or empty; refusing to call litellm" ) # 3. Prompt pack preflight. pack = _load_prompt_pack() # 4. Lazy import litellm. try: import litellm # type: ignore from app.core.openai_keys import ( # type: ignore llm_completion as _llm_completion, ) except Exception as exc: # pragma: no cover — env-dependent raise ShotAwareBgRenderPlanProviderError( f"litellm import failed: {type(exc).__name__}: {exc}" ) from exc # 5. Litellm capability preflight (pre-network). try: supports_ok = bool(litellm.supports_response_schema(model=model)) except Exception as exc: raise ShotAwareBgRenderPlanProviderError( f"litellm.supports_response_schema raised: " f"{type(exc).__name__}: {exc}" ) from exc if not supports_ok: raise ShotAwareBgRenderPlanProviderError( f"model {model!r} does not advertise response_schema " f"support in litellm; refusing to call" ) try: supported_params = ( litellm.get_supported_openai_params( model=model, custom_llm_provider="openai" ) or [] ) except Exception as exc: raise ShotAwareBgRenderPlanProviderError( f"litellm.get_supported_openai_params raised: " f"{type(exc).__name__}: {exc}" ) from exc if "response_format" not in supported_params: raise ShotAwareBgRenderPlanProviderError( f"model {model!r} does not list response_format in " f"supported openai params; refusing to call" ) # 6. Assemble request payload. readback_status_raw = geometry.get("readback_status") readback_status = ( readback_status_raw if isinstance(readback_status_raw, str) and readback_status_raw else "unknown" ) user_prompt = _build_user_prompt( user_template=pack["user_template"], fp_id=fp_id, readback_status=readback_status, dossier=dossier, geometry=geometry, shot_readiness=shot_readiness, candidate_catalog=candidate_catalog, ) messages = _build_messages( system_prompt=pack["system_prompt"], user_prompt=user_prompt, ) sanitized_schema = _sanitize_schema_for_openai_strict(pack["schema"]) response_format = _build_response_format( sanitized_schema=sanitized_schema ) # 7. Single completion call. ``num_retries=0`` so a transient # failure does not silently inflate counters; ``temperature`` # omitted (gpt-5 family does not expose tunable temperature and # strict response_schema rejects unknown params on some routes). try: response = _llm_completion( model=model, messages=messages, response_format=response_format, timeout=timeout_seconds, max_completion_tokens=max_completion_tokens, num_retries=0, ) except Exception as exc: raise ShotAwareBgRenderPlanProviderError( f"litellm.completion raised: {type(exc).__name__}: {exc}" ) from exc # 8. Response guards (fail-closed at every joint). choices = getattr(response, "choices", None) or [] if not choices: raise ShotAwareBgRenderPlanProviderError( "response.choices is empty" ) choice = choices[0] msg = getattr(choice, "message", None) if msg is None: raise ShotAwareBgRenderPlanProviderError( "response.choices[0].message is missing" ) refusal = getattr(msg, "refusal", None) if refusal: raise ShotAwareBgRenderPlanProviderError( f"LLM emitted a refusal: {str(refusal)[:200]!r}" ) content = getattr(msg, "content", None) if not content: raise ShotAwareBgRenderPlanProviderError( "response.choices[0].message.content is empty" ) finish_reason = getattr(choice, "finish_reason", None) if finish_reason != "stop": raise ShotAwareBgRenderPlanProviderError( f"finish_reason must be 'stop' exactly (got " f"{finish_reason!r}); length/content_filter fail closed" ) # 9. Parse JSON content. try: parsed = json.loads(content) except (TypeError, ValueError) as exc: raise ShotAwareBgRenderPlanProviderError( f"response content is not valid JSON: " f"{type(exc).__name__}: {exc}" ) from exc if not isinstance(parsed, dict): raise ShotAwareBgRenderPlanProviderError( f"response content is not a JSON object " f"(got {type(parsed).__name__})" ) # 10. Local jsonschema validation against ORIGINAL pack schema. try: import jsonschema except Exception as exc: # pragma: no cover — env-dependent raise ShotAwareBgRenderPlanProviderError( f"jsonschema import failed: {type(exc).__name__}: {exc}" ) from exc try: jsonschema.Draft202012Validator(pack["schema"]).validate(parsed) except jsonschema.ValidationError as exc: raise ShotAwareBgRenderPlanProviderError( f"local jsonschema validation failed: {str(exc.message)[:300]}" ) from exc except Exception as exc: raise ShotAwareBgRenderPlanProviderError( f"local jsonschema setup failed: {type(exc).__name__}: {exc}" ) from exc # 11. Production validators (DAG / refs / camera / etc.). Lazy # import to keep the validator module free of any LLM-coupled # imports in its own right. from app.modules.pipeline.shot_aware_bg_render_plan import ( snap_camera_cells_to_candidates, validate_llm_output, ) same_fp_bg_ids = frozenset( (dossier.get("per_bg_render_facts_by_bg_id") or {}).keys() ) # W20E6-B: graph completeness validates against the staged-shot # renderable subset. Prefer the assemble step's pre-computed list # when present; otherwise derive from per_bg ok flags. renderable_raw = (shot_readiness or {}).get("renderable_bg_ids") if isinstance(renderable_raw, (list, tuple)): renderable_bg_ids = frozenset( b for b in renderable_raw if isinstance(b, str) ) else: renderable_bg_ids = frozenset( bgid for bgid, entry in ( (shot_readiness or {}).get("per_bg") or {} ).items() if isinstance(entry, dict) and entry.get("ok") ) # W20E7-C: clean anchor candidate surface from the dossier. The # anchor's bg_id must be in this set; an empty set fails closed # inside the validator. Always derived (frozenset, possibly empty). clean_anchor_candidate_bg_ids: FrozenSet[str] = frozenset( bid for bid in ( (dossier.get("anchor_selection_metadata") or {}).get( "candidate_bg_ids" ) or [] ) if isinstance(bid, str) and bid ) nodes = (parsed.get("graph") or {}).get("nodes") or [] validators = validate_llm_output( fp_id=fp_id, nodes=nodes, same_fp_bg_ids=same_fp_bg_ids, geometry=geometry, readback_status=readback_status, renderable_bg_ids=renderable_bg_ids, clean_anchor_candidate_bg_ids=clean_anchor_candidate_bg_ids, ) if not validators["all_validators_passed"]: # W20F9 (Codex 2026-05-28) — camera validator-fail targeted retry. # camera_in_candidates 만 fail 이고 다른 검사 전부 통과인 경우에만 # 정확히 1회 retry. empty / parse / shape / graph / anchor / ref / # dedup fail 은 retry 금지. silent repair 금지 — retry 시도 사실 + # 성공/실패 카운터를 parsed dict 에 sentinel 키로 carry, caller 가 # per_fp diagnostics 로 surface. diag_list = validators.get("diagnostics") or [] if _is_camera_only_failure(validators): cam_diags = [ d for d in diag_list if ( "camera_cell" in d or "look_at_cell" in d or "lens_enum" in d or "fov_deg" in d ) ] repair_user_prompt = _build_camera_repair_user_prompt( previous_parsed=parsed, camera_diagnostics=cam_diags, geometry=geometry, ) repair_messages = _build_messages( system_prompt=pack["system_prompt"], user_prompt=repair_user_prompt, ) try: retry_response = _llm_completion( model=model, messages=repair_messages, response_format=response_format, timeout=timeout_seconds, max_completion_tokens=max_completion_tokens, num_retries=0, ) except Exception as exc: raise ShotAwareBgRenderPlanProviderError( f"production validate_llm_output failed (camera-only " f"first pass); W20F9 retry litellm.completion raised: " f"{type(exc).__name__}: {exc}" ) from exc # Reuse the same fail-closed response/parse guards as the # first pass. Differs only in error message prefix so the # retry attempt is auditable. retry_choices = getattr(retry_response, "choices", None) or [] if not retry_choices: raise ShotAwareBgRenderPlanProviderError( "W20F9 retry response.choices is empty" ) retry_choice = retry_choices[0] retry_msg = getattr(retry_choice, "message", None) if retry_msg is None: raise ShotAwareBgRenderPlanProviderError( "W20F9 retry response.choices[0].message is missing" ) retry_refusal = getattr(retry_msg, "refusal", None) if retry_refusal: raise ShotAwareBgRenderPlanProviderError( f"W20F9 retry LLM emitted a refusal: " f"{str(retry_refusal)[:200]!r}" ) retry_content = getattr(retry_msg, "content", None) if not retry_content: raise ShotAwareBgRenderPlanProviderError( "W20F9 retry response.choices[0].message.content is empty" ) retry_finish = getattr(retry_choice, "finish_reason", None) if retry_finish != "stop": raise ShotAwareBgRenderPlanProviderError( f"W20F9 retry finish_reason must be 'stop' (got " f"{retry_finish!r}); length/content_filter fail closed" ) try: retry_parsed = json.loads(retry_content) except (TypeError, ValueError) as exc: raise ShotAwareBgRenderPlanProviderError( f"W20F9 retry content not valid JSON: " f"{type(exc).__name__}: {exc}" ) from exc if not isinstance(retry_parsed, dict): raise ShotAwareBgRenderPlanProviderError( f"W20F9 retry content not a JSON object " f"(got {type(retry_parsed).__name__})" ) try: jsonschema.Draft202012Validator(pack["schema"]).validate(retry_parsed) except jsonschema.ValidationError as exc: raise ShotAwareBgRenderPlanProviderError( f"W20F9 retry local jsonschema validation failed: " f"{str(exc.message)[:300]}" ) from exc retry_nodes = (retry_parsed.get("graph") or {}).get("nodes") or [] retry_validators = validate_llm_output( fp_id=fp_id, nodes=retry_nodes, same_fp_bg_ids=same_fp_bg_ids, geometry=geometry, readback_status=readback_status, renderable_bg_ids=renderable_bg_ids, clean_anchor_candidate_bg_ids=clean_anchor_candidate_bg_ids, ) if not retry_validators["all_validators_passed"]: # W21B final deterministic fallback. If the W20F9 re-prompt # STILL fails on camera candidate membership ONLY (every # other validator passes — synthetic / shape / DAG defects # are excluded by _is_camera_only_failure and never reach # the snap), snap the out-of-candidate camera_cell / # look_at_cell to the nearest valid candidate so a # constrained-choice field can never hard-fail the fp. # The snap masks NO non-camera defect: the gate below is # the same camera-only classifier the W20F9 retry uses. if _is_camera_only_failure(retry_validators): snapped_nodes, snap_repairs = snap_camera_cells_to_candidates( nodes=retry_nodes, geometry=geometry ) if snap_repairs: snapped_validators = validate_llm_output( fp_id=fp_id, nodes=snapped_nodes, same_fp_bg_ids=same_fp_bg_ids, geometry=geometry, readback_status=readback_status, renderable_bg_ids=renderable_bg_ids, clean_anchor_candidate_bg_ids=( clean_anchor_candidate_bg_ids ), ) if snapped_validators["all_validators_passed"]: retry_parsed["graph"]["nodes"] = snapped_nodes retry_parsed["_w20f9_retry_metadata"] = { "camera_validator_retry_attempted": 1, "camera_validator_retry_succeeded": 0, "camera_candidate_snap_repairs": ( snap_repairs[:20] ), "camera_candidate_snap_repairs_total": len( snap_repairs ), "camera_candidate_snap_repairs_truncated": ( len(snap_repairs) > 20 ), "first_pass_camera_diagnostics": cam_diags[:10], } return retry_parsed retry_diag = "; ".join( retry_validators.get("diagnostics") or [] )[:400] raise ShotAwareBgRenderPlanProviderError( f"W20F9 retry production validate_llm_output still " f"failed: {retry_diag}" ) # Success — surface retry metadata to caller via sentinel key # the planner module strips before persisting. retry_parsed["_w20f9_retry_metadata"] = { "camera_validator_retry_attempted": 1, "camera_validator_retry_succeeded": 1, "first_pass_camera_diagnostics": cam_diags[:10], } return retry_parsed if _is_graph_anchor_only_failure( validators, nodes=nodes, renderable_bg_ids=renderable_bg_ids, clean_anchor_candidate_bg_ids=clean_anchor_candidate_bg_ids, ): # W20F10 (2026-07-23 Codex 합의) — graph/anchor validator-fail # targeted retry 정확히 1회 (총 completion 상한=2, W20F9 와 # 상호 배타). 슬라이스 E 실측: 단일 bg 퇴화 fp 에서 무피드백 # 독립 샘플 4회 전부 빈 그래프 — validator diagnostics+기대 # bg/anchor 목록+이전 응답 전문을 교정 블록으로 병기해 재호출. # 교정본은 원 schema+validate_llm_output 전체를 다시 통과해야 # 하며, 재실패(다른 축으로 전환 포함)=연쇄 없이 fail-closed — # 결정론 fallback/snap 금지 (Codex 조건 2·4). repair_user_prompt = _build_graph_anchor_repair_user_prompt( repair_template=pack["graph_anchor_repair_template"], original_user_prompt=user_prompt, previous_content=str(content), diagnostics=list(diag_list), renderable_bg_ids=renderable_bg_ids, clean_anchor_candidate_bg_ids=clean_anchor_candidate_bg_ids, ) repair_messages = _build_messages( system_prompt=pack["system_prompt"], user_prompt=repair_user_prompt, ) try: ga_response = _llm_completion( model=model, messages=repair_messages, response_format=response_format, timeout=timeout_seconds, max_completion_tokens=max_completion_tokens, num_retries=0, ) except Exception as exc: raise _provider_error( f"production validate_llm_output failed (graph/anchor " f"first pass); W20F10 retry litellm.completion raised: " f"{type(exc).__name__}: {exc}", completion_calls=2, ) from exc ga_choices = getattr(ga_response, "choices", None) or [] if not ga_choices: raise _provider_error( "W20F10 retry response.choices is empty", completion_calls=2, ) ga_choice = ga_choices[0] ga_msg = getattr(ga_choice, "message", None) if ga_msg is None: raise _provider_error( "W20F10 retry response.choices[0].message is missing", completion_calls=2, ) ga_refusal = getattr(ga_msg, "refusal", None) if ga_refusal: raise _provider_error( f"W20F10 retry LLM emitted a refusal: " f"{str(ga_refusal)[:200]!r}", completion_calls=2, ) ga_content = getattr(ga_msg, "content", None) if not ga_content: raise _provider_error( "W20F10 retry response.choices[0].message.content is " "empty", completion_calls=2, ) ga_finish = getattr(ga_choice, "finish_reason", None) if ga_finish != "stop": raise _provider_error( f"W20F10 retry finish_reason must be 'stop' (got " f"{ga_finish!r}); length/content_filter fail closed", completion_calls=2, ) try: ga_parsed = json.loads(ga_content) except (TypeError, ValueError) as exc: raise _provider_error( f"W20F10 retry content not valid JSON: " f"{type(exc).__name__}: {exc}", completion_calls=2, ) from exc if not isinstance(ga_parsed, dict): raise _provider_error( f"W20F10 retry content not a JSON object " f"(got {type(ga_parsed).__name__})", completion_calls=2, ) try: jsonschema.Draft202012Validator( pack["schema"] ).validate(ga_parsed) except jsonschema.ValidationError as exc: raise _provider_error( f"W20F10 retry local jsonschema validation failed: " f"{str(exc.message)[:300]}", completion_calls=2, ) from exc ga_nodes = (ga_parsed.get("graph") or {}).get("nodes") or [] ga_validators = validate_llm_output( fp_id=fp_id, nodes=ga_nodes, same_fp_bg_ids=same_fp_bg_ids, geometry=geometry, readback_status=readback_status, renderable_bg_ids=renderable_bg_ids, clean_anchor_candidate_bg_ids=clean_anchor_candidate_bg_ids, ) if not ga_validators["all_validators_passed"]: # 교정본이 camera-only 실패로 바뀌어도 세 번째 호출/snap # 으로 연쇄하지 않는다 — 호출 상한 2 잠금 (Codex 조건 2) ga_diag = "; ".join( ga_validators.get("diagnostics") or [] )[:400] raise _provider_error( f"W20F10 retry production validate_llm_output still " f"failed: {ga_diag}", completion_calls=2, ) ga_parsed["_graph_anchor_retry_metadata"] = { "graph_anchor_validator_retry_attempted": 1, "graph_anchor_validator_retry_succeeded": 1, "first_pass_diagnostics": list(diag_list)[:10], } return ga_parsed diag = "; ".join(diag_list[:5]) raise ShotAwareBgRenderPlanProviderError( f"production validate_llm_output failed: {diag[:400]}" ) return parsed