"""표적 씬 실행 scope — 씬 이미지 슬라이스 실행 결정론 계약 (2026-07-23).

사용자 지시(슬라이스 E)의 범용화: 에피소드 전체가 아니라 표적 씬
부분집합만 씬 이미지(스틸)를 생성한다. Codex 설계 합의 조건:

- BLOCKING-1: 전체 stills/stills_orm 은 컨텍스트 SOT 로 끝까지 유지
  (월드 가이드·프롬프트 population·groupbg 컨텍스트/group_sig·canonical
  origin 전부 전체 목록 전제) — **생성 실행 루프만** effective allowlist
  로 제한한다(recipe ordered loop / legacy scene_batch).
- BLOCKING-2: 의존 클로저=소비 경로별 **최종 그래프의 재귀 고정점**.
  legacy=base+zoom+composition+immobilized edge 합산 그래프,
  recipe=shot_ref_classify+background_share_plan 적용 후 effective
  prev 체인. cycle/결손=fail-closed.
- HIGH-3: requested→dependency→effective 를 이 모듈의 순수 함수가
  단일 산출하고 audit(requested_scenes/requested_ids/dependency_added/
  effective_ids)을 영속 — 서비스 실행·스텝 카운트·verify 가 동일 audit
  을 소비(독자 필터 드리프트 금지). 표적 씬이 selected still 0개면
  fail-fast(오타/재세그먼트 mismatch).
- 빈 설정("")=전체 생성 — 모든 소비처 byte-identical.

시나리오 의존 없음 — 씬 index 는 범용 정수 목록이다.
"""
from __future__ import annotations

import json
from pathlib import Path
from typing import Any, Dict, List, Mapping, Sequence, Set, Tuple

AUDIT_FILENAME = "scene_target_scope.json"
SCOPE_CONTRACT_VERSION = 1


def parse_target_scenes(raw: Any) -> Tuple[int, ...]:
    """설정 문자열 → 정규화 씬 튜플 (정렬·중복 제거).

    ""/None = () (표적 미설정 — 전체 생성). 정수 아닌 토큰/음수 =
    ValueError fail-closed (조용한 무시 금지 — 오타가 전체 생성으로
    조용히 확장되는 경로 차단).
    """
    text = str(raw or "").strip()
    if not text:
        return ()
    out: Set[int] = set()
    for tok in text.split(","):
        tok = tok.strip()
        if not tok:
            continue
        if not tok.lstrip("-").isdigit():
            raise ValueError(
                f"scene_image_target_scenes 파싱 실패: {tok!r} (정수 목록 필요)")
        val = int(tok)
        if val < 0:
            raise ValueError(
                f"scene_image_target_scenes 음수 씬 index: {val}")
        out.add(val)
    if not out:
        raise ValueError(
            f"scene_image_target_scenes 비어 있지 않은데 유효 씬 0개: {raw!r}")
    return tuple(sorted(out))


def requested_still_ids(
    stills: Sequence[Mapping[str, Any]],
    target_scenes: Sequence[int],
) -> List[str]:
    """표적 씬의 생성 대상 still id 목록 (stills=selected·non-stale 우주).

    표적 씬 중 still 0개인 씬 존재 = ValueError fail-fast (Codex HIGH-3:
    completed=0 위장 금지 — 오타/재세그먼트 mismatch 신호).
    """
    targets = set(target_scenes)
    by_scene: Dict[int, List[str]] = {t: [] for t in targets}
    for s in stills:
        si = s.get("scene_index")
        if isinstance(si, int) and si in targets and s.get("id"):
            by_scene[si].append(str(s["id"]))
    empty = sorted(t for t, ids in by_scene.items() if not ids)
    if empty:
        raise ValueError(
            "표적 씬에 생성 대상 still 이 없음 — 씬 번호 오타 또는 "
            f"재세그먼테이션 mismatch (fail-fast): scenes={empty}")
    out: List[str] = []
    for t in sorted(by_scene):
        out.extend(by_scene[t])
    return out


def closure_over_index_graph(
    stills: Sequence[Mapping[str, Any]],
    requested_ids: Sequence[str],
    deps: Mapping[int, Set[int]],
) -> Tuple[List[str], List[str]]:
    """(legacy 경로) 최종 합산 dep 그래프의 재귀 클로저.

    deps=stills 인덱스 기반 target→source 그래프 (base+zoom+composition+
    immobilized edge 가 전부 합쳐진 뒤 호출 — Codex BLOCKING-2).
    반환=(effective_ids, dependency_added_ids). 미해석 인덱스=결손
    fail-closed.
    """
    id_by_idx = {i: str(s.get("id") or "") for i, s in enumerate(stills)}
    idx_by_id = {v: k for k, v in id_by_idx.items() if v}
    missing = [rid for rid in requested_ids if rid not in idx_by_id]
    if missing:
        raise ValueError(
            f"requested still 이 stills 우주에 없음 (fail-closed): "
            f"{missing[:5]}")
    seen: Set[int] = set()
    stack = [idx_by_id[rid] for rid in requested_ids]
    while stack:
        i = stack.pop()
        if i in seen:
            continue
        seen.add(i)
        for src in deps.get(i, ()):  # target→source 방향
            if src not in seen:
                if not id_by_idx.get(src):
                    raise ValueError(
                        f"dep source still[{src}] id 결손 (fail-closed)")
                stack.append(src)
    requested = set(requested_ids)
    effective = sorted(id_by_idx[i] for i in seen)
    added = sorted(x for x in effective if x not in requested)
    return effective, added


def build_effective_prev_map(
    tags: Sequence[str],
    classify_shots: Mapping[str, Mapping[str, Any]],
    share_plans: Mapping[str, Mapping[str, Any]],
) -> Dict[str, str]:
    """(recipe 경로) tag→effective prev tag — 실행 루프와 동일 해석.

    shot_ref_classify 의 prev 에 background_share_plan 의
    apply_share_plan_prev 를 적용한 값 (Codex BLOCKING-2: 두 CP 적용
    후의 effective prev 를 클로저 입력으로). 보수적 상한 — 실행 루프의
    bg_only 등 조건 분기로 prev 를 안 쓸 수 있으나, 클로저는 초과
    포함이 안전하고 미포함이 결손이다.
    """
    from app.modules.pipeline.still_recipe import apply_share_plan_prev

    out: Dict[str, str] = {}
    for tag in tags:
        prev = (classify_shots.get(tag) or {}).get("prev")
        plan = share_plans.get(tag)
        if plan:
            prev = apply_share_plan_prev(prev, plan)
        if prev:
            out[tag] = str(prev)
    return out


def closure_over_prev_chain(
    requested_tags: Sequence[str],
    prev_of: Mapping[str, str],
    known_tags: Set[str],
) -> Tuple[List[str], List[str]]:
    """(recipe 경로) effective prev 체인 재귀 클로저.

    반환=(effective_tags 정렬, dependency_added_tags). prev 가 known
    (스틸 우주) 밖=결손 fail-closed, cycle=fail-closed.
    """
    effective: Set[str] = set()
    for start in requested_tags:
        path: Set[str] = set()
        cur: str | None = start
        while cur:
            if cur in path:
                raise ValueError(
                    f"prev 체인 cycle 감지 (fail-closed): {cur} (시작 {start})")
            path.add(cur)
            if cur not in known_tags:
                raise ValueError(
                    f"prev 체인이 스틸 우주 밖을 참조 (fail-closed): "
                    f"{cur} (시작 {start})")
            if cur in effective:
                break  # 이미 닫힌 체인
            effective.add(cur)
            cur = prev_of.get(cur)
    requested = set(requested_tags)
    added = sorted(t for t in effective if t not in requested)
    return sorted(effective), added


def build_scope_audit(
    *,
    target_scenes: Sequence[int],
    requested_ids: Sequence[str],
    dependency_added_ids: Sequence[str],
    path_kind: str,
) -> Dict[str, Any]:
    """audit 레코드 — 서비스 실행·스텝 카운트·verify 의 단일 SOT."""
    effective = sorted({*requested_ids, *dependency_added_ids})
    return {
        "contract_version": SCOPE_CONTRACT_VERSION,
        "path_kind": path_kind,  # "recipe" | "batch"
        "requested_scenes": sorted(set(int(x) for x in target_scenes)),
        "requested_ids": sorted(set(str(x) for x in requested_ids)),
        "dependency_added": sorted(set(str(x) for x in dependency_added_ids)),
        "effective_ids": effective,
    }


def check_scope_universe(
    audit: Mapping[str, Any],
    universe_scene_by_id: Mapping[str, int],
) -> None:
    """소비 직전 — audit 를 **현재 selected-still 우주**와 결합 검증.

    Codex 재재리뷰 HIGH-1: expected_scenes 대조만으로는 같은 씬 번호의
    stale audit(스틸 세대 교체 후 old id)가 로더를 통과하고, 소비측
    교집합이 0 이 되어 false CLEAN 이 가능했다. 잠금 2건:
    - 현재 우주에서 재구성한 표적 씬 selected id 집합 == audit.
      requested_ids (동등성 — 표적 씬 선택 샷 일부 누락도 검출)
    - audit.effective_ids ⊆ 현재 우주 (unknown dependency id 조용한
      드롭 차단)
    위반=ValueError fail-closed (표적 씬 이미지 재실행 필요 신호).
    universe_scene_by_id = 현재 selected·non-stale still id→scene_index.
    """
    targets = {int(x) for x in audit.get("requested_scenes") or []}
    expected_requested = sorted(
        sid for sid, si in universe_scene_by_id.items()
        if isinstance(si, int) and si in targets)
    if sorted(audit.get("requested_ids") or []) != expected_requested:
        raise ValueError(
            "표적 scope audit 이 현재 selected-still 우주와 불일치 — "
            "스틸 세대 교체/선택 변경 후 stale audit (fail-closed): "
            f"audit.requested={sorted(audit.get('requested_ids') or [])!r} "
            f"vs 현재 표적 씬 selected={expected_requested!r} — "
            "씬 이미지 재실행 필요")
    unknown = set(audit.get("effective_ids") or []) - set(
        universe_scene_by_id)
    if unknown:
        raise ValueError(
            "표적 scope audit effective 에 현재 우주 밖 still id — "
            f"unknown dependency (fail-closed): {sorted(unknown)[:5]!r}")


def scope_audit_path(
    projects_dir: str | Path, project_id: str, episode_id: str,
) -> Path:
    return (
        Path(projects_dir) / project_id / "checkpoints" / "images"
        / episode_id / AUDIT_FILENAME
    )


def save_scope_audit(path: Path, audit: Mapping[str, Any]) -> None:
    """atomic replace 저장 — 크래시 창에서 절단 파일이 fail-closed 로더를
    영구 막는 것 방지 (Codex 재리뷰 HIGH-3)."""
    import os

    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(".json.tmp")
    tmp.write_text(
        json.dumps(dict(audit), ensure_ascii=False, indent=1, sort_keys=True),
        encoding="utf-8",
    )
    os.replace(tmp, path)


def _require_id_list(data: Mapping[str, Any], key: str, path: Path) -> List[str]:
    vals = data.get(key)
    if not isinstance(vals, list) or any(
            not isinstance(x, str) or not x.strip() for x in vals):
        raise ValueError(
            f"표적 scope audit {key} 형상 오류 (fail-closed): {path}")
    if len(set(vals)) != len(vals):
        raise ValueError(
            f"표적 scope audit {key} 중복 id (fail-closed): {path}")
    return list(vals)


def load_scope_audit(
    path: Path, *, expected_scenes: Sequence[int],
) -> Dict[str, Any]:
    """audit 로드 — 부재/형상/정합 오류=ValueError fail-closed.

    표적이 설정된 스텝 카운트/verify 는 반드시 실행이 남긴 audit 과 동일
    scope 를 소비해야 한다(독자 필터 재구현 금지 — Codex HIGH-3). 검증:
    contract_version 정확 일치, path_kind enum, requested_scenes=현재
    정규화 표적과 정확 일치(다른 표적의 stale audit 소비 차단), id 목록=
    비공백 문자열·중복 없음, requested 비어 있지 않음, requested⊆effective,
    requested∪dependency_added==effective.
    """
    if not path.exists():
        raise ValueError(
            f"표적 scope audit 부재 (fail-closed): {path} — "
            "표적 설정 상태에서 씬 이미지 실행이 선행돼야 함")
    data = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(data, dict):
        raise ValueError(f"표적 scope audit 형상 오류 (fail-closed): {path}")
    if data.get("contract_version") != SCOPE_CONTRACT_VERSION:
        raise ValueError(
            "표적 scope audit contract_version 불일치 (fail-closed): "
            f"{data.get('contract_version')!r} != {SCOPE_CONTRACT_VERSION} "
            f"({path})")
    if data.get("path_kind") not in ("recipe", "batch"):
        raise ValueError(
            "표적 scope audit path_kind 오류 (fail-closed): "
            f"{data.get('path_kind')!r} ({path})")
    scenes = data.get("requested_scenes")
    expected_norm = sorted(set(int(x) for x in expected_scenes))
    if scenes != expected_norm:
        raise ValueError(
            "표적 scope audit 이 현재 표적과 불일치 — 다른 표적의 stale "
            f"audit 소비 차단 (fail-closed): audit={scenes!r} vs "
            f"현재={expected_norm!r} ({path}) — 씬 이미지 재실행 필요")
    requested = _require_id_list(data, "requested_ids", path)
    dep_added = _require_id_list(data, "dependency_added", path)
    effective = _require_id_list(data, "effective_ids", path)
    if not requested:
        raise ValueError(
            f"표적 scope audit requested_ids 빈 목록 (fail-closed): {path}")
    if not set(requested) <= set(effective):
        raise ValueError(
            f"표적 scope audit requested⊄effective (fail-closed): {path}")
    if set(requested) | set(dep_added) != set(effective):
        raise ValueError(
            "표적 scope audit requested∪dependency_added != effective "
            f"(fail-closed): {path}")
    return data