"""BackgroundPromptStep — Phase 7 Step 5 / Phase 8 v2.

각 master plan의 backgrounds[]에 대해 LLM 1회씩 t2i prompt 생성.
depends_on_bg DAG로 level 병렬화. (이 step은 prompt만 — PNG는 T14 background_render).

Phase 8 v2: ``floor_plan_prompt`` 체크포인트(``data.floor_plans[fp_id]``)에서
``numbered_elements`` + ``camera_recommendations`` 를 로드해 bg_id 기준 flat
매핑을 만들고, ``build_bg_user_prompt`` 에 inject한다. 매핑 키는 master_plan
의 ``backgrounds[].depends_on_fp[0]`` 으로 fp_id 를 결정 (master_plan
invariant 4: 같은 sub_location 은 같은 fp).
"""
from __future__ import annotations

import json
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

from app.core.step_runner import StepRunner
from app.modules.pipeline.background_prompt import (
    PROMPT_VERSION_MAP,
    BackgroundPromptError,
    BackgroundPromptOverlayError,
    build_bg_user_prompt,
    resolve_prompt_version,
    run_background_prompt,
)
from app.modules.pipeline.background_prompt_projection import (
    resolve_projection_plate_injection,
)

logger = logging.getLogger(__name__)
SCHEMA_VERSION = 3  # D6 v6: bg_id pattern → BG_ID_RE + state_label_raw 통합.
# G3.2: v5 prompt — Rule 12 (objects_owned_by_background) 추가. config_hash 가
# invalidate 되어 기존 v4 cp 자동 stale. user 가 force 실행 시점에 v5 LLM 호출.
# D6 v6 (2026-05-09): bg_id pattern → BG_ID_RE + user_template state_label_raw /
# sub_location_label 통합 (D6 raw intent 정합) + runtime enum injection.
# 옛 v5 schema 가 D6 L##B## 거부 → LLM lowercase 출력 → validator bg_id mismatch
# exhaustion BLOCK. SCHEMA_VERSION 2→3 bump + step_manifest 동기 (cp_mismatch
# 안전망) + config_hash invalidate → 기존 cp 자동 rebuild.
# W19B-2 (2026-05-26): default selector "6" 가 가리키는 pack 이름을 backward-
# compat marker 로 유지. 실제 _config_hash 가 stamp 하는 prompt_version 은
# settings.background_prompt_version 의 resolved pack name (default "6" 일 때
# 본 상수와 동일 — 기존 v6 cp 보존).
PROMPT_VERSION = PROMPT_VERSION_MAP["6"]


def _resolved_prompt_version() -> str:
    """settings.background_prompt_version selector 를 on-disk pack name 으로 풀이.

    default selector "6" → ``PROMPT_VERSION_MAP["6"]`` 그대로 → 기존 v6 hash
    byte-identical. opt-in "7" → ``PROMPT_VERSION_MAP["7"]``. floor_plan_prompt
    의 ``_resolved_prompt_version()`` 패턴과 동일.
    """
    from app.core.config import settings
    return resolve_prompt_version(settings.background_prompt_version)


# W21B (2026-06-28): overlay-required prompt versions. v7+ packs consume the
# floor_plan_overlay_payload SOT (marker/base/transient blocks). bgs WITHOUT a
# floor plan (e.g. exterior locations) produce no overlay entry, so a v7+
# selector would fail-closed for them. 이 집합은 그 게이팅 기준이다.
_OVERLAY_REQUIRED_VERSIONS = frozenset({"7", "8", "9", "10", "11", "12"})


def _resolve_effective_bg_version(
    requested_version: str, overlay_present: bool
) -> Tuple[str, Optional[str]]:
    """per-bg effective prompt version 결정 (overlay-missing fallback).

    v7+ pack 은 floor_plan_overlay_payload 입력을 전제로 한다. floor plan 이
    없는 bg(외부/exterior 등)는 overlay entry 가 없어 v7+ 가 fail-closed 된다.
    이때 fail-open 이 아니라 **non-overlay v6 pack 으로 explicit degrade** 하고
    진단 사유를 남긴다 (build_bg_user_prompt 와 run_background_prompt 양쪽에
    동일 effective version 을 넘겨 pack mismatch 0).

    - requested ∈ overlay-required + overlay 존재 → (requested, None) [기존 path 불변]
    - requested ∈ overlay-required + overlay 부재 → ("6", fallback_reason)
    - requested ∉ overlay-required (v6 등) → (requested, None) [byte-identical]
    """
    if requested_version in _OVERLAY_REQUIRED_VERSIONS and not overlay_present:
        return "6", "overlay_payload_missing_for_requested_overlay_prompt"
    return requested_version, None


def _build_fp_lookup_for_bg(
    *,
    fp_prompt_cp: Optional[Dict[str, Any]],
    master_plan_cp: Optional[Dict[str, Any]],
) -> Tuple[Dict[str, List[Dict[str, Any]]], Dict[str, Dict[str, Any]]]:
    """``bg_id`` → (numbered_elements list, camera_recommendation entry) flat 매핑.

    - ``numbered_elements`` 는 fp 단위로 ``fp_prompt_cp.data.floor_plans[fp_id]``
      에 저장되어 있고, 각 bg 는 ``depends_on_fp[0]`` 으로 fp_id 를 가리킨다
      (master_plan invariant 4: 같은 sub_location 은 같은 fp 공유).
    - ``camera_recommendation`` 은 fp 안 list 에서 ``bg_id`` 로 일치하는 단일
      entry 를 lookup. 없으면 entry 자체를 dict 에 넣지 않는다 (caller 가
      ``lookup.get(bg_id)`` 로 None 처리).
    - cp 가 None / data 누락 / fp 미존재 등 어떤 결손에도 안전하게 빈 매핑
      반환 (KeyError 발생 X).
    """
    fp_prompts = (
        ((fp_prompt_cp or {}).get("data", {}) or {}).get("floor_plans", {}) or {}
    )
    plans = (
        ((master_plan_cp or {}).get("data", {}) or {}).get("plans", {}) or {}
    )

    bg_to_fp: Dict[str, str] = {}
    for _gid, gp in plans.items():
        plan = (gp or {}).get("plan") or {}
        for bg in plan.get("backgrounds", []) or []:
            dep = bg.get("depends_on_fp") or []
            bid = bg.get("bg_id")
            if bid and dep:
                bg_to_fp[bid] = dep[0]

    numbered_lookup: Dict[str, List[Dict[str, Any]]] = {}
    camera_lookup: Dict[str, Dict[str, Any]] = {}
    for bg_id, fp_id in bg_to_fp.items():
        fp = fp_prompts.get(fp_id) or {}
        numbered_lookup[bg_id] = list(fp.get("numbered_elements") or [])
        for cam in fp.get("camera_recommendations") or []:
            if cam.get("bg_id") == bg_id:
                camera_lookup[bg_id] = cam
                break
    return numbered_lookup, camera_lookup


def _build_plan_node_lookup(
    plan_cp: Optional[Dict[str, Any]],
) -> Dict[str, Dict[str, Any]]:
    """``{bg_id: plan_node}`` from the shot_aware_bg_render_plan checkpoint (C3).

    Each fp's ``data.per_fp[fp_id].graph.nodes`` carries one node per bg with
    the C3 projection-card pointer fields. bg_id is unique across the dwelling
    graph; a duplicate (defensive) keeps the first seen. Any structural gap
    (cp absent / shape broken) yields an empty mapping → projection path inert.
    """
    lookup: Dict[str, Dict[str, Any]] = {}
    per_fp = ((plan_cp or {}).get("data", {}) or {}).get("per_fp", {}) or {}
    if not isinstance(per_fp, dict):
        return lookup
    for plan in per_fp.values():
        if not isinstance(plan, dict):
            continue
        nodes = ((plan.get("graph") or {}).get("nodes")) or []
        if not isinstance(nodes, list):
            continue
        for node in nodes:
            if not isinstance(node, dict):
                continue
            bg_id = node.get("bg_id")
            if isinstance(bg_id, str) and bg_id and bg_id not in lookup:
                lookup[bg_id] = node
    return lookup


def _build_projection_card_lookup(
    card_cp: Optional[Dict[str, Any]],
) -> Dict[Tuple[str, str], Dict[str, Any]]:
    """``{(bg_id, shot_id): card_entry}`` from the shot_projection_card cp (C2).

    Re-keys ``data.cards`` (keyed ``{bg_id}::{shot_id}``) into the
    ``(bg_id, shot_id)`` tuple the guard re-joins on. Entries without a
    bg_id/shot_id are skipped. The card envelope (``card`` /
    ``bg_plate_visible_description``) and ``card_state`` are passed through
    unchanged — the guard, not this builder, decides injectability.
    """
    lookup: Dict[Tuple[str, str], Dict[str, Any]] = {}
    cards = ((card_cp or {}).get("data", {}) or {}).get("cards", {}) or {}
    if not isinstance(cards, dict):
        return lookup
    for entry in cards.values():
        if not isinstance(entry, dict):
            continue
        bg_id = entry.get("bg_id")
        shot_id = entry.get("shot_id")
        if not bg_id or not shot_id:
            continue
        lookup[(str(bg_id), str(shot_id))] = entry
    return lookup


class BackgroundPromptStep(StepRunner):
    def _config_hash(self) -> str:
        import hashlib
        import json as _json

        from app.core.config import settings

        payload = {
            "background_mode": settings.background_mode,
            "schema_version": SCHEMA_VERSION,
            # W19B-2: default selector "6" 일 때 ``PROMPT_VERSION_MAP["6"]`` 가
            # 기존 PROMPT_VERSION 상수와 byte-identical → 기존 v6 hash 보존.
            # selector "7" 만 별도 hash. selector 자체는 stamp 하지 않고
            # resolved pack name 만 stamp.
            "prompt_version": _resolved_prompt_version(),
        }
        # E2E11 fix①: 휴먼 스케일 앵커 부착은 저작 프롬프트 실질 입력 —
        # ON 시만 스탬프 (OFF byte-identical)
        if getattr(settings, "background_scale_anchor_enabled", False):
            from app.modules.pipeline.background_prompt import (
                HUMAN_SCALE_RULE_PACK,
            )

            payload["background_scale_anchor_pack"] = HUMAN_SCALE_RULE_PACK
        return hashlib.sha256(
            _json.dumps(payload, sort_keys=True).encode("utf-8")
        ).hexdigest()[:16]

    def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict[str, Any]]:
        from app.core.config import settings

        cp = (
            Path(settings.projects_dir)
            / self.project_id
            / "checkpoints"
            / "episodes"
            / self.episode_id
            / step_id
            / "manifest.json"
        )
        if cp.exists():
            try:
                return json.loads(cp.read_text(encoding="utf-8"))
            except Exception as exc:
                logger.warning(
                    "background_prompt: %s parse failed: %s", step_id, exc
                )
        return None

    def _not_applicable(self) -> Dict[str, Any]:
        return {
            "applicable_count": 0,
            "completed_count": 0,
            "failed_count": 0,
            "schema_version": SCHEMA_VERSION,
            "config_hash": self._config_hash(),
            "data": {},
        }

    def _execute(self, mode: str = "resume") -> Dict[str, Any]:
        from app.core.config import settings

        if settings.background_mode not in {"on", "floor_plan_anchored"}:
            return self._not_applicable()

        # W19B-2 / W21B: selector ("6" default | "7"/"8"/"9"/"10" opt-in). unknown selector 는
        # resolve_prompt_version 안에서 ValueError fail-fast → silent latest-pack
        # auto-resolve 방지.
        bg_prompt_selector = settings.background_prompt_version
        resolve_prompt_version(bg_prompt_selector)

        plans_cp = self._load_prev_checkpoint("background_master_plan")
        fp_render_cp = self._load_prev_checkpoint("floor_plan_render")
        # Phase 8 v2: floor_plan_prompt cp 에서 numbered_elements +
        # camera_recommendations 를 로드 (bg_id 별 flat lookup 구성).
        fp_prompt_cp = self._load_prev_checkpoint("floor_plan_prompt")
        # W19B-2 / W21B: v7/v8/v9/v10 selector 일 때 floor_plan_overlay_payload
        # cp 를 load. v6 default path 는 overlay cp 가 없어도 동작 (v5 → W19B-1
        # 미생성 checkpoint 회귀 0). v8 은 v7 와 동일한 overlay payload SOT 를
        # 유지하며 차이는 prompt pack 의 system / user template 뿐이다.
        overlay_cp: Optional[Dict[str, Any]] = None
        if bg_prompt_selector in {"7", "8", "9", "10", "11", "12"}:
            overlay_cp = self._load_prev_checkpoint("floor_plan_overlay_payload")
        # W21B-wave-4 C4 (selector "12" only): consume the projection-card
        # pipeline. The plan cp carries per-bg projection pointers (C3) and the
        # card cp is the authoritative card source (C2). The guard
        # (resolve_projection_plate_injection) re-joins both and decides, per
        # bg, whether a verified anchor plate prose may be injected. All other
        # selectors leave these None → projection path inert.
        plan_node_lookup: Dict[str, Dict[str, Any]] = {}
        card_lookup: Dict[Tuple[str, str], Dict[str, Any]] = {}
        if bg_prompt_selector == "12":
            plan_cp = self._load_prev_checkpoint("shot_aware_bg_render_plan")
            card_cp = self._load_prev_checkpoint("shot_projection_card")
            plan_node_lookup = _build_plan_node_lookup(plan_cp)
            card_lookup = _build_projection_card_lookup(card_cp)
        scene_save_cp = self._load_prev_checkpoint("scene_save")
        shot_validator_cp = self._load_prev_checkpoint("shot_validator")
        shot_selection_cp = self._load_prev_checkpoint("shot_selection")
        rules_cp = self._load_prev_checkpoint("visual_world_rules")
        # Phase 8.1: world_guide cp 추가 — era/costume/style guardrails inject.
        world_guide_cp = self._load_prev_checkpoint("world_guide")

        plans_map = ((plans_cp or {}).get("data", {}) or {}).get("plans", {}) or {}
        fp_paths = {
            fid: entry.get("png_path", "")
            for fid, entry in (
                ((fp_render_cp or {}).get("data", {}) or {}).get("floor_plans", {})
                or {}
            ).items()
            if entry.get("status") == "ok"
        }

        # Reuse the scene-selection helper from BackgroundMasterPlanStep so that
        # group→scene mapping stays consistent with Step 2's master plan.
        from app.core.steps.background_master_plan_step import (
            _select_scenes_for_group,
        )
        from app.modules.llm.llm_client import call_structured
        from app.modules.pipeline._dag_levels import compute_dag_levels

        # 모든 bg job 수집 — prev_shot_only 그룹은 plan=None이므로 자동 skip
        bg_jobs: List[Dict[str, Any]] = []
        items: Dict[str, Dict[str, Any]] = {}
        order: List[str] = []
        for gid, entry in plans_map.items():
            if entry.get("status") != "ok":
                continue
            plan = entry.get("plan") or {}
            for bg in plan.get("backgrounds") or []:
                bid = bg.get("bg_id")
                if not bid:
                    continue
                bg_jobs.append(
                    {
                        "bg_id": bid,
                        "spec": bg,
                        "group_id": gid,
                    }
                )
                deps = bg.get("depends_on_bg") or []
                items[bid] = {"parent_id": deps[0] if deps else ""}
                order.append(bid)

        if not bg_jobs:
            # D6 T5b: empty path 도 hash stamp.
            plans_data_empty = (plans_cp or {}).get("data", {}) or {}
            return {
                "applicable_count": 1,
                "completed_count": 1,
                "failed_count": 0,
                "schema_version": SCHEMA_VERSION,
                "config_hash": self._config_hash(),
                "data": {
                    "backgrounds": {},
                    "consumed_bg_catalog_hash": plans_data_empty.get("bg_catalog_hash", "") or "",
                    "consumed_shot_binding_hash": plans_data_empty.get("shot_binding_hash", "") or "",
                },
            }

        renderable = set(order)
        levels = compute_dag_levels(
            order, items, renderable, parent_field="parent_id"
        )

        # group별 scenes 캐시 (group 내 모든 bg가 동일 scenes 공유)
        scene_segments = (
            ((scene_save_cp or {}).get("data", {}) or {}).get("segments", []) or []
        )
        group_scene_cache: Dict[str, List[Dict[str, Any]]] = {}
        for gid, entry in plans_map.items():
            if entry.get("status") != "ok":
                continue
            plan = entry.get("plan") or {}
            group_member_locs = {
                bg.get("loc_id") for bg in (plan.get("backgrounds") or [])
            }
            synthetic_group = {
                "group_id": gid,
                "members": [
                    {"loc_id": loc, "label": ""}
                    for loc in group_member_locs
                    if loc
                ],
            }
            scenes, _shots = _select_scenes_for_group(
                synthetic_group,
                scene_segments,
                shot_validator_cp,
                shot_selection_cp,
            )
            group_scene_cache[gid] = scenes

        # Phase 8.1: world_guide(풍부한 cultural cues) + visual_world_rules 통합.
        from app.services.visual_context_helper import (
            build_visual_context_block,
            extract_source_language,
        )
        rules_text = build_visual_context_block(
            world_guide_cp=world_guide_cp,
            visual_world_rules_cp=rules_cp,
        )
        # Phase 8.1: source_language 추출 (visual_world_rules v4 또는 patch).
        # 누락이면 Korean 시나리오 기본값 fallback 대신 LLM이 scene_segments에서
        # 자체 추론하도록 빈 문자열로 inject — 시나리오 의존 hardcode 0 유지.
        source_language = extract_source_language(rules_cp) or ""

        # Phase 8 v2: bg_id → (numbered list, camera entry) flat 매핑 구성.
        # depends_on_fp 가 비어 있는 bg(prev_shot_only 등)는 매핑에서 제외되어
        # 자연스럽게 None / [] fallback 으로 흘러간다.
        numbered_lookup, camera_lookup = _build_fp_lookup_for_bg(
            fp_prompt_cp=fp_prompt_cp,
            master_plan_cp=plans_cp,
        )

        # W19B-2 / W21B: v7/v8/v9/v10 path 는 floor_plan_overlay_payload cp 의
        # ``overlays`` dict 를 bg_id 키로 lookup. v6 default path 는
        # overlay_lookup={} 로 두고 build_bg_user_prompt 가 overlay_payload=None
        # 으로 호출됨 (v6 placeholder 에 사용되지 않으므로 영향 0).
        overlay_lookup: Dict[str, Dict[str, Any]] = {}
        if bg_prompt_selector in {"7", "8", "9", "10", "11", "12"}:
            overlay_lookup = (
                ((overlay_cp or {}).get("data", {}) or {}).get("overlays", {})
                or {}
            )

        results: Dict[str, Any] = {}
        failed = 0
        from app.modules.pipeline._workers import resolve_workers
        max_workers = resolve_workers(default=4, cap=8)
        opik = self.build_opik_metadata()
        jobs_by_id = {j["bg_id"]: j for j in bg_jobs}

        def _process(bid: str) -> Tuple[str, Dict[str, Any]]:
            job = jobs_by_id[bid]
            spec = job["spec"]
            # floor_plan PNG 경로 — depends_on_fp[0] 의 fp_path
            fp_first = (spec.get("depends_on_fp") or [""])[0]
            fp_path = fp_paths.get(fp_first, "")
            # prior bg PNG 경로 — 이 step은 prompt만 만들고 PNG는 T14에서 생성하므로
            # path는 placeholder marker로 inject (LLM에 ref 의존성을 알리는 용도).
            prior_bg_paths: List[str] = []
            for dep in spec.get("depends_on_bg") or []:
                prior_entry = results.get(dep) or {}
                if prior_entry.get("status") == "ok":
                    prior_bg_paths.append(f"<bg ref: {dep}>")
            # W21B-wave-4 C4 (selector "12"): per-bg projection-card guard +
            # per-bg effective prompt version. The guard re-joins the plan
            # pointer + card cp; ONLY the fully-checked pass path injects the
            # verified anchor plate prose under the v12 pack. Every other
            # outcome (any fallback_reason / plan node missing) falls back to
            # the verified v10 pack/path so its user_prompt AND its
            # run_background_prompt call are byte-identical to selector "10"
            # (Codex C4 part2 Required — fallback must be real v10, not a v12
            # sentinel surface). plate prose is the only card field that ever
            # reaches the prompt.
            effective_prompt_version = bg_prompt_selector
            projection_plate_prose: Optional[str] = None
            projection_decision: Optional[Dict[str, Any]] = None
            if bg_prompt_selector == "12":
                node = plan_node_lookup.get(bid)
                if isinstance(node, dict):
                    projection_decision = resolve_projection_plate_injection(
                        plan_node=node, card_lookup=card_lookup
                    )
                else:
                    projection_decision = {
                        "inject": False,
                        "fallback_reason": "projection_plan_node_missing",
                    }
                if projection_decision.get("inject"):
                    projection_plate_prose = projection_decision.get("plate_prose")
                    effective_prompt_version = "12"
                else:
                    # All fallbacks (non-pass / stale / missing / leak / no
                    # plan node) → verified v10 pack, no projection surface.
                    effective_prompt_version = "10"
            # W21B (2026-06-28): overlay-missing fallback. floor plan 없는 bg
            # (외부/exterior)는 overlay payload 가 없어 v7+ pack 이 fail-closed
            # 된다 → non-overlay v6 pack 으로 explicit degrade (fail-open 아님).
            # build_bg_user_prompt + run_background_prompt 양쪽이 동일 effective
            # version 을 받으므로 pack mismatch 0. internal bg(overlay 보유)는
            # 분기 미진입 → 기존 v7 path byte-identical.
            requested_prompt_version = effective_prompt_version
            effective_prompt_version, overlay_fallback_reason = (
                _resolve_effective_bg_version(
                    effective_prompt_version, overlay_lookup.get(bid) is not None
                )
            )
            if overlay_fallback_reason:
                logger.info(
                    "background_prompt: bg %s requested v%s but has no overlay "
                    "payload (no floor plan) — degrade to v6 (%s)",
                    bid, requested_prompt_version, overlay_fallback_reason,
                )
            try:
                up = build_bg_user_prompt(
                    bg_spec=spec,
                    floor_plan_path=fp_path,
                    prior_bg_paths=prior_bg_paths,
                    scene_segments=group_scene_cache.get(job["group_id"], []),
                    visual_world_rules=rules_text,
                    # Phase 8 v2: numbered + camera 를 fp_prompt cp 에서 lookup.
                    # 매핑 누락(prev_shot_only / fp 미존재)이면 None 자연 fallback.
                    numbered_elements=numbered_lookup.get(bid),
                    camera_recommendation=camera_lookup.get(bid),
                    # Phase 8.1: source_language 강제 — t2i_prompt 본문 언어 결정.
                    source_language=source_language,
                    # W19B-2 / W21B: per-bg effective version + overlay payload
                    # plumbing. v8/v9/v10/v12 도 v7 와 동일한 overlay SOT 를 사용한다.
                    prompt_version=effective_prompt_version,
                    overlay_payload=(
                        overlay_lookup.get(bid)
                        if effective_prompt_version
                        in {"7", "8", "9", "10", "11", "12"}
                        else None
                    ),
                    # C4: pass-only injection (guard 가 inject=True 일 때만 prose,
                    # 이때만 effective version 이 "12").
                    projection_plate_prose=projection_plate_prose,
                )
            except BackgroundPromptOverlayError as exc:
                # v7 path 의 overlay 누락 / shape 오류 — bg 단위 fail-closed.
                # LLM 호출 진입 전이라 image API / token 비용 0.
                logger.error(
                    "background_prompt: bg %s overlay invalid: %s", bid, exc
                )
                return bid, {
                    "status": "failed",
                    "error": str(exc)[:200],
                    "spec": spec,
                    "group_id": job["group_id"],
                }
            try:
                out = run_background_prompt(
                    user_prompt=up,
                    expected_bg_id=bid,
                    applies_to_shots=spec.get("applies_to_shots") or [],
                    call_structured_fn=call_structured,
                    project_config=self.project_config,
                    opik_metadata=opik,
                    # C4: fallback bg 는 v10 pack(system/schema) 으로 run —
                    # inject pass 만 v12. selector!="12" 는 그대로 selector.
                    prompt_version=effective_prompt_version,
                    # E2E11 fix① (Codex BLOCKING-1 재설계): 휴먼 스케일
                    # 앵커 — 활성 selector(v7 shot-aware 포함) 무변경 부착
                    scale_anchor=bool(
                        getattr(settings, "background_scale_anchor_enabled",
                                False)
                    ),
                )
                entry: Dict[str, Any] = {
                    "status": "ok",
                    "t2i_prompt": out["t2i_prompt"],
                    "ref_guide": out.get("ref_guide", ""),
                    "shot_guides": out.get("shot_guides", []),
                    # G3.2 round 4 critical missed (Task 7.5): owned 를 cp entry
                    # 에 carry — loader 가 읽어 scene_detail prepend block 구성.
                    # validate_bg_prompt_output() 가 in-place normalize 한
                    # strip/dedupe/sort 통과 list.
                    "objects_owned_by_background": out["objects_owned_by_background"],
                    "spec": spec,
                    "group_id": job["group_id"],
                    # W21B (2026-06-28): per-bg version provenance — overlay-missing
                    # fallback 진단. internal bg 는 requested==effective, reason=None.
                    "requested_prompt_version": requested_prompt_version,
                    "effective_prompt_version": effective_prompt_version,
                    "overlay_fallback_reason": overlay_fallback_reason,
                }
                if projection_decision is not None:
                    # C4: per-bg projection verdict for the manifest (visual-review
                    # / Codex audit aid). plate prose is NOT stored here — only the
                    # boolean + reason + anchor pointers (no card content leak).
                    entry["projection_card"] = {
                        "inject": bool(projection_decision.get("inject")),
                        "fallback_reason": projection_decision.get(
                            "fallback_reason", ""
                        ),
                        "source_bg_id": projection_decision.get("source_bg_id", ""),
                        "anchor_shot_id": projection_decision.get(
                            "anchor_shot_id", ""
                        ),
                        "card_id": projection_decision.get("card_id", ""),
                    }
                return bid, entry
            except BackgroundPromptError as exc:
                logger.error(
                    "background_prompt: bg %s failed: %s", bid, exc
                )
                return bid, {
                    "status": "failed",
                    "error": str(exc)[:200],
                    "spec": spec,
                    "group_id": job["group_id"],
                }

        for level in levels:
            if not level:
                continue
            level_workers = max(1, min(max_workers, len(level)))
            with ThreadPoolExecutor(max_workers=level_workers) as pool:
                futures = {pool.submit(_process, bid): bid for bid in level}
                for fut in as_completed(futures):
                    bid = futures[fut]
                    try:
                        _bid, res = fut.result()
                    except Exception as exc:
                        logger.error(
                            "background_prompt: bg %s thread raised: %s",
                            bid,
                            exc,
                        )
                        res = {"status": "failed", "error": str(exc)[:200]}
                    results[bid] = res
                    if res.get("status") != "ok":
                        failed += 1

        # input order 보존
        ordered = {bid: results[bid] for bid in order if bid in results}

        # D6 T5b: master_plan cp 의 catalog + binding hash 양쪽 stamp.
        # background_prompt 는 catalog (각 bg detail) + shot_background_map
        # (어느 shot 에 prompt 적용) 둘 다 의존 → 양쪽 추적 의무.
        plans_data = (plans_cp or {}).get("data", {}) or {}

        # C1-fix (W21B-w5): per-bg 단위로 실제 ok 수 / 전체 수를 보고한다.
        # 이전엔 step 전체를 1 unit 으로 취급하고 completed_count 를
        # ``1 if failed == 0 else 0`` 로 이진화 → bg 1개라도 실패하면 completed=0
        # → step_runner 가 status='failed' (partial 아님) → analysis_dispatch 가
        # 즉시 break → 나머지 정상 bg + 모든 downstream 이 통째로 stop 됐다.
        # 실제 ok/total 보고로 step_runner 가 'partial' 을 도출하게 하여
        # (allow_partial_downstream 기본 True) 정상 bg 와 downstream 이 진행되도록 한다.
        total_bgs = len(ordered)
        completed_bgs = sum(
            1 for e in ordered.values()
            if isinstance(e, dict) and e.get("status") == "ok"
        )

        return {
            "applicable_count": max(1, total_bgs),
            "completed_count": completed_bgs,
            "failed_count": failed,
            "schema_version": SCHEMA_VERSION,
            "config_hash": self._config_hash(),
            "data": {
                "backgrounds": ordered,
                "consumed_bg_catalog_hash": plans_data.get("bg_catalog_hash", "") or "",
                "consumed_shot_binding_hash": plans_data.get("shot_binding_hash", "") or "",
            },
        }