"""W21B-W8 (2026-06-12): OutdoorSiteLayoutStep — 야외 site layout 좌표 SOT.

order 21.76 (t2i_review 21.72 / zoom_continuity_anchor 21.73 이후, image phase
22 이전). Codex W21B_W8_SITE_LAYOUT_DESIGN_REVIEW=APPROVED_WITH_NARROW_SCOPE.

배경 (ablation 실측, 세션 메모리 21~24절): B-run "가는 방향/멀어짐 구분 못함"
의 결정 변수는 **텍스트 위치 구절** — top-down layout 이미지는 이미지 ref
채널로는 기하 명세 효과가 없다(clay 교훈 동계열). 따라서 좌표는 텍스트 생성의
입력으로만 쓴다 (FP→VLM→T2I 패턴):

  1. seed = selected shot × outdoor(background_classify is_indoor==false)
     × staging character_angles 2인+ (전부 structured 조인 — 단어/regex 의미
     판별 0). zoom continuity 멤버/1인 이동은 제외 + diagnostic.
     outdoor location 별 1 그룹, cap (default 8).
  2. LLM1: location 별 sparse site layout 좌표 emit (씬 원문 전체 — 자르지
     않음). validator 는 **shape/범위만** (의미는 visual canary).
  3. 코드: 카메라-인물 거리/방향/상대크기 deterministic 공간 요약.
  4. LLM2: 멤버 shot 의 t2i variations 에서 위치/깊이/상대크기/카메라거리
     구절만 최소 보정 (entity ID 불변 audit — 위반 variation 폐기 + 진단,
     비차단. edited_spans + no_edit_reason 산출. 카메라워크 언어 금지).
  5. manifest{groups(layouts), prompt_overrides} + review_html (layout 은
     inline SVG — **review 갤러리 전용**, scene 생성 ref 부착 금지. ablation
     실측: 이미지 ref 채널 무력 + 오염 위험. Codex ⓓ).

소비: ``load_outdoor_site_layout_context`` + plan 의 ``merge_prompt_overrides``
— image-phase prompt override 로만 (scene_detail cp 불변, 실사용 프롬프트는
prompt_used, hash provenance 는 이 cp). 우선순위 custom > zoom > site > original.

opt-in: settings.outdoor_site_layout_enabled (default False). OFF 시
not_applicable — default 경로 영향 0. layout/재작문 품질은 deterministic
테스트 비대상 — canary + 육안 gate.

설계: docs/w21b-w8-outdoor-site-layout-production-brief-20260612/.
"""
from __future__ import annotations

import hashlib
import html
import json
import logging
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Set, Tuple

from app.core.framing_scale import FRAMING_CLOSE, FRAMING_INSERT
from app.core.step_runner import StepRunner
from app.modules.pipeline.outdoor_site_layout_plan import (
    DEFAULT_GROUP_CAP,
    PROMPT_SOURCE_SITE,
    SCHEMA_VERSION,
    build_manifest,
    build_shot_spatial_summary,
    camera_missing_shots,
    composition_continuity_chain,
    composition_guide_shot_keys,
    compute_camera_brief,
    detect_site_seeds,
    layout_side_conflicts,
    outdoor_loc_ids,
    revised_prompt_token_violations,
    render_pose_brief_text,
    shared_model_candidate_groups,
    single_shot_complexity_candidates,
    shot_hint_key,
    staged_sides_for_shot,
    validate_site_layout,
    validate_site_manifest,
    zoom_member_shots,
)

logger = logging.getLogger(__name__)

_HTML_SUBDIR_NAME = "review_html"
_GUIDE_SUBDIR_NAME = "composition_guides"


def _sketch_prompt_template() -> str:
    from app.modules.pipeline.outdoor_site_layout_provider import (
        COMPOSITION_SKETCH_PROMPT,
    )
    return COMPOSITION_SKETCH_PROMPT

ShotKey = Tuple[int, int]


def _hash16(text: str) -> str:
    return hashlib.sha256((text or "").encode("utf-8")).hexdigest()[:16]


# ───────────────────── consumer 진입점 (module-level) ─────────────────────


def load_outdoor_site_layout_context(project_id: str, episode_id: str) -> Dict[str, Any]:
    """outdoor site layout consumer 공용 로더 (scene image 생성 경로).

    flag OFF / cp 부재·미완료 / override 0 이면 빈 dict — 소비자 전원 no-op
    (default 경로 byte-identical).

    Returns (비어있지 않으면):
        {"prompt_overrides": {(si, shi): {"group_id", "revised": {int: str},
                                           "provenance": {str: {...hash...}},
                                           "prompt_source": "outdoor_site_layout"}}}
        — zoom ``source_overrides`` entry 와 shape 호환. 소비자는 plan 의
        ``merge_prompt_overrides`` 로 zoom 과 명시적 merge 후 적용한다
        (custom > zoom > site > original).
    """
    from app.core.config import settings

    if not bool(getattr(settings, "outdoor_site_layout_enabled", False)):
        return {}
    cp_path = (
        Path(settings.projects_dir) / project_id / "checkpoints"
        / "episodes" / episode_id / "outdoor_site_layout" / "manifest.json"
    )
    if not cp_path.exists():
        return {}
    try:
        cp = json.loads(cp_path.read_text(encoding="utf-8"))
    except Exception as exc:
        logger.warning("outdoor_site_layout: consumer cp parse 실패: %s", exc)
        return {}
    if cp.get("status") != "completed":
        return {}

    overrides: Dict[ShotKey, Dict[str, Any]] = {}
    for key_s, entry in ((cp.get("data") or {}).get("prompt_overrides") or {}).items():
        try:
            si_s, shi_s = key_s.split(":")
            key = (int(si_s), int(shi_s))
        except (ValueError, AttributeError):
            continue
        revised = {
            int(k): v for k, v in (entry.get("revised") or {}).items()
            if isinstance(v, str) and v.strip()
        }
        if not revised:
            continue
        overrides[key] = {
            "group_id": entry.get("group_id"),
            "revised": revised,
            "provenance": entry.get("provenance") or {},
            "prompt_source": PROMPT_SOURCE_SITE,
        }
    if not overrides:
        return {}
    return {"prompt_overrides": overrides}


# ──────────── composition guide consumer (scene image 생성 경로) ────────────

# scene 생성에 부착되는 가이드 ref 라벨 — composition 전용 (identity/scene ref
# 와 별개 채널, 스케치 모사 금지). 마네킹+Loomis 스케치(재배선) 검증본의 generic
# 라벨 — 인물은 포즈/머리방향을 명시하는 featureless 마네킹이라, 구도뿐 아니라
# 포즈/머리방향까지 재현하고 실제 옷 입은 인물로 치환한다 (실루엣 라벨 supersede:
# 마네킹은 facing 을 신뢰 가능하게 인코딩 — 0d9a0d69 gaze fix 의 '자세 추론 금지'
# 는 방향 정보 0 인 실루엣 전제였다).
COMPOSITION_GUIDE_LABEL = (
    "POSE + COMPOSITION sketch of THIS exact shot (composition guide only): the "
    "human figures are drawn as featureless posed mannequins with constructed "
    "heads ONLY to show each person's exact body pose, stance and head-facing "
    "direction, plus the built features and receding path at their screen "
    "positions and relative sizes. Render the real photographic scene with exactly "
    "this composition, figure placement, relative sizes, body pose and head "
    "direction — but replace each mannequin with a real clothed person (identity, "
    "face, hair and outfit from the prompt text and the character reference "
    "images, never from the mannequin). Do NOT draw any mannequin, joints or "
    "construction lines in the final image, and do NOT copy any environment "
    "detail from this sketch."
)

# guide 부착 샷 한정 bg ref 라벨 완화 — "use as-is" 구도 잠금 해제, 장소
# identity 는 유지 (Codex guard: attach 시점 label override, plate provenance
# 불변).
COMPOSITION_RELAXED_BG_LABEL = (
    "previous shot at same location (SAME PLACE) — place identity reference: "
    "keep this location's materials, architecture, furniture identity, lighting "
    "and weather. Do NOT copy this image's camera position or composition — the "
    "attached composition sketch and the prompt text define this shot's framing. "
    "Do NOT copy any standing/moving people from this image."
)

# W21B-W8 재배선 v2 — 같은 그룹 인접 샷의 '완성 프레임'(prev-shot same-place ref)을
# 연속성 anchor 로 승격할 때의 inline 라벨. 마네킹 스케치를 대체. 장소·인물·broad
# staging 연속성은 이 프레임에서, 이 샷의 카메라/POV/crop 은 본문 SOT (충돌 시 본문
# 우선 — 같은 그룹 샷 고유 프레이밍 보존). prompt_service 가 continuity 지시 재발화.
COMPOSITION_CONTINUITY_ANCHOR_LABEL = (
    "same-place completed frame from an adjacent shot at the same moment — "
    "VISUAL CONTINUITY reference: keep this location's environment, materials, "
    "lighting and weather, and the same people's appearance, and match its broad "
    "staging and where the figures stand and their relative scale for continuity. "
    "This shot's specific camera position, framing, point of view and crop come "
    "from the prompt text, which wins wherever they differ. Do NOT redraw this as "
    "a brand-new composition."
)


def load_composition_guide_context(
    project_id: str, episode_id: str,
) -> Dict[ShotKey, Dict[str, Any]]:
    """composition guide consumer 로더 — {(si, shi): {"mode", ...}}.

    두 flag(outdoor_site_layout_enabled + outdoor_composition_guide_enabled)
    모두 ON + cp completed 시 cp 의 composition_guides 엔트리를 그대로 반환한다.
    option C 엔트리는 mode 별로 모양이 다르다:
      - ``mode='sketch'`` : ``path``(절대경로 보강) + ``forced_character_names`` 등.
      - ``mode='continuity_anchor'`` : ``anchor_source`` + ``forced_character_names``
        (path 없음 — gen 시점에 anchor_source 의 현재-run 완성 프레임을 명시 참조).
    그 외(flag OFF/cp 미완료) 빈 dict = 소비자 no-op (default 경로 byte-identical)."""
    from app.core.config import settings

    if not bool(getattr(settings, "outdoor_site_layout_enabled", False)):
        return {}
    if not bool(getattr(settings, "outdoor_composition_guide_enabled", False)):
        return {}
    cp_dir = (
        Path(settings.projects_dir) / project_id / "checkpoints"
        / "episodes" / episode_id / "outdoor_site_layout"
    )
    cp_path = cp_dir / "manifest.json"
    if not cp_path.exists():
        return {}
    try:
        cp = json.loads(cp_path.read_text(encoding="utf-8"))
    except Exception as exc:
        logger.warning("composition_guide: consumer cp parse 실패: %s", exc)
        return {}
    if cp.get("status") != "completed":
        return {}

    out: Dict[ShotKey, Dict[str, Any]] = {}
    for key_s, entry in ((cp.get("data") or {}).get("composition_guides") or {}).items():
        try:
            si_s, shi_s = key_s.split(":")
            key = (int(si_s), int(shi_s))
        except (ValueError, AttributeError):
            continue
        # 재배선 v2: continuity_anchor marker (sketch path 불요) — admitted 샷 자체가
        # consumer attach 의 신호. 구 sketch 엔트리(path 보유)도 그대로 통과(attach 가
        # path 를 더는 읽지 않음). 절대경로 보강은 path 가 있을 때만(하위호환).
        out_entry = dict(entry)
        rel = entry.get("path")
        if isinstance(rel, str) and rel:
            out_entry["path"] = str(cp_dir / rel)
        out[key] = out_entry
    return out


def load_outdoor_prev_frame_context(
    project_id: str, episode_id: str,
) -> Dict[str, Any]:
    """B (2026-07-02) consumer 로더 — outdoor prev-frame 의무첨부 판정 컨텍스트.

    flag(outdoor_prev_frame_required_enabled) OFF / cp 부재·미완료 → 빈 dict =
    소비자 no-op (default 경로 byte-identical). 반환:
      {"outdoor_loc_sids": set[str L##]  (background_classify is_indoor==false),
       "primary_location_by_scene": {scene_index: "L##"}  (scene_director)}
    구조 필드 조인만 — 이름/라벨 텍스트를 보지 않는다."""
    from app.core.config import settings

    if not bool(getattr(settings, "outdoor_prev_frame_required_enabled", False)):
        return {}
    base = (
        Path(settings.projects_dir) / project_id / "checkpoints"
        / "episodes" / episode_id
    )

    def _load_cp(step_name: str) -> Dict[str, Any]:
        p = base / step_name / "manifest.json"
        if not p.exists():
            return {}
        try:
            cp = json.loads(p.read_text(encoding="utf-8"))
        except Exception as exc:
            logger.warning("outdoor_prev_frame: %s cp parse 실패: %s", step_name, exc)
            return {}
        return cp if cp.get("status") == "completed" else {}

    from app.modules.pipeline.outdoor_site_layout_plan import outdoor_loc_ids
    bc = _load_cp("background_classify")
    outdoor_sids = outdoor_loc_ids(
        (bc.get("data") or {}).get("building_groups") or [])
    sd = _load_cp("scene_director")
    primary_by_scene: Dict[int, str] = {}
    for sc in (sd.get("data") or {}).get("scenes") or []:
        si, pl = sc.get("scene_index"), sc.get("primary_location")
        if isinstance(si, int) and isinstance(pl, str) and pl:
            primary_by_scene[si] = pl
    if not outdoor_sids:
        return {}
    return {
        "outdoor_loc_sids": outdoor_sids,
        "primary_location_by_scene": primary_by_scene,
    }


# 연속성 anchor 로 승격 가능한 prev-shot ref role (zoom crop 계열
# previous_shot_same_frame_zoomed 는 제외 — zoom_continuity 와 충돌).
_CONTINUITY_ANCHOR_ROLES = frozenset({
    "previous_shot_same_room", "previous_shot_continuity",
})

# W-B (2026-07-03) 프레이밍 게이트 — 전신 구도 스케치가 구조적으로 불일치하는
# framing_scale enum (shot_staging SOT — app.core.framing_scale 상수 재사용).
_SKETCH_SKIP_FRAMINGS = frozenset({FRAMING_CLOSE, FRAMING_INSERT})


def _attach_continuity_anchor(
    labeled_refs: List[Any],
    ref_roles: List[str],
    ref_role_metadata: List[Dict[str, Any]],
    attached_meta: List[Any],
    *,
    scene_index: Optional[int],
    shot_index: Optional[int],
    entry: Dict[str, Any],
    source_bytes_resolver: Optional[Callable[[ShotKey], Optional[bytes]]],
) -> bool:
    """continuity_anchor 샷 — ``anchor_source`` 샷의 **현재-run 완성 프레임 bytes**
    (resolver 로 명시 해석)를 prev-shot ref 로 부착/교체한다.

    핵심 (Codex 합의): location_scene_history 단일 슬롯에 기대지 않고 anchor_source
    still 의 현재-run primary bytes 를 **명시 참조** — 기존 prev-shot same-place ref
    의 bytes 를 그 source v2 로 교체(없으면 background_prev_shot 로 append). source
    bytes 부재(아직 미생성) 시 **stale DB primary 로 fallback 하지 않고 no-op** —
    option C 계약은 '현재-run v2 source' 이므로 stale 사용이 곧 버그.
    """
    src = entry.get("anchor_source")
    if not (isinstance(src, (list, tuple)) and len(src) == 2):
        return False
    src_key: ShotKey = (int(src[0]), int(src[1]))
    src_bytes: Optional[bytes] = None
    if source_bytes_resolver is not None:
        try:
            src_bytes = source_bytes_resolver(src_key)
        except Exception as exc:
            logger.warning(
                "composition_guide: S%ssh%s anchor source S%dsh%d bytes resolve "
                "실패 (비차단): %s", scene_index, shot_index, src_key[0],
                src_key[1], exc)
            src_bytes = None
    if not src_bytes:
        logger.info(
            "composition_guide: S%ssh%s continuity anchor no-op — anchor source "
            "S%dsh%d 의 현재-run 완성 프레임 bytes 부재 (stale fallback 안 함)",
            scene_index, shot_index, src_key[0], src_key[1])
        return False

    meta_anchor = {
        "composition_continuity_anchor": True,
        "anchor_source": [src_key[0], src_key[1]],
        "location_id": entry.get("location_id"),
        "group_id": entry.get("group_id"),
    }
    # 기존 prev-shot same-place ref 슬롯의 bytes 를 source v2 로 명시 교체 (in-place
    # — image index 불변). previous_shot_same_frame_zoomed 는 제외(zoom 충돌).
    for i, role in enumerate(ref_roles):
        if role in _CONTINUITY_ANCHOR_ROLES:
            labeled_refs[i] = (COMPOSITION_CONTINUITY_ANCHOR_LABEL, src_bytes)
            ref_role_metadata[i] = {**ref_role_metadata[i], **meta_anchor}
            logger.info(
                "composition_guide: S%ssh%s 연속성 anchor — prev-shot ref(idx=%d "
                "role=%s) bytes 를 source S%dsh%d 현재-run v2 로 명시 교체",
                scene_index, shot_index, i, role, src_key[0], src_key[1])
            return True

    # prev-shot same-place ref 부재 → append. 단 zoom crop ref(previous_shot_same_
    # frame_zoomed)가 있으면 zoom_continuity 와 충돌하므로 append 안 함(no-op) —
    # continuity_anchor 샷은 zoom 멤버에서 제외되므로 실제로는 드문 방어 경로.
    if "previous_shot_same_frame_zoomed" in ref_roles:
        logger.info(
            "composition_guide: S%ssh%s continuity anchor no-op — eligible prev-shot "
            "ref 부재 + zoom crop ref 존재 (zoom 충돌 회피)", scene_index, shot_index)
        return False
    labeled_refs.append((COMPOSITION_CONTINUITY_ANCHOR_LABEL, src_bytes))
    ref_roles.append("previous_shot_same_room")
    ref_role_metadata.append(meta_anchor)
    attached_meta.append(("background_prev_shot", str(entry.get("location_id") or "")))
    logger.info(
        "composition_guide: S%ssh%s 연속성 anchor — prev-shot ref 부재, source "
        "S%dsh%d 현재-run v2 를 background_prev_shot 로 append",
        scene_index, shot_index, src_key[0], src_key[1])
    return True


def attach_composition_guide_ref(
    labeled_refs: List[Any],
    ref_roles: List[str],
    ref_role_metadata: List[Dict[str, Any]],
    attached_meta: List[Any],
    *,
    scene_index: Optional[int],
    shot_index: Optional[int],
    guide_ctx: Dict[ShotKey, Dict[str, Any]],
    source_bytes_resolver: Optional[Callable[[ShotKey], Optional[bytes]]] = None,
    framing: Optional[str] = None,
) -> bool:
    """option C — admitted guide 샷에 mode 별 구도 가이드 부착 (4-list parallel mutate).

    - ``mode='sketch'`` (그룹 첫 샷): 마네킹 구도 스케치 PNG 를 ``composition_guide``
      ref 로 **마지막 1장** append + 같은 샷 same-room bg 라벨 identity-only 완화.
      스케치+본문이 framing SOT (구도/포즈/머리방향).
    - ``mode='continuity_anchor'`` (이후 샷): ``anchor_source`` 샷의 현재-run 완성
      프레임 bytes 를 명시 참조해 prev-shot ref 로 부착/교체 (``_attach_continuity_anchor``;
      stale fallback 금지). 사전 스케치 미부착.

    ``framing`` (W-B 2026-07-03, shot_staging.framing_scale enum SOT): 전신 구도
    스케치는 close/insert 샷과 구조적으로 불일치(전수 육안 (e) 7/16 최다 패턴 —
    S24sh2 스타일 누출의 트리거) — mode='sketch' 한정 attach skip. continuity_anchor
    는 유지(close×zoom/ref_usage 계약 별도 존재). None(구 호출부)이면 게이트 비활성.

    공통: guide 부재/계약 미충족 → False (lists 불변 = byte-identical). worker thread
    안전 — list mutate + 호출자 제공 resolver(메인스레드 path map) 만, 이 함수 자체의
    db 접근 0. 단건/배치 두 경로가 같은 helper 를 쓴다.
    """
    entry = (guide_ctx or {}).get((scene_index, shot_index))
    if not entry:
        return False

    if entry.get("mode") == "continuity_anchor":
        return _attach_continuity_anchor(
            labeled_refs, ref_roles, ref_role_metadata, attached_meta,
            scene_index=scene_index, shot_index=shot_index, entry=entry,
            source_bytes_resolver=source_bytes_resolver)

    # W-B 프레이밍 게이트 (정확 enum 비교, 글자패턴 0) — close류 sketch attach skip.
    if framing in _SKETCH_SKIP_FRAMINGS:
        logger.info(
            "composition_guide: S%ssh%s sketch attach SKIPPED — framing_scale=%r "
            "(close류 샷에 전신 구도 스케치 부착 금지, W-B 게이트)",
            scene_index, shot_index, framing)
        return False

    # mode='sketch' (또는 하위호환 path 보유 엔트리) — 마네킹 스케치 PNG 부착.
    rel = entry.get("path")
    if not rel:
        return False
    path = Path(rel)
    if not path.exists():
        return False
    try:
        png = path.read_bytes()
    except Exception as exc:
        logger.warning(
            "composition_guide: S%ssh%s sketch read 실패 (비차단): %s",
            scene_index, shot_index, exc)
        return False

    # bg ref 라벨 완화 — guide 부착 샷 한정, same-room role 만 (구도 잠금 해제,
    # 장소 identity 유지; 스케치+본문이 framing SOT).
    for i, role in enumerate(ref_roles):
        if role != "previous_shot_same_room":
            continue
        _bytes = labeled_refs[i][1]
        meta = ref_role_metadata[i]
        relaxed = COMPOSITION_RELAXED_BG_LABEL
        keep = meta.get("keep_elements") or []
        keep_labels = [
            e.get("label", "") for e in keep
            if isinstance(e, dict) and e.get("label")
        ]
        if keep_labels:
            relaxed += " Keep: " + ", ".join(keep_labels)
        if meta.get("ignore"):
            relaxed += f" {meta['ignore']}"
        labeled_refs[i] = (relaxed, _bytes)
        ref_role_metadata[i] = {**meta, "composition_relaxed": True}

    labeled_refs.append((COMPOSITION_GUIDE_LABEL, png))
    ref_roles.append("composition_guide")
    ref_role_metadata.append({
        "source": "outdoor_site_layout",
        "location_id": entry.get("location_id"),
        "group_id": entry.get("group_id"),
        # goal#3 (2026-07-01): camera_sketch ImageAsset UUID 를 구조 lineage 로 실어
        # final scene input_image_ids 로 복원(라벨파싱 금지·UUID SOT, indoor_pose_guide
        # 동일 패턴). None 이면 coordinator 가 lineage 에서 제외(엣지 미생성).
        "asset_id": entry.get("guide_asset_id"),
        "pipeline_role": "composition_guide",
    })
    attached_meta.append(("composition_guide", f"{scene_index}:{shot_index}"))
    logger.info(
        "composition_guide: S%ssh%s 구도 스케치 ref 부착 (group=%s loc=%s)",
        scene_index, shot_index, entry.get("group_id"), entry.get("location_id"),
    )
    return True


def _evaluate_guide_judge(
    verdict: Any,
) -> Tuple[bool, Optional[str], str]:
    """의미 게이트 default-deny 평가 → (admit, skip_reason, evidence_quote).

    guide 생성 = is_open_exterior_departure True + confidence∈{medium,high} +
    evidence_quote 존재일 때만. 그 외(거짓/저신뢰/근거없음/판정실패)는 skip —
    가이드 없이 기존 경로 fallback (비차단, failed_count 미증가)."""
    if not isinstance(verdict, dict) or not verdict:
        return False, "judge_failed", ""
    evidence = str(verdict.get("evidence_quote") or "").strip()
    if not verdict.get("is_open_exterior_departure"):
        return False, "judge_not_open_departure", evidence
    if verdict.get("confidence") not in ("medium", "high"):
        return False, "judge_low_confidence", evidence
    if not evidence:
        return False, "judge_missing_evidence", evidence
    return True, None, evidence


def _has_grounded_evidence(evidence: Any) -> bool:
    """evidence-backed 가드 (Codex 코드리뷰 BLOCKING 1) — 배열 non-empty 만으론 부족.
    최소 1개 항목이 dict 이고 shot_key/source_field/quote **모두 strip non-empty** 여야
    근거 있음으로 본다. 빈 문자열로 채운 object 는 LLM 현실적 실패 모드 → deny."""
    if not isinstance(evidence, list):
        return False
    for e in evidence:
        if (isinstance(e, dict)
                and str(e.get("shot_key") or "").strip()
                and str(e.get("source_field") or "").strip()
                and str(e.get("quote") or "").strip()):
            return True
    return False


def _single_shot_target_in_evidence(evidence: Any, target_hint_key: str) -> bool:
    """★Codex NARROW 1 — single-shot judge evidence 가 후보 anchor 샷을 실제 가리키는지.

    evidence 항목 중 shot_key 가 target_hint_key 와 정확히 일치하는 게 1개 이상이면 True.
    single-shot 은 target 이 1개라 wrong-shot evidence 로 attach 되는 사고를 막는다
    (wrong single-shot guide 는 dormant 보다 위험 — Codex)."""
    if not isinstance(evidence, list):
        return False
    return any(
        isinstance(e, dict)
        and str(e.get("shot_key") or "").strip() == target_hint_key
        for e in evidence)


def _evaluate_shared_model_guide_judge(
    verdict: Any,
) -> Tuple[bool, Optional[str]]:
    """shared-model v2 route judge default-deny 평가 → (attach, deny_reason).

    production attach = needs_shared_model_guide True + decision_type∈{cross_shot_
    continuity, single_shot_complexity, both} + confidence∈{medium,high} + **근거 있는**
    evidence (shot_key/source_field/quote 전부 non-empty 1개+, Codex 정렬). no_guide/
    저신뢰/근거없음·빈근거/판정실패 = no attach(diagnostic).

    ★사용자 결정(2026-07-01) — 실외도 실내처럼 단일 복잡샷 lane 개방: single_shot_
    complexity 단독도 attach(LLM 이 구도 복잡도 판단). 어느 샷에 붙일지는 step 의 단일
    후보(single_shot_complexity_candidates, figure>=1 pre-filter)가 결정한다.
    """
    if not isinstance(verdict, dict) or not verdict:
        return False, "judge_failed"
    if not verdict.get("needs_shared_model_guide"):
        return False, "judge_no_need"
    if verdict.get("decision_type") not in (
        "cross_shot_continuity", "single_shot_complexity", "both",
    ):
        return False, "judge_no_guide_decision"
    if verdict.get("confidence") not in ("medium", "high"):
        return False, "judge_low_confidence"
    if not _has_grounded_evidence(verdict.get("evidence")):
        return False, "judge_missing_evidence"
    return True, None


def _staged_character_names(staging: Optional[Dict[str, Any]]) -> Set[str]:
    """staging.character_angles 의 character 이름 집합 (구조 필드 조인만, 글자
    패턴 0). option C 연속성 그룹에서 강제 attach 할 staged character 대상 산출에
    쓴다 — continuity_anchor 샷은 이 집합 ∩ anchor_source staged 집합만 강제한다
    (source 에 없는 인물을 target 에 끌어오지 않음 — Codex 안전 가드)."""
    out: Set[str] = set()
    for ca in (staging or {}).get("character_angles") or []:
        name = ca.get("character")
        if isinstance(name, str) and name:
            out.add(name)
    return out


def _resolve_outdoor_chain_lineage(
    db: Any,
    project_id: str,
    episode_id: str,
    composition_guides: Dict[str, Dict[str, Any]],
) -> Dict[str, Any]:
    """goal#3 — outdoor 항공뷰 체인(aerial_base→shot_blocking→camera_sketch)의
    ``input_image_ids`` 를 **구조키(group_id + producer_stage + shot_index) post-hoc
    resolve** 로 채우고(annotate_generated_asset), camera_sketch ImageAsset UUID 를
    ``composition_guides[g_key]["guide_asset_id"]`` 로 회수한다.

    indoor ``_resolve_guide_asset_ids`` 의 outdoor 버전 — 라벨파싱 금지·UUID SOT.
    producer 는 capture 시점에 parent asset UUID 를 알 수 없다(generation_context
    flush 가 비동기) → 생성 후 DB 가 평탄해진 뒤 구조키로 부모를 회수한다.

    체인 규칙:
      - shot_blocking.input_image_ids = [같은 group_id 의 aerial_base]
      - camera_sketch.input_image_ids = parent_stage 에 따라
          shot_blocking 사용 시 [같은 group_id+shot 의 shot_blocking],
          블로킹 skip(parent_stage=aerial_base) 시 [같은 group_id 의 aerial_base]
      - aerial_base = root(T2I) → input 미설정(None 유지, 엣지 없음)
    재실행 누적으로 parent 후보가 여러 개면 ``child.created_at`` 이하에서 가장 최근
    (created_at <=) 을 택해 같은 run 의 parent 를 가리킨다(created_at = ISO 8601 →
    문자열 비교가 시간순). raise 하지 않음 — db 부재/쿼리 실패는 진단만(비차단).
    """
    diag: Dict[str, Any] = {
        "annotated": [], "unresolved": [],
        "guide_asset_resolved": [], "guide_asset_unresolved": [],
    }
    if db is None:
        return diag
    from app.models.project import ImageAsset
    from app.services.image_capture.annotate import annotate_generated_asset

    try:
        assets = (
            db.query(ImageAsset)
            .filter(
                ImageAsset.project_id == project_id,
                ImageAsset.episode_id == episode_id,
                ImageAsset.pipeline_role.in_([
                    "outdoor_aerial_base", "outdoor_shot_blocking",
                    "outdoor_camera_sketch"]),
                # ★Codex MINOR: indoor resolver 와 일관 — 중간 생성물만(최종물/legacy 제외).
                ImageAsset.is_intermediate.is_(True),
                ImageAsset.asset_type == "generated",
            )
            .order_by(ImageAsset.created_at.asc())
            .all()
        )
    except Exception as exc:  # noqa: BLE001 — 비차단
        logger.warning(
            "outdoor_site_layout: chain lineage 쿼리 실패 (비차단): %s", exc)
        return diag
    if not assets:
        return diag

    def _meta(a: Any) -> Dict[str, Any]:
        try:
            raw = a.pipeline_metadata_json
            return json.loads(raw) if raw else {}
        except Exception:  # noqa: BLE001
            return {}

    # 인덱스: aerial by group_id / blocking by (group_id, shot_index).
    # 값 = (created_at, asset). assets 가 created_at asc 라 list 도 asc 정렬.
    aerial_by_group: Dict[str, List[Tuple[str, Any]]] = {}
    blocking_by_group_shot: Dict[Tuple[str, Any], List[Tuple[str, Any]]] = {}
    for a in assets:
        m = _meta(a)
        gid = str(m.get("group_id") or "")
        if not gid:
            continue
        if a.pipeline_role == "outdoor_aerial_base":
            aerial_by_group.setdefault(gid, []).append((a.created_at, a))
        elif a.pipeline_role == "outdoor_shot_blocking":
            blocking_by_group_shot.setdefault(
                (gid, m.get("shot_index")), []).append((a.created_at, a))

    def _pick_before(cands: List[Tuple[str, Any]], child_created: str) -> Optional[Any]:
        # child.created_at 이하 중 가장 최근. ★Codex NARROW 1: child 보다 늦게 생성된
        # parent 만 있으면 None — future parent 를 붙이지 않는다(wrong edge 가 missing
        # 보다 나쁨, 재실행/부분실패 후 잘못된 체인 방지).
        eligible = [c for c in cands if c[0] <= child_created]
        if not eligible:
            return None
        return max(eligible, key=lambda c: c[0])[1]

    sketch_by_group_shot: Dict[Tuple[str, Any], str] = {}
    changed = False
    for a in assets:
        m = _meta(a)
        gid = str(m.get("group_id") or "")
        if not gid:
            continue
        role = a.pipeline_role
        if role == "outdoor_shot_blocking":
            parent = _pick_before(aerial_by_group.get(gid) or [], a.created_at)
            if parent is not None:
                annotate_generated_asset(
                    a, pipeline_role=role, stage=a.stage,
                    input_image_ids=[parent.id],
                    pipeline_metadata={"lineage_resolved": "structural_key"})
                changed = True
                diag["annotated"].append(
                    {"asset": a.id, "role": role, "input": parent.id})
            else:
                diag["unresolved"].append(
                    {"asset": a.id, "role": role, "reason": "no_aerial_parent"})
        elif role == "outdoor_camera_sketch":
            shi = m.get("shot_index")
            parent = None
            if m.get("parent_stage") == "shot_blocking":
                parent = _pick_before(
                    blocking_by_group_shot.get((gid, shi)) or [], a.created_at)
            if parent is None:
                # 블로킹 skip(parent_stage=aerial_base) 또는 블로킹 미해결 → aerial.
                parent = _pick_before(aerial_by_group.get(gid) or [], a.created_at)
            if parent is not None:
                annotate_generated_asset(
                    a, pipeline_role=role, stage=a.stage,
                    input_image_ids=[parent.id],
                    pipeline_metadata={"lineage_resolved": "structural_key"})
                changed = True
                diag["annotated"].append(
                    {"asset": a.id, "role": role, "input": parent.id})
            else:
                diag["unresolved"].append(
                    {"asset": a.id, "role": role, "reason": "no_parent"})
            # camera_sketch UUID → final scene lineage (group+shot 최신, asc 정렬이라
            # 마지막에 들어온 게 최신).
            sketch_by_group_shot[(gid, shi)] = a.id

    if changed:
        try:
            db.commit()
        except Exception as exc:  # noqa: BLE001
            logger.warning(
                "outdoor_site_layout: chain lineage commit 실패 (비차단): %s", exc)
            try:
                db.rollback()
            except Exception:  # noqa: BLE001
                pass

    # composition_guides[g_key]["guide_asset_id"] 채우기 (mode='sketch' 만 — anchor
    # 후속 샷은 prev-frame 연속성으로 lineage 가 별도). final scene input_image_ids SOT.
    for g_key, entry in composition_guides.items():
        if entry.get("mode") != "sketch":
            continue
        gid = str(entry.get("group_id") or "")
        try:
            _si_s, shi_s = g_key.split(":")
            shi: Any = int(shi_s)
        except (ValueError, AttributeError):
            continue
        aid = sketch_by_group_shot.get((gid, shi))
        if aid:
            entry["guide_asset_id"] = aid
            diag["guide_asset_resolved"].append({"shot": g_key, "asset": aid})
        else:
            diag["guide_asset_unresolved"].append({"shot": g_key, "group_id": gid})
    return diag


class OutdoorSiteLayoutStep(StepRunner):
    """Step 21.76: outdoor site layout 좌표 SOT + 위치 구절 재작문 (W21B-W8)."""

    # Test-only injection slots.
    _layout_override: Optional[Callable[..., Dict[str, Any]]] = None
    _rewrite_override: Optional[Callable[..., Dict[int, Dict[str, Any]]]] = None
    _brief_override: Optional[Callable[..., str]] = None
    _sketch_override: Optional[Callable[..., bytes]] = None
    _judge_override: Optional[Callable[..., Dict[str, Any]]] = None
    # P1(2026-07-01) — Stage C pose brief(구조화 자세/동작) 주입. shared-model v2 전용.
    _pose_brief_override: Optional[Callable[..., Dict[str, Any]]] = None
    # Phase II shared-model v2 — 테스트 주입. base=Stage A(반환 (png, meta)),
    # shared_guide=Stage B+C orchestrator(반환 (png, meta)), shared_judge=route judge.
    _shared_base_override: Optional[Callable[..., Tuple[bytes, Dict[str, Any]]]] = None
    _shared_guide_override: Optional[Callable[..., Tuple[bytes, Dict[str, Any]]]] = None
    _shared_judge_override: Optional[Callable[..., Dict[str, Any]]] = None
    _guide_openai_client: Any = None

    def set_overrides_for_testing(
        self, *, layout=None, rewrite=None, brief=None, sketch=None, judge=None,
        shared_base=None, shared_guide=None, shared_judge=None, pose_brief=None,
    ) -> None:
        self._layout_override = layout
        self._rewrite_override = rewrite
        self._brief_override = brief
        self._sketch_override = sketch
        self._judge_override = judge
        self._pose_brief_override = pose_brief
        self._shared_base_override = shared_base
        self._shared_guide_override = shared_guide
        self._shared_judge_override = shared_judge

    def _resolve_guide_openai_client(self):
        """gpt-image(스케치 ②) 용 OpenAI 클라이언트 (lazy, 다른 image step 패턴)."""
        if self._guide_openai_client is None:
            from app.core.openai_keys import openai_client
            from app.core.config import settings
            self._guide_openai_client = openai_client(
                timeout=float(settings.llm_timeout_image_gen),
            )
        return self._guide_openai_client

    # ─────────────── shared-model v2 (Phase II 재배선) ───────────────

    def _shared_base_cache_key(
        self, *, group_id: str, loc_id: str, layout: Dict[str, Any],
        member_keys: List[ShotKey], model: str, base_prompt_version: str,
        building_fp_id: str = "",
    ) -> str:
        """group 단위 base 캐시 key (Codex 정렬) — group_id + loc + layout_hash +
        base_prompt_version + model + member_set_hash. 같은 그룹은 base 1회 생성·재사용
        (set 일관성). 같은 location 의 다른 group/layout 은 별도 base (보수적 v1).

        W-G: same-building fp ref 로 승격된 base 는 별도 키(``building_fp_id`` 조건부
        포함 — 빈 값이면 payload 불변 = 기존 키와 동일)."""
        payload = {
            "group_id": group_id,
            "location_id": loc_id,
            "layout_hash": _hash16(
                json.dumps(layout, sort_keys=True, ensure_ascii=False)),
            "base_prompt_version": base_prompt_version,
            "model": model,
            "member_set_hash": _hash16(
                json.dumps(sorted([list(k) for k in member_keys]))),
        }
        if building_fp_id:
            payload["building_fp_id"] = building_fp_id
        return _hash16(json.dumps(payload, sort_keys=True))

    def _generate_shared_model_guides(
        self, *, loc_id: str, layout: Dict[str, Any],
        member_keys: List[ShotKey], spatial_summaries: Dict[str, str],
        staging_by_shot: Dict[ShotKey, Dict[str, Any]],
        guide_model: str, judge_model: str, judge_enabled: bool,
        base_fn: Callable[..., Tuple[bytes, Dict[str, Any]]],
        guide_fn: Callable[..., Tuple[bytes, Dict[str, Any]]],
        judge_fn: Callable[..., Dict[str, Any]],
        composition_guides: Dict[str, Dict[str, Any]],
        guide_failed: List[Dict[str, Any]],
        guide_skipped: List[Dict[str, Any]],
        shared_diag: Dict[str, Any],
        base_cache: Dict[str, Tuple[bytes, Dict[str, Any]]],
        seed_meta: Optional[Dict[ShotKey, Dict[str, Any]]] = None,
        t2i_by_shot: Optional[Dict[ShotKey, List[Tuple[int, str]]]] = None,
        shot_descriptions: Optional[Dict[ShotKey, str]] = None,
        pose_brief_fn: Optional[Callable[..., Dict[str, Any]]] = None,
        brief_model: str = "gemini-3.1-pro-preview",
        building_fp: Optional[Dict[str, str]] = None,
    ) -> None:
        """Phase II shared-model v2 (ON 경로). cross-shot continuity candidate → route
        judge → attach 시 group base(Stage A, 캐싱) → anchor 샷 3-stage 가이드(B 조건부
        → C) + 나머지 continuity_anchor. 전부 비차단 — 실패=no-guide+diagnostic
        (old fallback 금지). 부착 mode(sketch/continuity_anchor)·4-list attach 정책은
        OFF 와 동일(consumer 불변). composition_guides/guide_failed/guide_skipped/
        shared_diag/base_cache 를 mutate. judge 는 route 만 — 시각 디자인 미결정."""
        from app.modules.pipeline.outdoor_site_layout_provider import (
            SHARED_MODEL_GUIDE_VERSION,
        )

        candidates = shared_model_candidate_groups(
            layout, member_keys, set(spatial_summaries))
        # ★사용자 결정(2026-07-01): 실외 단일 복잡샷 lane 개방. cross-shot 그룹에 이미
        # 든 샷은 제외하고, 나머지 복잡 단일샷을 후보로 추가(LLM judge 가 최종 판단).
        _covered: set = set()
        for _c in candidates:
            for _k in _c["member_keys"]:
                _covered.add(tuple(_k))
        # 4a fix: broad lane ON 이면 is_complex 구조 게이트 없이 figure>=1 단일샷을
        # 모두 후보화(judge/QC 가 최종 게이트). OFF = 기존 구조 게이트(byte-identical).
        from app.core.config import settings as _settings_bl
        _broad_single = bool(getattr(
            _settings_bl, "outdoor_single_shot_seed_lane_enabled", False))
        single_cands = single_shot_complexity_candidates(
            layout, member_keys, set(spatial_summaries), exclude_keys=_covered,
            broad=_broad_single)
        for cand in candidates + single_cands:
            scene_i = cand["scene_index"]
            per_shot = cand["per_shot"]
            cand_members: List[ShotKey] = [tuple(k) for k in cand["member_keys"]]
            # ★Codex NARROW 2: single-shot 은 cross-shot 과 group_id suffix 로 구분
            # (같은 scene/location 공존 시 충돌 방지 + 캔버스/디버깅 추적).
            is_single = bool(cand.get("is_single_shot"))
            _guide_scope_type = (
                "single_shot_complexity" if is_single else "cross_shot_continuity")
            if is_single:
                _a_si, _a_shi = cand_members[0]
                group_id = f"osl-single-{loc_id.lower()}-s{_a_si}-sh{_a_shi}"
            else:
                group_id = f"osl-shared-{loc_id.lower()}-s{scene_i}"

            shared_diag["candidate_signals"].append({
                "group_id": group_id, "location_id": loc_id, "scene_index": scene_i,
                "member_shots": [f"S{m_si}sh{m_shi}" for m_si, m_shi in cand_members],
                "cameras_non_identical": cand["cameras_non_identical"],
                "landmark_count": cand["landmark_count"],
                "per_shot_signals": per_shot,
            })

            # judge disabled (내부옵션) → candidate-only diagnostic, attach 안 함.
            if not judge_enabled:
                shared_diag["judge_decisions"].append({
                    "group_id": group_id,
                    "decision": "judge_disabled_diagnostic_only"})
                continue

            # route judge 입력 — name-free geography(camera_view_brief) + 구조 staging.
            payload_shots: List[Dict[str, Any]] = []
            for m_si, m_shi in cand_members:
                hk = shot_hint_key(m_si, m_shi)
                cam = next((c for c in layout.get("cameras") or []
                            if c.get("shot_index") == m_shi), None)
                staging = staging_by_shot.get((m_si, m_shi)) or {}
                # 4a fix (Codex): seed prefilter 구조 메타를 judge payload 에 전달 —
                # judge 가 non-frame-visible(primary_location fallback) source 를 risk 로
                # 보고 admit/deny 판단 가능(코드는 이 값으로 의미판정 안 함, 구조 전달만).
                _sm = (seed_meta or {}).get((m_si, m_shi)) or {}
                payload_shots.append({
                    "shot_key": hk,
                    "signals": per_shot.get(hk),
                    "camera_direction": str(staging.get("camera_direction") or ""),
                    "frame_spatial_contract": staging.get("frame_spatial_contract"),
                    "camera_view_brief": (
                        compute_camera_brief(cam, layout) if cam else None),
                    # 4d fix (B', Codex): director 원문 camera/framing directive 를
                    # judge 가 camera_or_framing_risk 근거로 인용할 수 있게 명시 번들로
                    # 전달. ★코드는 이 텍스트를 절대 의미판정하지 않는다(구조 전달만) —
                    # 극단 앵글/벽·바닥 관계 등 spatial ambiguity 판단은 LLM judge 계층만.
                    "raw_director_fields_for_judge": {
                        "camera_direction": str(staging.get("camera_direction") or ""),
                        "framing_scale": staging.get("framing_scale"),
                        "shot_type": staging.get("shot_type"),
                    },
                    # seed prefilter 구조 메타(judge risk 판정 근거).
                    "seed_lane": ("single_shot_complexity" if is_single
                                  else "cross_shot_continuity"),
                    "why_candidate": _sm.get("why_candidate"),
                    "location_source": _sm.get("location_source"),
                    "visible_location_present": _sm.get("location_source") == "frame_visible",
                    "figure_count": _sm.get("figure_count"),
                    "framing_scale": _sm.get("framing_scale"),
                    "character_angle_count": _sm.get("character_angle_count"),
                    "frame_spatial_contract_present": _sm.get("frame_spatial_contract_present"),
                    "available_structural_signal_names": _sm.get(
                        "available_structural_signal_names"),
                })
            group_payload = {
                "location_id": loc_id, "scene_index": scene_i,
                "cameras_non_identical": cand["cameras_non_identical"],
                "landmark_count": cand["landmark_count"], "shots": payload_shots,
            }
            try:
                verdict = judge_fn(
                    group_payload, model=judge_model,
                    log_context={
                        "project_id": self.project_id, "episode_id": self.episode_id,
                        "operation_type": "outdoor_shared_model_guide_judge",
                        "step_name": "outdoor_site_layout",
                    })
            except Exception as exc:
                logger.warning(
                    "outdoor_site_layout: %s S%s shared-model route judge 실패 "
                    "(비차단, skip): %s", loc_id, scene_i, exc)
                shared_diag["judge_decisions"].append({
                    "group_id": group_id, "decision": "judge_failed",
                    "error": str(exc)[:200]})
                guide_skipped.append({"group": group_id, "reason": "judge_failed"})
                continue

            admit, deny_reason = _evaluate_shared_model_guide_judge(verdict)
            v = verdict if isinstance(verdict, dict) else {}
            # ★Codex NARROW 1: single-shot 은 target 이 1개라 judge evidence 가 실제 그 샷을
            # 가리키는지 검증한다(wrong single-shot guide 는 dormant 보다 위험). evidence 의
            # shot_key 중 하나라도 후보 anchor 의 shot_hint_key 와 일치해야 attach.
            if admit and is_single:
                _target_hk = shot_hint_key(*cand_members[0])
                if not _single_shot_target_in_evidence(v.get("evidence"), _target_hk):
                    admit = False
                    deny_reason = "judge_missing_single_shot_target_evidence"
            shared_diag["judge_decisions"].append({
                "group_id": group_id,
                "needs_shared_model_guide": v.get("needs_shared_model_guide"),
                "decision_type": v.get("decision_type"),
                "confidence": v.get("confidence"),
                "reasons": v.get("reasons"),
                "evidence": v.get("evidence"),
                "guide_scope": v.get("guide_scope"),
                "risk_notes": v.get("risk_notes"),
                "admit": admit, "deny_reason": deny_reason,
            })
            if v.get("decision_type") in ("single_shot_complexity", "both"):
                # ★사용자 결정(2026-07-01): single_shot_complexity 도 production attach
                # 대상(실내와 동일 lane). 이 기록은 관측용 — 실제 attach 여부는 admit.
                shared_diag["single_shot_complexity_observation"].append({
                    "group_id": group_id, "decision_type": v.get("decision_type"),
                    "is_single_shot_candidate": bool(cand.get("is_single_shot")),
                    "admit": admit,
                    "note": "single_shot_complexity lane 개방 — admit 시 production attach."})
            if not admit:
                guide_skipped.append({
                    "group": group_id, "reason": deny_reason,
                    "confidence": v.get("confidence")})
                continue

            # ── ATTACH: group base(Stage A, 캐싱) → anchor 3-stage → 나머지 anchor ──
            judge_block = {
                "confidence": v.get("confidence"),
                "decision_type": v.get("decision_type"),
                "evidence": v.get("evidence"), "reasons": v.get("reasons"),
            }
            # W-G: same-building indoor fp — png 존재 시에만 유효(없으면 링크 무시,
            # 기존 T2I 경로 — 비차단). 캐시 키에 fp_id 를 접어 fp 승격 base 와
            # 비승격 base 가 섞이지 않게 한다.
            _bfp_png_bytes: Optional[bytes] = None
            _bfp_id = ""
            if building_fp and building_fp.get("fp_id"):
                try:
                    _bfp_path = Path(str(building_fp.get("png_path") or ""))
                    if _bfp_path.is_file():
                        _bfp_png_bytes = _bfp_path.read_bytes()
                        _bfp_id = str(building_fp["fp_id"])
                except OSError as exc:
                    logger.warning(
                        "outdoor_site_layout: %s building fp %s 읽기 실패 "
                        "(비차단, fp 없이 진행): %s",
                        loc_id, building_fp.get("fp_id"), exc)
            cache_key = self._shared_base_cache_key(
                group_id=group_id, loc_id=loc_id, layout=layout,
                member_keys=cand_members, model=guide_model,
                base_prompt_version=SHARED_MODEL_GUIDE_VERSION,
                building_fp_id=_bfp_id)
            attach_decision: Dict[str, Any] = {
                "group_id": group_id, "scene_index": scene_i,
                "base_cache_key": cache_key, "attached_shots": [], "attached": True,
            }
            # optical/reflective 구도 게이트(2026-07-02): 반사/투과 표면 너머로
            # 피사체를 보는 샷은 마네킹 스케치가 피사체의 실제 위치와 반사상/투과상
            # 위치를 혼동시킨다(가이드가 무-가이드보다 해로움). judge 의
            # evidence-backed per-shot 판정(optical_risk_shot_keys)만 사용 — 코드
            # 텍스트 파싱 0. 해당 샷은 그룹 admit 이어도 엔트리를 만들지 않는다.
            optical_risk_keys = {
                str(k) for k in (v.get("optical_risk_shot_keys") or []) if k}
            # ★Codex 리뷰 NARROW(2026-07-02): 체인 구성 **전에** optical 멤버를
            # 걸러야 한다 — composition_continuity_chain 은 첫 멤버=sketch anchor/
            # 나머지=continuity_anchor(anchor_source=첫 멤버)를 만들므로, 원본
            # 멤버로 체인을 만들고 루프에서 skip 하면 후속 샷의 anchor_source 가
            # 스케치도 현재-run 프레임도 없는 skip 샷을 가리키는 모순이 생긴다.
            eligible_members = []
            for _m in cand_members:
                _mk = shot_hint_key(*_m)
                if _mk in optical_risk_keys:
                    shared_diag.setdefault("optical_risk_excluded", []).append({
                        "shot": f"S{_m[0]}sh{_m[1]}", "group_id": group_id})
                    guide_skipped.append({
                        "group": group_id, "shot": f"S{_m[0]}sh{_m[1]}",
                        "reason": "optical_composition_risk"})
                else:
                    eligible_members.append(_m)
            if not eligible_members:
                attach_decision["attached"] = False
                attach_decision["fail_stage"] = "all_members_optical_risk"
                guide_skipped.append({
                    "group": group_id, "reason": "all_members_optical_risk"})
                shared_diag["attach_decisions"].append(attach_decision)
                continue
            for c in composition_continuity_chain(eligible_members):
                g_si, g_shi = c["key"]
                g_key = shot_hint_key(g_si, g_shi)
                this_staged = _staged_character_names(
                    staging_by_shot.get((g_si, g_shi)))
                if c["mode"] == "sketch":
                    # ── Stage C pose SOT(P1 2026-07-01) — 씬 액션(무절단)에서 인물 자세/
                    #   동작을 structured 추출해 posed-mannequin 스케치의 pose SOT 로 운반.
                    #   ★Codex 가드: t2i(씬 액션)는 Stage C 에서만 사용 — Stage A(항공 base)/
                    #   B(블로킹)는 set/layout 채널로 불변. usable figure 0(추출 실패/저신뢰)
                    #   이면 fail-closed: 이 그룹 가이드 미부착 → 무-가이드 baseline(가이드가
                    #   generic 서있는 마네킹으로 포즈를 오염하던 회귀 재발 방지).
                    g_variations = (t2i_by_shot or {}).get((g_si, g_shi)) or []
                    g_body = (
                        min(g_variations, key=lambda vv: vv[0])[1]
                        if g_variations
                        else (shot_descriptions or {}).get((g_si, g_shi), ""))
                    g_summary = spatial_summaries.get(g_key, "")
                    # 3-stage 필수화(2026-07-02): figure 0(구조/공간 샷)은 마네킹으로
                    # 그릴 인물 자체가 없다 — pose 추출을 걸지 않고 featureless 구도
                    # 스케치(build_blocking_sketch_prompt pose_brief=None 경로)로
                    # 진행한다. pose fail-closed 는 'figure 가 있는데 자세를 못 얻은'
                    # 경우의 오염 가드이므로 figure 0 엔 적용하지 않는다.
                    # ★기준=staging character_angles(텍스트 SOT, detect_site_seeds 와
                    # 동일) — layout 의 entity_count 는 LLM 이 staging 에 없는 figure
                    # 를 발명할 수 있어(2차 canary 실측: staging 0 샷에 entity_count
                    # 1) 기준으로 쓰면 pose 추출(t2i 텍스트 근거 없음)이 항상 실패해
                    # establishing 샷이 전부 fail-closed 로 죽는다.
                    g_has_figures = bool(
                        (staging_by_shot.get((g_si, g_shi)) or {})
                        .get("character_angles") or [])
                    pose_text: Optional[str] = None
                    pose_figs = 0
                    if g_has_figures and g_body and pose_brief_fn is not None:
                        try:
                            pose_res = pose_brief_fn(
                                g_body, g_summary, model=brief_model,
                                log_context={
                                    "project_id": self.project_id,
                                    "episode_id": self.episode_id,
                                    "operation_type": "outdoor_pose_brief",
                                    "step_name": "outdoor_site_layout",
                                })
                            pose_figs = len((pose_res or {}).get("figures") or [])
                            pose_text = render_pose_brief_text(pose_res)
                        except Exception as exc:
                            logger.warning(
                                "outdoor_site_layout: %s S%ssh%s pose brief 실패 "
                                "(비차단, fail-closed): %s", loc_id, g_si, g_shi, exc)
                            pose_text = None
                    if g_has_figures and not pose_text:
                        shared_diag.setdefault("pose_fail_closed", []).append({
                            "shot": f"S{g_si}sh{g_shi}", "group_id": group_id,
                            "figure_count": pose_figs, "had_body": bool(g_body)})
                        guide_skipped.append({
                            "group": group_id, "shot": f"S{g_si}sh{g_shi}",
                            "reason": "pose_brief_unusable_fail_closed"})
                        attach_decision["attached"] = False
                        attach_decision["fail_stage"] = "pose_brief"
                        break

                    # group base — 캐시 miss 면 Stage A 생성(group 1회).
                    base_entry = base_cache.get(cache_key)
                    if base_entry is None:
                        shared_diag["predicted_image_calls"]["base"] += 1
                        try:
                            client = self._resolve_guide_openai_client()
                            # Phase C: Stage A 항공뷰 base 영속화 scope(메인 스레드).
                            # group 공통 → scene_index/shot_index None(group_id metadata
                            # 가 grouping 키, Decision #2 ⓑ/#3). flag OFF 면 이 함수 자체가
                            # 미호출 → scope 미개방(byte-identical).
                            from app.services.image_capture.context import (
                                generation_context,
                            )
                            _base_cap_meta = {
                                "group_id": group_id, "location_id": loc_id,
                                "guide_scope_type": _guide_scope_type,
                                "decision_type": v.get("decision_type")}
                            _base_kwargs: Dict[str, Any] = {}
                            if _bfp_png_bytes is not None:
                                # W-G: same-building fp ref 승격(I2I). kwargs 조건부 —
                                # 링크 없으면 기존 호출 형태 그대로(byte-identical).
                                _base_kwargs["building_fp_png"] = _bfp_png_bytes
                                _base_cap_meta["building_fp_id"] = _bfp_id
                                _base_cap_meta["building_fp_indoor_loc"] = str(
                                    (building_fp or {}).get("indoor_loc_sid") or "")
                            with generation_context(
                                self.project_id, self.episode_id,
                                stage="outdoor_site_layout",
                                scene_index=None, shot_index=None,
                            ):
                                base_png, base_meta = base_fn(
                                    layout, openai_client=client, model=guide_model,
                                    capture_extra_metadata=_base_cap_meta,
                                    **_base_kwargs)
                        except Exception as exc:
                            logger.warning(
                                "outdoor_site_layout: %s S%ssh%s shared-model base "
                                "(Stage A) 생성 실패 (비차단, no-guide): %s",
                                loc_id, g_si, g_shi, exc)
                            guide_failed.append({
                                "shot": f"S{g_si}sh{g_shi}", "reason": str(exc)[:200],
                                "producer_kind": "shared_model_aerial_v2",
                                "stage": "base"})
                            attach_decision["attached"] = False
                            attach_decision["fail_stage"] = "base"
                            break
                        base_entry = (base_png, base_meta)
                        base_cache[cache_key] = base_entry
                        shared_diag["actual_image_calls"]["base"] += 1
                    base_png, base_meta = base_entry

                    # Stage B(블로킹) 필수화 — 3-stage 고정(사용자 정책 2026-07-02):
                    # aerial→blocking(엔티티+카메라 주입)→sketch. blocking 없이 sketch
                    # 직행은 스케일/카메라 앵커가 없어 스케치가 구조물 크기·시점을
                    # 임의 축소/오해한다(skipped_simple 이 만들던 결함). figure 0
                    # 이어도 카메라 아이콘+view-cone 주입은 유효(ent_block "(none)").
                    # shot_blocking_recommended 는 구조 signal 관측용으로만 남는다.
                    use_blocking, blk_reason = True, "mandatory_3stage"

                    shared_diag["predicted_image_calls"]["camera_view_sketch"] += 1
                    if use_blocking:
                        shared_diag["predicted_image_calls"]["blocking"] += 1
                    try:
                        client = self._resolve_guide_openai_client()
                        # Phase C: Stage B(블로킹)+C(스케치) 영속화 scope(메인 스레드).
                        # anchor 샷 → scene_index=g_si/shot_index=g_shi. guide_fn 내부에서
                        # producer_stage/parent_stage(chaining) metadata 부여.
                        from app.services.image_capture.context import (
                            generation_context,
                        )
                        with generation_context(
                            self.project_id, self.episode_id,
                            stage="outdoor_site_layout",
                            scene_index=g_si, shot_index=g_shi,
                        ):
                            png, sm_meta = guide_fn(
                                layout, g_shi, base_png=base_png,
                                use_blocking=use_blocking, openai_client=client,
                                model=guide_model,
                                pose_brief=pose_text,
                                capture_extra_metadata={
                                    "group_id": group_id, "location_id": loc_id,
                                    "guide_scope_type": _guide_scope_type,
                                    "decision_type": v.get("decision_type")})
                        rel = f"{_GUIDE_SUBDIR_NAME}/S{g_si}sh{g_shi}.png"
                        out_path = self._checkpoint_dir() / rel
                        out_path.parent.mkdir(parents=True, exist_ok=True)
                        out_path.write_bytes(png)
                    except Exception as exc:
                        logger.warning(
                            "outdoor_site_layout: %s S%ssh%s shared-model 가이드(B/C) "
                            "생성 실패 (비차단, no-guide): %s", loc_id, g_si, g_shi, exc)
                        guide_failed.append({
                            "shot": f"S{g_si}sh{g_shi}", "reason": str(exc)[:200],
                            "producer_kind": "shared_model_aerial_v2",
                            "stage": "blocking_or_sketch"})
                        attach_decision["attached"] = False
                        attach_decision["fail_stage"] = "blocking_or_sketch"
                        break
                    shared_diag["actual_image_calls"]["camera_view_sketch"] += 1
                    if use_blocking:
                        shared_diag["actual_image_calls"]["blocking"] += 1
                    composition_guides[g_key] = {
                        "mode": "sketch", "path": rel, "location_id": loc_id,
                        "group_id": group_id, "admitted_order": c["admitted_order"],
                        "guide_scope_type": _guide_scope_type,
                        "is_single_shot_candidate": is_single,
                        "model": guide_model, "judge_model": judge_model,
                        "judge": judge_block,
                        "producer_kind": "shared_model_aerial_v2",
                        "guide_generation_status": "ok",
                        "base_cache_key": cache_key,
                        "base_hash": base_meta.get("base_png_hash"),
                        "base_prompt_hash": base_meta.get("base_prompt_hash"),
                        # W-G: base 가 same-building fp ref 로 승격됐는지 (빈 값 =
                        # 비승격 — meta 의 building_fp_used 와 함께 감사 필드).
                        "building_fp_id": (
                            _bfp_id if base_meta.get("building_fp_used") else ""),
                        "blocking_stage": ("used" if use_blocking else blk_reason),
                        "blocking_stage_reason": blk_reason,
                        "blocking_hash": sm_meta.get("blocking_png_hash"),
                        # P1: Stage C pose SOT — figure 있는 샷은 pose_text 통과 시에만
                        # 도달(fail-closed). figure 0 샷은 featureless 경로라 False.
                        "pose_brief_present": bool(pose_text),
                        "pose_brief_figures": pose_figs,
                        "pose_brief_hash": sm_meta.get("pose_brief_hash"),
                        "camera_brief_hash": sm_meta.get("camera_brief_hash"),
                        "camera_view_sketch_hash": sm_meta.get(
                            "camera_view_sketch_hash"),
                        "sketch_prompt_hash": sm_meta.get("sketch_prompt_hash"),
                        "helper_version": sm_meta.get("helper_version"),
                        "forced_character_names": sorted(this_staged),
                    }
                    attach_decision["attached_shots"].append(g_key)
                else:
                    # 이후 샷 — continuity_anchor (사전 스케치 미생성, OFF 와 동일 계약).
                    psi, pshi = c["anchor_source"]
                    source_staged = _staged_character_names(
                        staging_by_shot.get((psi, pshi)))
                    composition_guides[g_key] = {
                        "mode": "continuity_anchor", "anchor_source": [psi, pshi],
                        "location_id": loc_id, "group_id": group_id,
                        "admitted_order": c["admitted_order"],
                        "guide_scope_type": _guide_scope_type,
                        "is_single_shot_candidate": is_single,
                        "judge_model": judge_model, "judge": judge_block,
                        "forced_character_names": sorted(this_staged & source_staged),
                    }
                    attach_decision["attached_shots"].append(g_key)
            shared_diag["attach_decisions"].append(attach_decision)

    # ───────────────────── plumbing ─────────────────────

    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("outdoor_site_layout: %s parse failed: %s", step_id, exc)
        return None

    def _checkpoint_dir(self) -> Path:
        from app.core.config import settings
        return (
            Path(settings.projects_dir) / self.project_id / "checkpoints"
            / "episodes" / self.episode_id / "outdoor_site_layout"
        )

    def _config_hash(self) -> str:
        from app.core.config import settings
        from app.modules.pipeline.outdoor_site_layout_provider import (
            PROMPT_VERSION,
            PROVIDER_VERSION,
            SHARED_MODEL_GUIDE_VERSION,
        )
        payload = {
            "enabled": bool(getattr(settings, "outdoor_site_layout_enabled", False)),
            "group_cap": int(getattr(
                settings, "outdoor_site_layout_group_cap", DEFAULT_GROUP_CAP)),
            "schema_version": SCHEMA_VERSION,
            "prompt_version": PROMPT_VERSION,
            "provider_version": PROVIDER_VERSION,
            "composition_guide_enabled": bool(getattr(
                settings, "outdoor_composition_guide_enabled", False)),
            "composition_guide_model": str(getattr(
                settings, "outdoor_composition_guide_model", "")),
            "composition_brief_model": str(getattr(
                settings, "outdoor_composition_brief_model", "")),
            "composition_judge_model": str(getattr(
                settings, "outdoor_composition_judge_model", "")),
        }
        # Phase II shared-model v2 — producer flag/helper version + judge enable 은 ON
        # 일 때만 config_hash 에 접는다 (OFF 면 payload 불변 = 기존 체크포인트
        # byte-identical, 불필요한 재실행 방지).
        if bool(getattr(
                settings, "outdoor_composition_guide_shared_model_enabled", False)):
            payload["composition_guide_shared_model_enabled"] = True
            payload["shared_model_guide_version"] = SHARED_MODEL_GUIDE_VERSION
            payload["shared_model_guide_judge_enabled"] = bool(getattr(
                settings, "outdoor_shared_model_guide_judge_enabled", True))
        # W-G (2026-07-03) — same-building fp ref 는 Stage A base 의 생성 입력
        # (T2I→I2I+정합 지시)을 실질 변경 → flag=True 일 때만 스탬프(shared-model
        # 정책 미러, OFF = payload 불변 = 기존 체크포인트 byte-identical).
        if bool(getattr(settings, "outdoor_building_fp_ref_enabled", False)):
            payload["outdoor_building_fp_ref_enabled"] = True
        return hashlib.sha256(
            json.dumps(payload, sort_keys=True).encode("utf-8")
        ).hexdigest()[:16]

    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": {},
        }

    # ───────────────────── input loaders ─────────────────────

    @staticmethod
    def _scene_texts(save_cp: Optional[Dict[str, Any]]) -> Dict[int, str]:
        out: Dict[int, str] = {}
        for seg in (save_cp or {}).get("data", {}).get("segments", []) or []:
            si = seg.get("scene_index")
            if isinstance(si, int):
                out[si] = seg.get("text") or ""
        return out

    @staticmethod
    def _shot_descriptions(shot_cp: Optional[Dict[str, Any]]) -> Dict[ShotKey, str]:
        out: Dict[ShotKey, str] = {}
        for sc in (shot_cp or {}).get("data", {}).get("scenes", []) or []:
            si = sc.get("scene_index")
            for sh in sc.get("shots", []) or []:
                shi = sh.get("shot_index")
                if isinstance(si, int) and isinstance(shi, int):
                    out[(si, shi)] = sh.get("description") or ""
        return out

    @staticmethod
    def _staging_by_shot(staging_cp: Optional[Dict[str, Any]]) -> Dict[ShotKey, Dict[str, Any]]:
        out: Dict[ShotKey, Dict[str, Any]] = {}
        for sh in (staging_cp or {}).get("data", {}).get("shots", []) or []:
            si, shi = sh.get("scene_index"), sh.get("shot_index")
            if isinstance(si, int) and isinstance(shi, int):
                out[(si, shi)] = sh
        return out

    @staticmethod
    def _ve_by_shot(shot_director_cp: Optional[Dict[str, Any]]) -> Dict[ShotKey, List[str]]:
        out: Dict[ShotKey, List[str]] = {}
        for sc in (shot_director_cp or {}).get("data", {}).get("scenes", []) or []:
            si = sc.get("scene_index")
            for sh in sc.get("shots", []) or []:
                shi = sh.get("shot_index")
                if isinstance(si, int) and isinstance(shi, int):
                    out[(si, shi)] = list(sh.get("visible_entity_ids") or [])
        return out

    @staticmethod
    def _primary_location_by_scene(
        scene_director_cp: Optional[Dict[str, Any]],
    ) -> Dict[int, str]:
        """scene_director.scenes[].primary_location — scene-present outdoor location
        fallback source (4a single-shot lane). frame-visible VE 에 location 이 없는
        샷의 seed location resolve 에 쓴다(구조필드, 라벨 파싱 없음)."""
        out: Dict[int, str] = {}
        for sc in (scene_director_cp or {}).get("data", {}).get("scenes", []) or []:
            si = sc.get("scene_index")
            pl = sc.get("primary_location")
            if isinstance(si, int) and isinstance(pl, str) and pl:
                out[si] = pl
        return out

    @staticmethod
    def _t2i_variations_by_shot(
        scene_detail_cp: Optional[Dict[str, Any]],
    ) -> Dict[ShotKey, List[Tuple[int, str]]]:
        """(si, shi) → [(variation_index, t2i_prompt)] — t2i_review(21.72) 가
        scene_detail cp 를 직접 수정하므로 여기서 읽는 값이 최종본."""
        out: Dict[ShotKey, List[Tuple[int, str]]] = {}
        for sc in (scene_detail_cp or {}).get("data", {}).get("scenes", []) or []:
            si, shi = sc.get("scene_index"), sc.get("_shot_index")
            if not isinstance(si, int) or not isinstance(shi, int):
                continue
            variations: List[Tuple[int, str]] = []
            for i, v in enumerate(sc.get("t2i_variations") or []):
                prompt = v.get("t2i_prompt") if isinstance(v, dict) else None
                if isinstance(prompt, str) and prompt.strip():
                    variations.append((i, prompt))
            if variations:
                out[(si, shi)] = variations
        return out

    # ───────────────────── execute ─────────────────────

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

        if not bool(getattr(settings, "outdoor_site_layout_enabled", False)):
            return self._not_applicable()

        shot_cp = self._load_prev_checkpoint("shot_validator")
        if not shot_cp or not shot_cp.get("data", {}).get("scenes"):
            raise AppError(code="step.no_shots", message="shot_validator 결과 없음", status_code=400)
        from app.core.steps.shot_validator_step import assert_no_failed_scenes
        assert_no_failed_scenes(shot_cp, self.project_config, consumer_step="outdoor_site_layout")

        from app.core.steps.episode_reference_policy_step import build_selected_map_or_raise
        selected_map = build_selected_map_or_raise(self._load_prev_checkpoint("shot_selection"))

        bgc_cp = self._load_prev_checkpoint("background_classify")
        outdoor_locs = outdoor_loc_ids(
            (bgc_cp or {}).get("data", {}).get("building_groups") or []
        )
        # zoom 멤버 제외 신호 — soft read (cp 부재 = 제외 0 + merge 가 2차 방어).
        dep_cp = self._load_prev_checkpoint("shot_dependency_t2i")
        zoom_members = zoom_member_shots(
            (dep_cp or {}).get("data", {}).get("dependencies") or []
        )

        scene_texts = self._scene_texts(self._load_prev_checkpoint("scene_save"))
        shot_descriptions = self._shot_descriptions(shot_cp)
        staging_by_shot = self._staging_by_shot(self._load_prev_checkpoint("shot_staging"))
        ve_by_shot = self._ve_by_shot(self._load_prev_checkpoint("shot_director"))
        t2i_by_shot = self._t2i_variations_by_shot(self._load_prev_checkpoint("scene_detail"))
        # 4a fix (2026-07-01): single-shot lane 용 primary_location fallback source —
        # scene_director.primary_location(scene-present outdoor location). VE(frame-
        # visible)에 location 이 없는 샷(S15 sh5)도 seed 후보로 올리는 데 쓴다.
        primary_location_by_scene = self._primary_location_by_scene(
            self._load_prev_checkpoint("scene_director"))
        # broad single-shot seed lane flag — ON 이면 단일 figure 복잡/구도샷도 seed
        # (판정은 shared-model judge/QC fail-closed). OFF(default) = legacy byte-identical.
        single_shot_seed_lane = bool(getattr(
            settings, "outdoor_single_shot_seed_lane_enabled", False))

        seed_groups, seed_skipped, seed_meta = detect_site_seeds(
            selected_map,
            ve_by_shot=ve_by_shot,
            staging_by_shot=staging_by_shot,
            outdoor_locs=outdoor_locs,
            zoom_members=zoom_members,
            primary_location_by_scene=primary_location_by_scene,
            single_shot_lane_enabled=single_shot_seed_lane,
        )

        cap = int(getattr(settings, "outdoor_site_layout_group_cap", DEFAULT_GROUP_CAP))
        ordered_locs = sorted(seed_groups)
        selected_locs = ordered_locs[:cap]
        cap_skipped = [
            {"seed": f"location:{loc}", "reason": "cap"} for loc in ordered_locs[cap:]
        ]

        layout_fn = self._layout_override
        rewrite_fn = self._rewrite_override
        brief_fn = self._brief_override
        sketch_fn = self._sketch_override
        judge_fn = self._judge_override
        shared_base_fn = self._shared_base_override
        shared_guide_fn = self._shared_guide_override
        shared_judge_fn = self._shared_judge_override
        pose_brief_fn = self._pose_brief_override
        if (layout_fn is None or rewrite_fn is None or brief_fn is None
                or sketch_fn is None or judge_fn is None or shared_base_fn is None
                or shared_guide_fn is None or shared_judge_fn is None
                or pose_brief_fn is None):
            from app.modules.pipeline import outdoor_site_layout_provider as provider
            layout_fn = layout_fn or provider.emit_site_layout
            rewrite_fn = rewrite_fn or provider.rewrite_position_phrases
            brief_fn = brief_fn or provider.generate_composition_brief
            sketch_fn = sketch_fn or provider.generate_composition_sketch
            judge_fn = judge_fn or provider.judge_composition_guide_applicable
            shared_base_fn = shared_base_fn or provider.generate_location_aerial_base
            shared_guide_fn = (
                shared_guide_fn or provider.generate_shared_model_camera_guide)
            shared_judge_fn = shared_judge_fn or provider.judge_shared_model_guide
            # P1: shared-model v2 Stage C 인물 자세/동작 structured brief(무-가이드보다
            # 나쁜 generic standing 오염 복구). OFF/legacy 경로는 미사용.
            pose_brief_fn = pose_brief_fn or provider.generate_pose_brief

        # composition guide (2026-06-13 재배선): departing 샷 한정 3-모델 구도
        # 가이드 (브리프 gemini-pro → 마네킹 스케치 gpt-image-2) — 독립 flag,
        # 실패/부재 전부 비차단(가이드 없이 기존 경로 fallback). 적용 선별 =
        # 기하(receding mover) + 의미 게이트(열린 외부 departure judge).
        guide_enabled = bool(getattr(
            settings, "outdoor_composition_guide_enabled", False))
        # Phase II shared-model v2 (2026-06-29 재배선) — ON 이면 guide 경로 전체를
        # 새 route 로 분기: cross-shot continuity candidate + route judge + 3-stage
        # 라벨 마커 항공뷰 producer. OFF(default)면 아래 legacy(old-sketch) 경로 그대로
        # (byte-identical). 실패 = no-guide + diagnostic (old fallback 금지).
        shared_model_enabled = bool(getattr(
            settings, "outdoor_composition_guide_shared_model_enabled", False))
        shared_judge_enabled = bool(getattr(
            settings, "outdoor_shared_model_guide_judge_enabled", True))
        guide_model = str(getattr(
            settings, "outdoor_composition_guide_model", "gpt-image-2.5-sunburst"))
        brief_model = str(getattr(
            settings, "outdoor_composition_brief_model", "gemini-3.1-pro-preview"))
        judge_model = str(getattr(
            settings, "outdoor_composition_judge_model", "gemini-3.1-pro-preview"))

        groups: List[Dict[str, Any]] = []
        prompt_overrides: Dict[str, Dict[str, Any]] = {}
        llm_failed: List[Dict[str, Any]] = []
        layout_invalid: List[Dict[str, Any]] = []
        side_conflicts: List[Dict[str, Any]] = []
        summary_skipped: List[Dict[str, Any]] = []
        audit_violations: List[Dict[str, Any]] = []
        rewrite_failed: List[Dict[str, Any]] = []
        composition_guides: Dict[str, Dict[str, Any]] = {}
        guide_failed: List[Dict[str, Any]] = []
        guide_skipped: List[Dict[str, Any]] = []  # 의미 게이트 default-deny 진단
        guide_openai_client = None  # lazy — 첫 실제 스케치 시 1회 생성, 그룹 간 공유
        # Phase II shared-model v2 — 단계별 진단(ON 일 때만 manifest 에 기록) + group
        # 단위 base 캐시(set 일관성: 같은 그룹 base 1회 생성 후 재사용).
        shared_diag: Dict[str, Any] = {
            "candidate_signals": [], "judge_decisions": [], "attach_decisions": [],
            "single_shot_complexity_observation": [],
            "predicted_image_calls": {"base": 0, "blocking": 0, "camera_view_sketch": 0},
            "actual_image_calls": {"base": 0, "blocking": 0, "camera_view_sketch": 0},
        }
        shared_base_cache: Dict[str, Tuple[bytes, Dict[str, Any]]] = {}

        # W-G (2026-07-03) — same-building indoor fp 링크: 같은 building_groups
        # 그룹에 실내·실외 loc 이 공존하면 그 건물 indoor floor plan 을 Stage A
        # aerial base 의 구조 참조(I2I ref)로 승격한다. 전부 구조 필드 조인
        # (building_fp_link 모듈) — flag OFF(default)면 로더 미호출 = byte-identical.
        building_fp_by_loc: Dict[str, Dict[str, str]] = {}
        if bool(getattr(settings, "outdoor_building_fp_ref_enabled", False)):
            try:
                from app.modules.pipeline.building_fp_link import (
                    build_outdoor_building_fp_link,
                    indoor_fp_ids_by_loc,
                )
                _mp_cp = self._load_prev_checkpoint("background_master_plan")
                _fpr_cp = self._load_prev_checkpoint("floor_plan_render")
                _catalog = (
                    (_mp_cp or {}).get("data", {}) or {}
                ).get("background_catalog") or {}
                _fp_link = build_outdoor_building_fp_link(
                    (bgc_cp or {}).get("data", {}).get("building_groups") or [],
                    indoor_fp_ids_by_loc(_catalog),
                )
                _fpr_map = (
                    (_fpr_cp or {}).get("data", {}) or {}
                ).get("floor_plans") or {}
                for _oloc, _lk in _fp_link.items():
                    _fp_entry = _fpr_map.get(_lk.get("fp_id")) or {}
                    _png = str(_fp_entry.get("png_path") or "")
                    if _fp_entry.get("status") == "ok" and _png:
                        building_fp_by_loc[_oloc] = {**_lk, "png_path": _png}
                shared_diag["building_fp_link"] = {
                    loc: {k: v for k, v in lk.items() if k != "png_path"}
                    for loc, lk in building_fp_by_loc.items()
                }
            except Exception as exc:  # noqa: BLE001 — 비차단(링크 없이 기존 경로)
                logger.warning(
                    "outdoor_site_layout: building fp 링크 로드 실패 (비차단): %s",
                    exc)
                building_fp_by_loc = {}

        for loc_id in selected_locs:
            member_keys = sorted(seed_groups[loc_id])
            member_scene_indices = sorted({si for si, _ in member_keys})
            group_scene_texts = [
                (si, scene_texts.get(si, "")) for si in member_scene_indices
            ]
            member_payload = [
                {
                    "scene_index": si,
                    "shot_index": shi,
                    "description": shot_descriptions.get((si, shi), ""),
                    "staging": staging_by_shot.get((si, shi)),
                }
                for si, shi in member_keys
            ]
            try:
                layout = layout_fn(
                    group_scene_texts,
                    member_payload,
                    project_config=self.project_config,
                    opik_metadata=self.build_opik_metadata(),
                )
            except Exception as exc:
                logger.warning("outdoor_site_layout: %s layout 추출 실패: %s", loc_id, exc)
                llm_failed.append({"seed": f"location:{loc_id}", "reason": "llm_error"})
                continue

            violations = validate_site_layout(layout)
            if violations:
                # shape/범위 위반 = 그 그룹 산출 폐기 + 진단 (비차단 — 의미
                # 게이트가 아니라 좌표 계약 위반).
                layout_invalid.append({
                    "seed": f"location:{loc_id}", "violations": violations,
                })
                llm_failed.append({
                    "seed": f"location:{loc_id}", "reason": "layout_shape_invalid",
                })
                continue

            # ── lateral SOT=staging: layout 좌/우 ↔ staged screen_zone 검증
            # (deterministic 조인). 충돌 시 1회 retry — 잔존 충돌 샷은 override
            # 생성에서 제외하고 원본 scene_detail prompt 로 fallback (Codex
            # R2~R4 리뷰 BLOCKING NARROW 1: R4 요약은 frame 좌/우를 prompt 에
            # 넣으므로, staging 과 충돌하는 layout 을 그대로 소비하면 안 된다).
            def _group_conflicts(candidate: Dict[str, Any]) -> List[Dict[str, Any]]:
                found: List[Dict[str, Any]] = []
                for c_si, c_shi in member_keys:
                    found.extend(layout_side_conflicts(
                        candidate, c_shi,
                        staged_sides_for_shot(staging_by_shot.get((c_si, c_shi))),
                    ))
                return found

            conflicted_shot_indices: Set[int] = set()
            conflicts = _group_conflicts(layout)
            if conflicts:
                try:
                    retried = layout_fn(
                        group_scene_texts,
                        member_payload,
                        project_config=self.project_config,
                        opik_metadata=self.build_opik_metadata(),
                        constraint_feedback=json.dumps(conflicts, ensure_ascii=False),
                    )
                    if not validate_site_layout(retried):
                        retry_conflicts = _group_conflicts(retried)
                        # 엄격히 개선된 경우만 채택 — 동수면 원본 유지.
                        if len(retry_conflicts) < len(conflicts):
                            layout, conflicts = retried, retry_conflicts
                except Exception as exc:
                    logger.warning(
                        "outdoor_site_layout: %s side-conflict retry 실패: %s",
                        loc_id, exc)
                if conflicts:
                    side_conflicts.append({
                        "seed": f"location:{loc_id}", "conflicts": conflicts,
                    })
                    conflicted_shot_indices = {
                        c["shot_index"] for c in conflicts
                    }

            for shi in sorted(camera_missing_shots(layout, {k[1] for k in member_keys})):
                summary_skipped.append({
                    "shot": f"{loc_id}:sh{shi}", "reason": "camera_missing_in_layout",
                })

            spatial_summaries: Dict[str, str] = {}
            for si, shi in member_keys:
                if shi in conflicted_shot_indices:
                    summary_skipped.append({
                        "shot": f"S{si}sh{shi}", "reason": "layout_side_conflict",
                    })
                    continue
                summary = build_shot_spatial_summary(layout, shi)
                if summary is None:
                    summary_skipped.append({
                        "shot": f"S{si}sh{shi}", "reason": "no_spatial_summary",
                    })
                    continue
                spatial_summaries[shot_hint_key(si, shi)] = summary

                variations = t2i_by_shot.get((si, shi)) or []
                if not variations:
                    summary_skipped.append({
                        "shot": f"S{si}sh{shi}", "reason": "no_t2i_variations",
                    })
                    continue
                try:
                    rewritten = rewrite_fn(
                        variations,
                        summary,
                        project_config=self.project_config,
                        opik_metadata=self.build_opik_metadata(),
                    )
                except Exception as exc:
                    logger.warning(
                        "outdoor_site_layout: S%ssh%s 재작문 실패: %s", si, shi, exc)
                    rewrite_failed.append({
                        "shot": f"S{si}sh{shi}", "reason": "llm_error",
                    })
                    continue

                original_by_idx = dict(variations)
                revised_map: Dict[str, str] = {}
                provenance: Dict[str, Dict[str, str]] = {}
                edited_spans: Dict[str, List[Dict[str, str]]] = {}
                no_edit_reason: Dict[str, Optional[str]] = {}
                for idx, item in sorted(rewritten.items()):
                    original = original_by_idx.get(idx)
                    revised = (item or {}).get("revised_prompt")
                    if original is None or not isinstance(revised, str) or not revised.strip():
                        continue
                    token_violations = revised_prompt_token_violations(original, revised)
                    if token_violations:
                        audit_violations.append({
                            "shot": f"S{si}sh{shi}",
                            "variation_index": idx,
                            **token_violations,
                        })
                        continue  # revised 폐기 — substitution 미적용 (비차단)
                    if revised.strip() == original.strip():
                        # 무수정 — override 불필요 (no_edit_reason 만 기록).
                        no_edit_reason[str(idx)] = item.get("no_edit_reason")
                        continue
                    revised_map[str(idx)] = revised
                    provenance[str(idx)] = {
                        "original_scene_detail_prompt_hash": _hash16(original),
                        "revised_prompt_hash": _hash16(revised),
                    }
                    edited_spans[str(idx)] = item.get("edited_spans") or []
                    if item.get("no_edit_reason"):
                        no_edit_reason[str(idx)] = item["no_edit_reason"]
                if revised_map:
                    prompt_overrides[shot_hint_key(si, shi)] = {
                        "location_id": loc_id,
                        "group_id": f"osl-{loc_id.lower()}",
                        "revised": revised_map,
                        "provenance": provenance,
                        "edited_spans": edited_spans,
                        "no_edit_reason": no_edit_reason,
                    }

            # ── composition guide — shared-model v2(ON) / option C legacy(OFF) ──
            # ON(shared_model): cross-shot continuity candidate + route judge + 3-stage
            #   라벨 마커 항공뷰 producer(group base 캐싱). 별도 메서드로 분리 — OFF 경로의
            #   departure candidate/G1·G2 judge/마네킹 sketch 를 호출하지 않는다(Codex).
            # OFF(legacy, byte-identical): 기하(receding) → 의미(열린 외부 departure
            #   judge) → admitted same-scene 체인(첫=마네킹 스케치 / 이후=continuity_anchor).
            # 둘 다 비차단(실패=가이드 없이 기존 경로 fallback). summary_keys=요약 생성 샷만.
            if guide_enabled and spatial_summaries and shared_model_enabled:
                self._generate_shared_model_guides(
                    loc_id=loc_id, layout=layout, member_keys=member_keys,
                    spatial_summaries=spatial_summaries,
                    staging_by_shot=staging_by_shot, guide_model=guide_model,
                    judge_model=judge_model, judge_enabled=shared_judge_enabled,
                    base_fn=shared_base_fn, guide_fn=shared_guide_fn,
                    judge_fn=shared_judge_fn, composition_guides=composition_guides,
                    guide_failed=guide_failed, guide_skipped=guide_skipped,
                    shared_diag=shared_diag, base_cache=shared_base_cache,
                    seed_meta=seed_meta,
                    # P1: Stage C 포즈 복원 — 씬 액션 원문(t2i)/shot 설명 + pose brief LLM.
                    t2i_by_shot=t2i_by_shot, shot_descriptions=shot_descriptions,
                    pose_brief_fn=pose_brief_fn, brief_model=brief_model,
                    # W-G: same-building indoor fp (없으면 None = 기존 경로).
                    building_fp=building_fp_by_loc.get(loc_id))
            elif guide_enabled and spatial_summaries:
                # 1) 기하 후보 → 의미 게이트(default-deny) → admitted 수집.
                admitted: List[Dict[str, Any]] = []
                for g_si, g_shi in composition_guide_shot_keys(
                        layout, member_keys, set(spatial_summaries)):
                    g_key = shot_hint_key(g_si, g_shi)
                    g_summary = spatial_summaries[g_key]
                    # scene action body = primary t2i 본문(없으면 shot 설명) —
                    # 브리프가 '누가 누굴 보나/이동하나' 의미 맥락에서 배향 도출.
                    g_variations = t2i_by_shot.get((g_si, g_shi)) or []
                    g_body = (
                        min(g_variations, key=lambda v: v[0])[1]
                        if g_variations
                        else shot_descriptions.get((g_si, g_shi), "")
                    )
                    # 열린 외부 departure 만 통과 — 창 너머/문턱/실내 구조물·근접
                    # 대치 배제 (judge 실패/저신뢰/근거없음 전부 skip = fallback).
                    g_cam = str((staging_by_shot.get((g_si, g_shi)) or {}).get(
                        "camera_direction") or "")
                    try:
                        verdict = judge_fn(
                            g_body, g_cam, g_summary, model=judge_model,
                            log_context={
                                "project_id": self.project_id,
                                "episode_id": self.episode_id,
                                "operation_type": "outdoor_composition_judge",
                                "step_name": "outdoor_site_layout",
                            },
                        )
                    except Exception as exc:
                        logger.warning(
                            "outdoor_site_layout: S%ssh%s guide 적용성 판정 실패 "
                            "(비차단, skip): %s", g_si, g_shi, exc)
                        guide_skipped.append({
                            "shot": f"S{g_si}sh{g_shi}", "reason": "judge_failed",
                        })
                        continue
                    admit, skip_reason, evidence = _evaluate_guide_judge(verdict)
                    if not admit:
                        guide_skipped.append({
                            "shot": f"S{g_si}sh{g_shi}",
                            "reason": skip_reason,
                            "confidence": (verdict or {}).get("confidence"),
                            "evidence_quote": evidence,
                        })
                        continue
                    admitted.append({
                        "key": (g_si, g_shi), "verdict": verdict,
                        "evidence": evidence, "summary": g_summary, "body": g_body,
                    })

                # 2) admitted 를 same-scene 연속성 체인으로 → 첫=sketch / 이후=anchor.
                admitted_by_key = {a["key"]: a for a in admitted}
                for chain in composition_continuity_chain(
                        [a["key"] for a in admitted]):
                    g_si, g_shi = chain["key"]
                    g_key = shot_hint_key(g_si, g_shi)
                    a = admitted_by_key[chain["key"]]
                    verdict, evidence = a["verdict"], a["evidence"]
                    group_id = f"oslcg-{loc_id.lower()}-s{g_si}"
                    judge_block = {
                        "confidence": (verdict or {}).get("confidence"),
                        "evidence_quote": evidence,
                    }
                    this_staged = _staged_character_names(
                        staging_by_shot.get((g_si, g_shi)))

                    if chain["mode"] == "sketch":
                        # 첫 admitted 샷 — 마네킹 브리프 + 구도 스케치 생성(3-model).
                        # 참조할 같은-그룹 완성 프레임이 없으므로 스케치로 구도 확립.
                        try:
                            brief = brief_fn(
                                a["body"], a["summary"], model=brief_model,
                                log_context={
                                    "project_id": self.project_id,
                                    "episode_id": self.episode_id,
                                    "operation_type": "outdoor_composition_brief",
                                    "step_name": "outdoor_site_layout",
                                },
                            )
                            if guide_openai_client is None:
                                guide_openai_client = (
                                    self._resolve_guide_openai_client())
                            png = sketch_fn(
                                brief, openai_client=guide_openai_client,
                                model=guide_model)
                            rel = f"{_GUIDE_SUBDIR_NAME}/S{g_si}sh{g_shi}.png"
                            out_path = self._checkpoint_dir() / rel
                            out_path.parent.mkdir(parents=True, exist_ok=True)
                            out_path.write_bytes(png)
                        except Exception as exc:
                            logger.warning(
                                "outdoor_site_layout: S%ssh%s composition guide "
                                "스케치 생성 실패 (비차단): %s", g_si, g_shi, exc)
                            guide_failed.append({
                                "shot": f"S{g_si}sh{g_shi}",
                                "reason": str(exc)[:200],
                            })
                            continue
                        composition_guides[g_key] = {
                            "mode": "sketch",
                            "path": rel,
                            "location_id": loc_id,
                            "group_id": group_id,
                            "admitted_order": chain["admitted_order"],
                            "model": guide_model,
                            "brief_model": brief_model,
                            "judge_model": judge_model,
                            "judge": judge_block,
                            "brief_hash": _hash16(brief),
                            "sketch_prompt_hash": _hash16(_sketch_prompt_template()),
                            "source_summary_hash": _hash16(a["summary"]),
                            # 첫 샷은 source 없음 — this-shot staged 인물을 강제 대상.
                            "forced_character_names": sorted(this_staged),
                        }
                    else:
                        # 이후 admitted 샷 — 직전 admitted 샷의 현재-run 완성 프레임을
                        # 연속성 anchor 로(사전 스케치 미생성). forced character 는
                        # this-shot ∩ anchor_source staged 교집합만 (source 에 없는
                        # 인물 강제 차단 — Codex 안전 가드).
                        psi, pshi = chain["anchor_source"]
                        source_staged = _staged_character_names(
                            staging_by_shot.get((psi, pshi)))
                        composition_guides[g_key] = {
                            "mode": "continuity_anchor",
                            "anchor_source": [psi, pshi],
                            "location_id": loc_id,
                            "group_id": group_id,
                            "admitted_order": chain["admitted_order"],
                            "judge_model": judge_model,
                            "judge": judge_block,
                            "source_summary_hash": _hash16(a["summary"]),
                            "forced_character_names": sorted(
                                this_staged & source_staged),
                        }

            groups.append({
                "group_id": f"osl-{loc_id.lower()}",
                "location_id": loc_id,
                "member_shots": [list(k) for k in member_keys],
                "layout": layout,
                "spatial_summaries": spatial_summaries,
            })

        # goal#3 (2026-07-01): outdoor 항공뷰 체인(aerial→blocking→sketch) input_image_ids
        # lineage 를 구조키로 post-hoc resolve+annotate + camera_sketch UUID 회수(final
        # scene edge). shared_model ON + 가이드 실제 생성됐을 때만 — OFF/legacy/미생성은
        # no-op(byte-identical, db 미접촉). 비차단(실패=진단만).
        chain_lineage_diag: Dict[str, Any] = {}
        if shared_model_enabled and composition_guides:
            chain_lineage_diag = _resolve_outdoor_chain_lineage(
                getattr(self, "db", None), self.project_id, self.episode_id,
                composition_guides)

        diagnostics = {
            "outdoor_location_count": len(outdoor_locs),
            "seed_group_count": len(seed_groups),
            "groups_extracted": len(groups),
            "skipped": seed_skipped + cap_skipped + llm_failed,
            "layout_shape_violations": layout_invalid,
            "layout_side_conflicts": side_conflicts,
            "spatial_summary_skipped": summary_skipped,
            "rewrite_failed": rewrite_failed,
            "revised_prompt_audit_violations": audit_violations,
            "zoom_member_exclusion_source": "shot_dependency_t2i",
            "composition_guide_failed": guide_failed,
            "composition_guide_candidate_skipped": guide_skipped,
            "composition_guide_count": len(composition_guides),
            # 4a fix: single-shot lane seed prefilter 진단(구조필드만, 의미판정 0).
            # 과발동 튜닝/audit 의 SOT. ShotKey tuple → "si:shi" 문자열 키로 직렬화.
            "seed_prefilter_meta": {
                f"{k[0]}:{k[1]}": v for k, v in seed_meta.items()
            },
            "single_shot_seed_lane_enabled": single_shot_seed_lane,
        }
        # Phase II shared-model v2 진단 — ON 일 때만 추가(OFF byte-identical). 단계별
        # candidate signals / judge route / attach 결정 + 예측·실제 이미지 호출수.
        if shared_model_enabled:
            predicted = shared_diag["predicted_image_calls"]
            actual = shared_diag["actual_image_calls"]
            diagnostics["composition_guide_shared_model_enabled"] = True
            diagnostics["shared_model_judge_enabled"] = shared_judge_enabled
            diagnostics["shared_model_candidate_signals"] = (
                shared_diag["candidate_signals"])
            diagnostics["shared_model_judge_decisions"] = (
                shared_diag["judge_decisions"])
            diagnostics["shared_model_attach_decisions"] = (
                shared_diag["attach_decisions"])
            diagnostics["shared_model_single_shot_complexity_observation"] = (
                shared_diag["single_shot_complexity_observation"])
            # P1: pose 추출 실패/저신뢰로 fail-closed(가이드 미부착)된 샷 진단.
            diagnostics["shared_model_pose_fail_closed"] = (
                shared_diag.get("pose_fail_closed") or [])
            # optical/reflective 구도 게이트(2026-07-02)로 제외된 샷 진단.
            diagnostics["shared_model_optical_risk_excluded"] = (
                shared_diag.get("optical_risk_excluded") or [])
            # W-G: same-building fp 링크(outdoor loc → indoor fp) 관측 필드.
            diagnostics["shared_model_building_fp_link"] = (
                shared_diag.get("building_fp_link") or {})
            diagnostics["shared_model_image_calls"] = {
                "predicted": predicted, "actual": actual,
                "predicted_call_count": sum(predicted.values()),
                "actual_call_count": sum(actual.values()),
                "continuity_anchor_generation_calls": 0,
            }
            # goal#3: 체인 lineage post-hoc resolve 결과(annotate/회수 진단).
            diagnostics["shared_model_chain_lineage"] = chain_lineage_diag
        manifest = build_manifest(groups, prompt_overrides, diagnostics)
        # composition guide (flag-gated 조건부 필드 — flag OFF 면 미기록,
        # config_hash 가 flag 토글을 감지한다).
        if guide_enabled:
            manifest["composition_guides"] = composition_guides

        violations = validate_site_manifest(manifest, selected_map)
        if violations:
            raise AppError(
                code="outdoor_site_layout.manifest_invalid",
                message="site manifest 조인 무결성 위반: " + "; ".join(violations),
                status_code=500,
            )

        gallery_rel = self._write_review_html(manifest)

        return {
            # Codex W8_CODE_REVIEW NARROW_1: 이 step 은 image-phase prompt
            # override quality layer — layout 실패 그룹은 원본 scene_detail
            # prompt 로 그대로 fallback 하므로 **비차단**이어야 한다.
            # failed_count > 0 은 StepRunner 가 status='partial' 로 저장해
            # (step_runner.py final_status) active analysis step 인 이 step 이
            # fresh run 의 하류 진입을 막는다 → 실패는 diagnostics(skipped/
            # layout_shape_violations)로만 surface 하고 failed_count=0 고정.
            "applicable_count": len(selected_locs),
            "completed_count": len(groups),
            "failed_count": 0,
            "schema_version": SCHEMA_VERSION,
            "config_hash": self._config_hash(),
            "data": {**manifest, "review_html": gallery_rel},
        }

    # ───────────────────── review gallery ─────────────────────

    def _write_review_html(self, manifest: Dict[str, Any]) -> Optional[str]:
        try:
            out_dir = self._checkpoint_dir() / _HTML_SUBDIR_NAME
            out_dir.mkdir(parents=True, exist_ok=True)
            path = out_dir / "index.html"
            path.write_text(render_site_review_html(manifest), encoding="utf-8")
            return f"{_HTML_SUBDIR_NAME}/index.html"
        except Exception as exc:
            logger.warning("outdoor_site_layout: review html 생성 실패: %s", exc)
            return None


def render_layout_svg(layout: Dict[str, Any]) -> str:
    """layout 좌표 → inline SVG (review 갤러리 **전용** — scene 생성 ref 부착
    금지: ablation 실측으로 이미지 ref 채널은 기하 명세 효과가 없고 오염 위험만
    있다. Codex ⓓ 문서화 의무)."""
    def pt(p: Any) -> Tuple[float, float]:
        # SVG y 축은 아래로 증가 — top-down 좌표(0-100, y=north)를 뒤집는다.
        return float(p[0]), 100.0 - float(p[1])

    parts: List[str] = []
    for lm in layout.get("landmarks") or []:
        points = lm.get("points") or []
        label = html.escape(str(lm.get("label") or ""))
        kind = lm.get("kind")
        coords = [pt(p) for p in points]
        if not coords:
            continue
        if kind == "area" and len(coords) >= 3:
            pts_s = " ".join(f"{x:.1f},{y:.1f}" for x, y in coords)
            parts.append(f"<polygon points='{pts_s}' fill='#3a4a5a' opacity='0.5'/>")
        elif kind == "line" and len(coords) >= 2:
            pts_s = " ".join(f"{x:.1f},{y:.1f}" for x, y in coords)
            parts.append(
                f"<polyline points='{pts_s}' fill='none' stroke='#888' stroke-width='2'/>")
        else:
            x, y = coords[0]
            parts.append(f"<rect x='{x - 1.2:.1f}' y='{y - 1.2:.1f}' width='2.4' height='2.4' fill='#caa84e'/>")
        x, y = coords[0]
        parts.append(
            f"<text x='{x:.1f}' y='{y - 1.8:.1f}' font-size='2.6' fill='#9ab'>{label}</text>")

    for fig in layout.get("figures") or []:
        label = html.escape(str(fig.get("label") or fig.get("figure_id") or ""))
        for pos_entry in fig.get("positions") or []:
            if pos_entry.get("pos") is None:
                continue
            x, y = pt(pos_entry["pos"])
            shot = pos_entry.get("shot_index")
            parts.append(f"<circle cx='{x:.1f}' cy='{y:.1f}' r='1.4' fill='#b03a2e'/>")
            parts.append(
                f"<text x='{x:.1f}' y='{y + 3.4:.1f}' font-size='2.4' fill='#e8a' "
                f"text-anchor='middle'>{label} (sh{shot})</text>")
            mt = pos_entry.get("moving_toward")
            if mt is not None:
                mx, my = pt(mt)
                parts.append(
                    f"<line x1='{x:.1f}' y1='{y:.1f}' x2='{mx:.1f}' y2='{my:.1f}' "
                    "stroke='#b03a2e' stroke-width='0.7' stroke-dasharray='2,1'/>")
                parts.append(f"<circle cx='{mx:.1f}' cy='{my:.1f}' r='0.7' fill='#b03a2e'/>")

    for cam in layout.get("cameras") or []:
        if cam.get("pos") is None or cam.get("look_at") is None:
            continue
        x, y = pt(cam["pos"])
        lx, ly = pt(cam["look_at"])
        parts.append(
            f"<line x1='{x:.1f}' y1='{y:.1f}' x2='{lx:.1f}' y2='{ly:.1f}' "
            "stroke='#7ee787' stroke-width='0.6' stroke-dasharray='1.5,1.5'/>")
        parts.append(f"<rect x='{x - 1.3:.1f}' y='{y - 1.3:.1f}' width='2.6' height='2.6' fill='#7ee787'/>")
        parts.append(
            f"<text x='{x:.1f}' y='{y - 2.2:.1f}' font-size='2.6' fill='#7ee787' "
            f"text-anchor='middle'>cam sh{cam.get('shot_index')}</text>")

    return (
        "<svg viewBox='-4 -4 108 108' width='480' height='480' "
        "style='background:#1b222c;border:1px solid #345;border-radius:6px'>"
        + "".join(parts) + "</svg>"
    )


def render_site_review_html(manifest: Dict[str, Any]) -> str:
    """groups(layout SVG + 공간 요약) + original↔revised 비교 — 사람 검토용."""
    def esc(v: Any) -> str:
        return html.escape(str(v if v is not None else ""))

    overrides = manifest.get("prompt_overrides") or {}
    rows: List[str] = []
    for g in manifest.get("groups", []) or []:
        members = ", ".join(
            f"S{si}sh{shi}" for si, shi in (tuple(m) for m in g.get("member_shots") or [])
        )
        summaries = "".join(
            f"<details><summary>spatial summary [{esc(k)}]</summary>"
            f"<pre>{esc(v)}</pre></details>"
            for k, v in sorted((g.get("spatial_summaries") or {}).items())
        )
        revised_blocks = ""
        for key_s, entry in sorted(overrides.items()):
            if entry.get("location_id") != g.get("location_id"):
                continue
            for idx, revised in sorted((entry.get("revised") or {}).items()):
                prov = (entry.get("provenance") or {}).get(idx, {})
                spans = "".join(
                    f"<li><del>{esc(sp.get('original'))}</del> → "
                    f"<ins>{esc(sp.get('revised'))}</ins></li>"
                    for sp in (entry.get("edited_spans") or {}).get(idx, [])
                )
                revised_blocks += (
                    f"<details><summary>revised [{esc(key_s)} / variation {esc(idx)}] "
                    f"<code>{esc(prov.get('original_scene_detail_prompt_hash'))}→"
                    f"{esc(prov.get('revised_prompt_hash'))}</code></summary>"
                    f"<ul>{spans}</ul><pre>{esc(revised)}</pre></details>"
                )
        rows.append(
            f"<section><h2>{esc(g.get('group_id'))} <small>{esc(g.get('location_id'))}</small></h2>"
            f"<p>{members}</p>{render_layout_svg(g.get('layout') or {})}"
            f"{summaries}{revised_blocks}</section>"
        )

    diag = json.dumps(manifest.get("diagnostics", {}), ensure_ascii=False, indent=1)
    return (
        "<!DOCTYPE html><html lang='ko'><head><meta charset='utf-8'>"
        "<title>outdoor_site_layout review</title>"
        "<style>body{font-family:sans-serif;background:#14181f;color:#dde;max-width:980px;"
        "margin:0 auto;padding:24px;line-height:1.5}section{border:1px solid #345;"
        "border-radius:8px;padding:12px 16px;margin:14px 0}h2 small{color:#8ac;font-weight:400}"
        "small{color:#9ab}code{color:#9ecbff;font-size:.85em}pre{background:#0d1117;"
        "padding:10px;border-radius:6px;overflow-x:auto;white-space:pre-wrap}"
        "details{margin:8px 0}summary{cursor:pointer;color:#7ee787}"
        "del{color:#f85149}ins{color:#7ee787;text-decoration:none}</style></head><body>"
        "<h1>outdoor_site_layout — groups "
        f"{len(manifest.get('groups', []) or [])}</h1>"
        "<p>layout 이미지는 review 전용 — scene 생성 ref 로 부착하지 않는다 "
        "(ablation 실측: 이미지 ref 채널은 기하 명세 효과 없음).</p>"
        + "".join(rows)
        + f"<h2>diagnostics</h2><pre>{html.escape(diag)}</pre></body></html>"
    )