"""Scene image pipeline checkpoint loaders — module-level stateless helpers.

W5 F22 Phase B.11 (2026-04-22): scene_image_service.generate_images()의
checkpoint 로딩 블록(shot_staging / shot_dependency)을 이관.

모든 loader는 stateless 순수 함수로, 체크포인트 JSON을 파싱하여 generate_images
파이프라인이 쓰는 key-to-data map 형태로 정규화한다. 파일이 없거나 파싱에
실패하면 빈 dict를 반환한다 (failure non-fatal).
"""
from __future__ import annotations

import json
import logging
import re
from pathlib import Path
from typing import Any, Dict, List, Optional

from app.core.errors import AppError
from app.core.keep_elements import (
    validate_keep_elements,
)
from app.core.steps._evidence_helpers import _normalize_evidence_fields

__all__ = [
    "load_shot_staging_map",
    "load_shot_dependency_map",
    "load_shot_t2i_variations",
    "load_background_chain_bg_map",
    "load_space_set_bg_map",
    "load_outdoor_direct_context",
]

logger = logging.getLogger(__name__)


def _ep_checkpoint_path(projects_dir: str, project_id: str, episode_id: str, step_name: str) -> Path:
    return (
        Path(projects_dir) / project_id
        / "checkpoints" / "episodes" / episode_id
        / step_name / "manifest.json"
    )


def load_shot_staging_map(
    projects_dir: str, project_id: str, episode_id: str,
) -> Dict[str, Dict[str, Any]]:
    """Load shot_staging checkpoint and index by 'scene_index_shot_index' key.

    Used to resolve character/gaze/camera metadata per shot. Returns empty
    dict on missing file or parse failure (non-fatal).
    """
    staging_map: Dict[str, Dict[str, Any]] = {}
    cp_path = _ep_checkpoint_path(projects_dir, project_id, episode_id, "shot_staging")
    if not cp_path.exists():
        return staging_map
    try:
        raw = json.loads(cp_path.read_text(encoding="utf-8"))
        for st in raw.get("data", raw).get("shots", []):
            key = f"{st.get('scene_index')}_{st.get('shot_index')}"
            staging_map[key] = st
    except Exception as exc:
        logger.warning("shot_staging checkpoint load failed: %s — staging_map 비움", exc)
    return staging_map


def load_shot_dependency_map(
    projects_dir: str, project_id: str, episode_id: str,
) -> Dict[str, Dict[str, Any]]:
    """Load shot_dependency_t2i (preferred) or shot_dependency checkpoint.

    Returns {'scene_index_shot_index': {ref_usage, ignore_elements,
    keep_elements: list[{label, kind}]}} using the first location_ref per shot.
    ignore_elements falls back to legacy 'removal_instruction' key.
    Empty dict on JSON parse failure (warning log).

    Area D-next-min — keep_elements entry validation (L2 intermediate fail-fast):
    - entry must be dict with required keys 'label' and 'kind'.
    - kind must be in KEEP_ELEMENT_KINDS (environment / static_prop only —
      person/character 묘사 금지, character state 는 별도 layer 책임).
    - legacy List[str] cp (v5) → AppError fail-fast (운영자 force 의무).
    - legacy v6 cp (kind=immobilized_character) → AppError fail-fast (운영자
      force 의무 — Area D-next-min 에서 enum 폐기).
    - broad except 가 shape validation AppError 를 absorb 안 함
      (`except AppError: raise` 분기, silent fallback 0).
    """
    dep_map: Dict[str, Dict[str, Any]] = {}
    t2i_path = _ep_checkpoint_path(projects_dir, project_id, episode_id, "shot_dependency_t2i")
    base_path = _ep_checkpoint_path(projects_dir, project_id, episode_id, "shot_dependency")
    source = t2i_path if t2i_path.exists() else base_path
    if not source.exists():
        return dep_map
    try:
        raw = json.loads(source.read_text(encoding="utf-8"))
        for dep in raw.get("data", raw).get("dependencies", []):
            key = f"{dep.get('scene_index')}_{dep.get('shot_index')}"
            for ref in dep.get("location_refs", [])[:1]:
                # Area D-next v3 (Codex I-4 흡수) — keep_elements required.
                # `.get("keep_elements", [])` silent fallback 금지.
                if "keep_elements" not in ref and source is base_path:
                    # ★기본 `shot_dependency`(코드 스텝) CP 에는 T2I 주석(keep_elements)이
                    #  **원래 없다** — 그 칸은 `shot_dependency_t2i`(LLM · still_recipe 판)가
                    #  더한다. t2i CP 가 없어 기본 CP 로 내려온 판(실측 2026-09-02 stage2a ·
                    #  still_recipe off · LP01 이 extra_entities 인 location_ref)에서 v7
                    #  필수를 기본 CP 에 요구하면 fallback 갈래가 통째로 죽는다. 기본 CP 는
                    #  「지시 없음」이 계약이라 빈 목록으로 읽고 로그로 남긴다.
                    #  t2i CP 에서 빠진 것은 아래 그대로 선다(silent fallback 아님).
                    logger.info(
                        "shot_dependency base cp dep[%s]: keep_elements 없음 — t2i 주석 전 "
                        "기본 CP 라 빈 목록으로 읽는다", key)
                    ref = {**ref, "keep_elements": []}
                if "keep_elements" not in ref:
                    raise AppError(
                        code="step.scene_checkpoint_loaders.keep_elements_entry_invalid",
                        message=(
                            f"shot_dependency cp {source.name}: dep[{key}] "
                            f"location_ref missing 'keep_elements' key — schema "
                            f"v7 required field. value={ref!r}"
                        ),
                        status_code=400,
                    )
                raw_keep = ref["keep_elements"]
                validate_keep_elements(
                    raw_keep,
                    label_source=f"shot_dependency cp {source.name}: dep[{key}]",
                    error_code_entry="step.scene_checkpoint_loaders.keep_elements_entry_invalid",
                    error_code_legacy_str="step.scene_checkpoint_loaders.keep_elements_legacy_str",
                )
                dep_map[key] = {
                    "ref_usage": ref.get("ref_usage", ""),
                    "ignore_elements": ref.get("ignore_elements", ref.get("removal_instruction", "")),
                    "keep_elements": raw_keep,
                    # W3 (2026-06-11 fresh full E2E S29 실측): location_ref 가
                    # 명시한 dep 타깃 (scene,shot) 보존 — bytes 공급원(scene_still.
                    # dependent_scene_id = shot_dependency 산출)과 지시문 공급원
                    # (shot_dependency_t2i)이 다른 샷을 가리키면 'SAME FRAME' 지시가
                    # 엉뚱한 프레임에 적용된다 (S29 sh11: bytes=sh4 사진 인서트,
                    # 지시=sh7 zoom → sh4 프레임 복제). consumer(coordinator)가
                    # 이 타깃의 still PNG 를 우선 resolve.
                    "dep_scene_index": ref.get("scene_index"),
                    "dep_shot_index": ref.get("shot_index"),
                    # ★어느 CP 였나 — 기본 CP(코드 스텝)는 ref_usage·keep_elements 주석이
                    #  없다. close×ref_usage matrix 가 이 칸으로 「주석 없음」과 「잘못
                    #  적힘」을 가른다 (실측 2026-09-02 stage2a S1_Shot2).
                    "t2i_annotated": source is t2i_path,
                }
    except AppError:
        # Area D-next — shape validation 위반 은 broad except 에 흡수되면
        # silent fallback. `except AppError: raise` 분기로 caller 까지 raise.
        # provider/JSON parse failure 만 아래 broad except 가 처리.
        raise
    except Exception as exc:
        logger.warning("Failed to parse shot_dependency checkpoint: %s", exc)
    return dep_map


_SHOT_ID_RE = re.compile(r"^S(\d+)_Shot(\d+)$")


def _ingest_phase7_groups_shape(
    bg_map: Dict[str, Dict[str, Any]], data: Dict[str, Any], source_label: str,
) -> int:
    """Phase 7 / Phase 5 공통 shape: data.groups[bg_id].{status, png_path, shot_ids, location_id}.

    각 'ok' 그룹의 shot_ids("Sxx_Shotyy")를 'xx_yy' key로 펼침. 첫 매칭 보존.
    Returns 추가된 매핑 수.
    """
    added = 0
    groups = (data or {}).get("groups") or {}
    if not isinstance(groups, dict):
        return added
    for bg_id, gres in groups.items():
        if not isinstance(gres, dict):
            continue
        if gres.get("status") != "ok":
            continue
        png_path = gres.get("png_path") or ""
        if not png_path:
            continue
        p = Path(png_path)
        if not p.exists():
            continue
        loc_id = gres.get("location_id", "")
        shot_ids = gres.get("shot_ids") or []
        if not isinstance(shot_ids, list):
            continue
        # 한 그룹의 PNG는 모든 shot에 동일하게 inject — 한 번만 read.
        try:
            image_bytes = p.read_bytes()
        except OSError as io_exc:
            logger.warning(
                "%s: %s vanished after exists check — skip: %s", source_label, p, io_exc,
            )
            continue
        for sid in shot_ids:
            m = _SHOT_ID_RE.match(str(sid))
            if not m:
                continue
            key = f"{int(m.group(1))}_{int(m.group(2))}"
            if key in bg_map:
                continue
            bg_map[key] = {
                "image_bytes": image_bytes,
                "label": (
                    f"background chain ref ({bg_id} for {loc_id}) — "
                    f"match wall/floor/ceiling/lighting"
                ),
                # D5 §4.2.1 — caller 가 라벨 파싱 안 하고 직접 read.
                # P1 source-of-truth: groups dict key + gres.location_id.
                "bg_id": bg_id,
                "location_id": loc_id,
            }
            added += 1
    return added


def load_background_chain_bg_map(
    projects_dir: str, project_id: str, episode_id: str,
) -> Dict[str, Dict[str, Any]]:
    """Load background_render(Phase 7) / background_chain_render(Phase 5/4) checkpoint
    and map shot → background image_bytes.

    on_demand step이라 체크포인트가 없을 수 있다. 이 경우 빈 dict 반환 (set_design
    시절과 동일한 optional fallback 패턴).

    P0-2 (2026-04-29): settings.background_chain_enabled=False면 체크포인트가
    있어도 무조건 빈 dict 반환. 토글 off 시 stale 체크포인트가 inject되어 회귀를
    일으키는 것을 차단 (Codex H1).

    지원 shape:
      - Phase 7: background_render — data.groups[bg_id].{status, png_path, shot_ids[], ...}
      - Phase 5: background_chain_render — 동일 shape
      - Phase 4 LEGACY: background_chain_render — data.locations[loc_id].shot_backgrounds[]

    Phase 7 우선, Phase 5/4 fallback. _ingest_phase7_groups_shape는 중복 key
    첫 번째 보존이라 자연스럽게 Phase 7 우선이 된다.

    Returns {'scene_index_shot_index': {image_bytes, label}}. Failed nodes /
    missing PNGs / 잘못된 shot_id 형식은 skip.
    """
    bg_map: Dict[str, Dict[str, Any]] = {}

    from app.core.config import settings
    if not settings.background_chain_enabled:
        return bg_map

    # Phase 7 (background_render) — data.groups shape.
    cp_p7_path = _ep_checkpoint_path(
        projects_dir, project_id, episode_id, "background_render",
    )
    if cp_p7_path.exists():
        try:
            data_p7 = json.loads(cp_p7_path.read_text(encoding="utf-8")).get("data", {})
            n = _ingest_phase7_groups_shape(bg_map, data_p7, "background_render")
            if n:
                logger.info("background_render: loaded %d shot background mappings", n)
        except Exception as exc:
            logger.warning(
                "background_render checkpoint load failed: %s — Phase 5 fallback 시도", exc,
            )

    # Phase 5 / Phase 4 LEGACY (background_chain_render).
    cp_p5_path = _ep_checkpoint_path(
        projects_dir, project_id, episode_id, "background_chain_render",
    )
    if cp_p5_path.exists():
        try:
            data_p5 = json.loads(cp_p5_path.read_text(encoding="utf-8")).get("data", {})
            # Phase 5: data.groups
            n_p5 = _ingest_phase7_groups_shape(bg_map, data_p5, "background_chain_render")
            if n_p5:
                logger.info(
                    "background_chain_render (Phase 5): loaded %d shot background mappings", n_p5,
                )
            # Phase 4 LEGACY: data.locations[].shot_backgrounds[]
            locations = data_p5.get("locations", {}) or {}
            for loc_id, loc_data in locations.items():
                for sb in loc_data.get("shot_backgrounds", []) or []:
                    si = sb.get("scene_index")
                    shi = sb.get("shot_index")
                    img_path = sb.get("image_path", "")
                    if si is None or shi is None or not img_path:
                        continue
                    p = Path(img_path)
                    if not p.exists():
                        continue
                    key = f"{si}_{shi}"
                    if key in bg_map:
                        continue
                    try:
                        image_bytes = p.read_bytes()
                    except OSError as io_exc:
                        logger.warning(
                            "background_chain_render: %s vanished after exists check — skip: %s",
                            p, io_exc,
                        )
                        continue
                    node_id = sb.get("node_id", "")
                    bg_map[key] = {
                        "image_bytes": image_bytes,
                        "label": (
                            f"background chain ref ({node_id} for {loc_id}) — "
                            f"match wall/floor/ceiling/lighting"
                        ),
                        # D5 §4.2.1 — Phase 4 LEGACY 도 동일 contract.
                        # bg_id = node_id (legacy naming, identity 동등).
                        "bg_id": node_id,
                        "location_id": loc_id,
                    }
        except Exception as exc:
            logger.warning(
                "background_chain_render checkpoint load failed: %s — bg_map 일부만 채워짐", exc,
            )

    # W21B Phase 2 (2026-06-10, Codex 합의 B): space_set_bg shot_plate_map overlay —
    # key 충돌 시 space plate 우선 (마지막 update). opt-in flag OFF 면 빈 dict 라
    # 기존 동작 byte-identical. background_chain_enabled gate 는 위 early return 으로
    # 그대로 적용된다 (space plate 도 이 injection point 에선 background ref 의 한 종류).
    space_map = load_space_set_bg_map(projects_dir, project_id, episode_id)
    if space_map:
        bg_map.update(space_map)
        logger.info(
            "space_set_bg: %d shot background mappings overlaid (space plate 우선)",
            len(space_map),
        )

    # W1 (2026-06-11 fresh full E2E 육안 피드백, Codex 합의): space_set_bg 가
    # 해당 shot 을 unassigned/connector_no_plate 로 진단했다면 legacy chain bg
    # fallback 도 억제한다 — space policy 가 'plate 없음/불확실' 이라고 판정한
    # shot 을 legacy exterior establishing 이 덮으면 카메라가 밖에 갇히는 오염
    # (S10 실측: 실내/문앞 장면에 옥상 외부 bg 주입 → 문틈/유리 평면 발명).
    # 억제된 shot 은 coordinator 의 prev_shot/entity-only fallback 으로 진행.
    no_plate_keys = load_space_set_bg_no_plate_keys(projects_dir, project_id, episode_id)
    if no_plate_keys:
        # W1-B (2026-06-11, 재생성 육안 반복으로 정책 정련): 진단 사유별 분기.
        #  - connector_no_plate: 정책 결정(전이부 plate 생략) → legacy bg 도
        #    sentinel 로 대체 + required waiver (전이부에 establishing 주입이
        #    카메라 오염의 근원이었음 — 실측).
        #  - plate_action_no_plate (Phase 3): 배정 LLM 의 명시적 'plate 참조
        #    부적합' 정책 → connector 와 같은 suppression 계열 (sentinel+waiver).
        #  - unassigned(증거 부족) + legacy entry 존재: legacy bg 유지 — bg 를
        #    통째로 빼면 모델이 환경을 '발명'한다(실측: 없는 지붕창 등).
        #    카메라 오염은 W1 의 environment-identity 역할 문구가 차단.
        #  - unassigned + legacy entry 부재: 부착할 것이 없으므로 sentinel
        #    (required waiver 만 — entity/prev_shot fallback 으로 진행).
        suppressed = []
        kept_legacy = []
        for k, reason in no_plate_keys.items():
            existing = bg_map.get(k)
            if existing is not None and existing.get("source") == "space_set_bg":
                continue  # 실 plate 배정이 있으면 진단보다 우선 (방어)
            if reason == "unassigned" and existing is not None:
                kept_legacy.append(k)
                continue
            if existing is not None:
                suppressed.append(k)
            # sentinel entry — image_bytes 없음 = coordinator 가 ref 주입 안 함.
            # suppress_background_required = required background waiver 신호
            # (space_set_bg no-plate policy ↔ render_prompt_card required
            # background 의 runtime reconciliation — validator 약화 아님).
            bg_map[k] = {
                "source": "space_set_bg_no_plate",
                "suppress_background_required": True,
                "reason": reason,
            }
        if suppressed or kept_legacy:
            logger.info(
                "space_set_bg diagnostics: %d legacy chain bg suppressed %s / "
                "%d kept (unassigned + legacy env-identity) %s",
                len(suppressed), sorted(suppressed),
                len(kept_legacy), sorted(kept_legacy),
            )
    return bg_map


def load_space_set_bg_no_plate_keys(
    projects_dir: str, project_id: str, episode_id: str,
) -> Dict[str, str]:
    """space_set_bg shot_assign_diagnostics 에서 'plate 없음/불확실' 판정 shot 맵.

    W1 (2026-06-11): reason ∈ {unassigned, connector_no_plate,
    plate_action_no_plate} 인 shot 의 `{scene}_{shot}` → reason 맵.
    plate_action_no_plate (Phase 3) = 배정 LLM 의 명시적 'plate 참조 부적합'
    정책 판정 — connector 와 같은 suppression 계열 (Codex 리뷰 BLOCKING 1:
    빠지면 legacy chain bg 가 도로 붙어 W1-B 억제 정책과 충돌).
    소비처는 load_background_chain_bg_map —
    이 key 들의 legacy chain bg fallback 을 sentinel 로 대체 (space policy 우선)
    하고, sentinel 이 required background waiver 신호를 coordinator 에 전달
    (Codex 합의: validator 는 pure checker 유지, explicit signal 만).
    flag OFF / checkpoint 부재 / 진단 부재 시 빈 dict (기존 동작 보존).
    """
    keys: Dict[str, str] = {}

    from app.core.config import settings
    if not bool(getattr(settings, "space_set_bg_enabled", False)):
        return keys

    cp_path = _ep_checkpoint_path(projects_dir, project_id, episode_id, "space_set_bg")
    if not cp_path.exists():
        return keys
    try:
        data = json.loads(cp_path.read_text(encoding="utf-8")).get("data", {}) or {}
        for _gid, gres in (data.get("groups") or {}).items():
            if not isinstance(gres, dict):
                continue
            for diag in (gres.get("shot_assign_diagnostics") or []):
                if not isinstance(diag, dict):
                    continue
                if diag.get("reason") not in (
                    "unassigned", "connector_no_plate", "plate_action_no_plate",
                ):
                    continue
                sc, sh = diag.get("scene"), diag.get("shot")
                if sc is None or sh is None:
                    continue
                keys[f"{sc}_{sh}"] = str(diag.get("reason"))
    except Exception as exc:
        logger.warning("space_set_bg no-plate diagnostics load failed: %s — 빈 dict", exc)
        return {}
    return keys


def load_space_set_bg_map(
    projects_dir: str, project_id: str, episode_id: str,
) -> Dict[str, Dict[str, Any]]:
    """Load space_set_bg checkpoint's shot_plate_map → shot 단위 background ref map.

    W21B Phase 2 (2026-06-10): space_set_bg step 이 상류에서 1회 확정한
    shot→space 배정(shot_plate_map)을 deterministic 조인만으로 펼친다 —
    이 loader 는 어떤 판단도 하지 않는다 (Codex 합의 B).

    - opt-in: settings.space_set_bg_enabled=False(default) 면 무조건 빈 dict
      (기존 파이프라인 byte-identical 보존).
    - bg_id 는 synthetic `space_set_bg:<gid>:<plate_key>` — 실 catalog bg_id 를
      가장하지 않는다 (provenance/freshness 계약 오염 방지, Codex C 반려 사유).
    - status!="ok" 그룹 / 누락 PNG 는 skip (해당 shot 은 기존 경로 유지).

    Returns {'scene_index_shot_index': {image_bytes, label, bg_id, source,
    group_id, space}}.
    """
    space_map: Dict[str, Dict[str, Any]] = {}

    from app.core.config import settings
    if not bool(getattr(settings, "space_set_bg_enabled", False)):
        return space_map

    cp_path = _ep_checkpoint_path(projects_dir, project_id, episode_id, "space_set_bg")
    if not cp_path.exists():
        return space_map
    try:
        data = json.loads(cp_path.read_text(encoding="utf-8")).get("data", {}) or {}
        for gid, gres in (data.get("groups") or {}).items():
            if not isinstance(gres, dict) or gres.get("status") != "ok":
                continue
            assets_dir = gres.get("assets_dir") or ""
            for key, entry in (gres.get("shot_plate_map") or {}).items():
                if not isinstance(entry, dict):
                    continue
                png_name = entry.get("plate_png") or ""
                if not png_name or not assets_dir:
                    continue
                p = Path(assets_dir) / png_name
                if not p.exists():
                    continue
                try:
                    image_bytes = p.read_bytes()
                except OSError as io_exc:
                    logger.warning(
                        "space_set_bg: %s vanished after exists check — skip: %s",
                        p, io_exc,
                    )
                    continue
                space = entry.get("space", "")
                space_map[key] = {
                    "image_bytes": image_bytes,
                    "label": (
                        f"space set background ref ({space}, {gid}) — "
                        f"match layout/material/lighting"
                    ),
                    "bg_id": f"space_set_bg:{gid}:{entry.get('plate_key', '')}",
                    "source": "space_set_bg",
                    "group_id": gid,
                    "space": space,
                }
    except Exception as exc:
        logger.warning("space_set_bg checkpoint load failed: %s — space map 비움", exc)
    return space_map


def load_shot_t2i_variations(
    projects_dir: str,
    project_id: str,
    episode_id: str,
    *,
    camera_json: Optional[str],
    scene_index: Optional[int],
    still_index: Optional[int],
    shot_index: Optional[int],
) -> List[Dict[str, Any]]:
    """Load t2i_variations for a single shot: camera_json → scene_detail fallback.

    W5 F22 Phase B.21.1 (2026-04-22): scene_image_service.generate_single_scene_image의
    T2I 변형 조회 블록을 이관. 조회 순서:
      1) still.camera_json의 't2i_variations' 필드 (JSON decode — 원본과 동일하게
         JSONDecodeError는 caller까지 propagate)
      2) 비어있으면 scene_detail 체크포인트에서 scene_index/shot_index 매칭 샷 조회
         - v4: shot_index 매칭 우선 (_shot_index 필드 기반)
         - fallback: scene_index 단독 매칭 (첫 씬 항목의 t2i_variations)

    target_si는 `scene_index or still_index` (원본 literal 그대로 — 0이면 fallback).
    scene_detail 체크포인트 자체의 파싱 실패는 빈 리스트로 fallback (non-fatal).
    """
    t2i_variations = json.loads(camera_json or "{}").get("t2i_variations", [])
    if t2i_variations:
        # G3.1: 옛 cp 4-field 누락 lazy backfill (마킹=legacy).
        for var in t2i_variations:
            _normalize_evidence_fields(
                var, where="scene_checkpoint_loaders.camera_json",
            )
        return t2i_variations

    cp_path = _ep_checkpoint_path(projects_dir, project_id, episode_id, "scene_detail")
    if not cp_path.exists():
        return []
    try:
        detail_data = json.loads(cp_path.read_text(encoding="utf-8")).get("data", {})
    except Exception as exc:
        logger.warning("scene_detail checkpoint parse failed: %s", exc)
        return []

    target_si = scene_index or still_index  # 원본: still.scene_index or still.still_index (0 falsy 보존)
    for sc in detail_data.get("scenes", []):
        if sc.get("scene_index") != target_si:
            continue
        # v4: shot_index 매칭 우선
        if shot_index is not None and sc.get("_shot_index") is not None:
            if sc.get("_shot_index") == shot_index:
                vars_list = sc.get("t2i_variations", [])
                for var in vars_list:
                    _normalize_evidence_fields(
                        var, where="scene_checkpoint_loaders.fallback",
                    )
                return vars_list
        else:
            vars_list = sc.get("t2i_variations", [])
            for var in vars_list:
                _normalize_evidence_fields(
                    var, where="scene_checkpoint_loaders.fallback",
                )
            return vars_list
    return []


def load_outdoor_direct_context(
    projects_dir: str, project_id: str, episode_id: str,
) -> Dict[str, Dict[str, Any]]:
    """W22 야외 직행 합성 컨텍스트 — shot 별 캐논 refs + 9절 프롬프트.

    Returns {"<scene_index>_<shot_index>": {
        "prompt": <s33 9절 결정론 조립 (REF_NOTE 제외 — role 지시문이 담당)>,
        "master_bytes": bytes, "map_bytes": bytes,
        "master_asset_id": str, "map_asset_id": str,
        "place_id": <building group_id>,
    }}

    flag OFF / cp 부재 / 그룹·샷 결측 / 조립 실패 = 해당 키 없음 → 소비자는
    기존 5a~5e 경로로 자연 fallback. flag OFF 면 무조건 빈 dict —
    load_background_chain_bg_map 의 stale 차단 패턴과 동일.
    """
    out: Dict[str, Dict[str, Any]] = {}

    from app.core.config import settings
    if not getattr(settings, "outdoor_direct_compose_enabled", False):
        return out

    def _load_data(step_id: str) -> Dict[str, Any]:
        p = _ep_checkpoint_path(projects_dir, project_id, episode_id, step_id)
        if p.exists():
            try:
                return json.loads(p.read_text(encoding="utf-8")).get("data", {}) or {}
            except Exception as exc:  # noqa: BLE001
                logger.warning("outdoor_direct: %s 로드 실패: %s", step_id, exc)
        return {}

    spec_groups = _load_data("outdoor_place_spec").get("groups", {}) or {}
    canon_groups = _load_data("outdoor_place_canon").get("groups", {}) or {}
    ground_groups = _load_data("outdoor_shot_grounding").get("groups", {}) or {}
    if not (spec_groups and canon_groups and ground_groups):
        return out

    # 샷 원문 (heading/description/characters) — shot_validator cp
    validator = _load_data("shot_validator")
    heading_by_scene: Dict[int, str] = {}
    shot_info: Dict[str, Dict[str, Any]] = {}
    for sc in validator.get("scenes", []) or []:
        si = sc.get("scene_index")
        if si is None:
            continue
        heading_by_scene[int(si)] = sc.get("scene_heading") or ""
        for sh in sc.get("shots", []) or []:
            shi = sh.get("shot_index")
            if shi is None:
                continue
            shot_info[f"{si}_{shi}"] = {
                "description": sh.get("description") or "",
                "characters": [
                    str(c) for c in (sh.get("characters") or []) if c
                ],
            }

    from app.modules.pipeline.outdoor_direct_compose import (
        build_direct_prompt,
        build_place_desc,
        load_blocks,
    )

    DIRECT_PROMPT_VERSION = "3"  # grounded camera (2026-07-11)
    try:
        blocks = load_blocks(DIRECT_PROMPT_VERSION)
    except Exception as exc:  # noqa: BLE001
        logger.warning("outdoor_direct: 프롬프트 팩 로드 실패 — 직행 비활성: %s", exc)
        return out

    # 2026-07-10 실측 fix: 정정 원문을 9절 WORLD FACTS 로 주입하지 않는다.
    # nb2 직행은 언급=묘사라 무관 장소 샷에까지 정정 대상이 그려지는 전역
    # 오염 실측(2회차 E2E). 정정 사실은 스펙 마커 서술명(SPOT 존/앵커)과
    # 캐논 실사(룩 SOT)로 관통 — 원문 재주입은 중복+오염.
    world_facts = ""

    for gid, g_entry in ground_groups.items():
        if not isinstance(g_entry, dict) or "shots" not in g_entry:
            continue  # skipped 그룹
        spec = (spec_groups.get(gid) or {}).get("spec") or {}
        canon = canon_groups.get(gid) or {}
        if not spec or canon.get("status") != "ok":
            continue
        try:
            master_bytes = Path(canon.get("master_png_path") or "").read_bytes()
            map_bytes = Path(canon.get("map_png_path") or "").read_bytes()
        except OSError as exc:
            logger.warning(
                "outdoor_direct: group=%s 캐논 read 실패 — skip: %s", gid, exc
            )
            continue
        codes = [
            it.get("code", "") for it in spec.get("items", []) or []
            if it.get("code")
        ]
        for key, sh_entry in (g_entry.get("shots") or {}).items():
            if not isinstance(sh_entry, dict) or sh_entry.get("status") != "ok":
                continue
            ground = sh_entry.get("ground") or {}
            # v3 (Codex 합의 2026-07-11): grounded camera 는 direct 의 필수
            # 계약 — 필드 결손 시 free_camera 로 조용히 강등하지 않고 해당
            # 샷을 direct 에서 제외(기존 경로 fallback) + 진단 로그.
            if not (ground.get("camera_position_en")
                    and ground.get("look_direction_en")):
                logger.warning(
                    "outdoor_direct: %s grounded camera 필드 결손 — direct "
                    "제외(fail-closed, 기존 경로 fallback)", key,
                )
                continue
            info = shot_info.get(key) or {}
            if not info.get("description"):
                # 샷 원문 없이 직행 프롬프트를 만들면 발명 — 기존 경로 fallback
                logger.warning(
                    "outdoor_direct: %s 샷 원문 없음 — skip", key
                )
                continue
            try:
                si = int(key.split("_")[0])
            except (ValueError, IndexError):
                continue
            try:
                prompt = build_direct_prompt(
                    blocks,
                    place_desc=build_place_desc(ground, spec),
                    scene_heading=heading_by_scene.get(si, ""),
                    shot_description=info["description"],
                    character_names=info.get("characters") or [],
                    moment_context_en=ground.get("moment_context_en") or "",
                    # v3: grounding 카메라 판단을 soft guidance 로 주입
                    camera_guidance={
                        "camera_position_en":
                            ground.get("camera_position_en") or "",
                        "look_direction_en":
                            ground.get("look_direction_en") or "",
                        "in_frame_en": ground.get("in_frame_en") or "",
                    },
                    world_facts_block=world_facts,
                    # style 은 build_final_scene_prompt(translate) 가 주입 — 중복 방지
                    style_context="",
                    spec_codes=codes,
                    # 참조 관할은 labeled_refs 라벨 + resolve_ref_roles 지시문 담당
                    include_ref_note=False,
                )
            except Exception as exc:  # noqa: BLE001
                logger.warning(
                    "outdoor_direct: %s 프롬프트 조립 실패 — skip: %s", key, exc
                )
                continue
            out[key] = {
                "prompt": prompt,
                "prompt_version": DIRECT_PROMPT_VERSION,
                "master_bytes": master_bytes,
                "map_bytes": map_bytes,
                "master_asset_id": canon.get("master_asset_id") or "",
                "map_asset_id": canon.get("map_asset_id") or "",
                "place_id": gid,
            }

    if out:
        logger.info("outdoor_direct: %d shots direct-compose 활성", len(out))
    return out