"""참조 묶음 **유료 canary 의 문**. ★유료 0 — 이 파일은 아무것도 안 산다.

## 무엇을 잠그나 (Codex 2026-08-31)

    이미지 생성 **논리 dispatch 1회**        ← 참조 장수와 **별개**다
    고정 샷의 **exact N**                   ← 조립이 실제로 만든 수
    provider capability                     ← **client instance** 가 낸다
    plan / provider / model / base refs / sidecar 신원

★하나라도 바뀌거나 실제 N 이 다르면 **provider 앞에서 선다**.
★상한을 **발명하지 않는다** — `max_images` 가 `None` 이면 막지 않고
 observed N 을 **기록**한다.

## 왜 조립을 **직접** 부르나

helper 만 부르는 주행은 **우회 시험**이다 (Codex). 실제 길은

    generate_single_scene_image → _build_single_scene_prompt_and_refs
      → build_scene_attached_refs → validate_attached_refs → provider

이므로 preflight 도 **그 조립 함수**가 만든 것을 센다.
"""
from __future__ import annotations

import hashlib
import json
import sys
from pathlib import Path
from typing import Any, Dict, List, Optional, Sequence, Tuple

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "tests"))

#: ★**손으로 적은 승인 수.** 이미지 생성은 **한 번**이다.
APPROVED_IMAGE_DISPATCH = 1


class CanaryScopeMismatch(RuntimeError):
    """계획이 **잠근 것과 다르다**. ★아무것도 안 사고 선다."""


def _sha(b: bytes) -> str:
    return hashlib.sha256(b).hexdigest()


def outbound_of(*, prompt: str, labeled_refs: Sequence[Tuple[str, Any]],
                ref_roles: Sequence[str],
                ref_role_metadata: Sequence[Any],
                attached_meta: Sequence[Any],
                shot: Dict[str, Any], capability: Dict[str, Any]
                ) -> Dict[str, Any]:
    """**실제로 나가는 한 벌**. ★잠금·장부·Opik 이 **이것만** 소비한다.

    ★★★필드를 preflight 와 runner 에 **다시 나열하면 두 벌**이 된다
    (Codex 2026-08-31). 그래서 한 자리에서 만든다.

    ★앞 판 잠금은 bytes sha·roles·attached·capability·required 뿐이라,
    **label·role metadata·샷 좌표·prompt** 가 바뀐 판이 **같은 승인으로
    통과**했다. 이 canary 의 목적이 provider args 의 N·차례·role·metadata
    대조이므로 그것들이 다 들어가야 한다.

    ★`attached_meta` 와 `ref_role_metadata` 는 production 에서 **다른 것**이다
    (`scene_generation_coordinator` 가 둘을 따로 들고 다닌다) — 둘 다 잠근다.
    """
    from app.modules.pipeline.grounding_reference_bundle import capability_lock

    def _fold(x):
        return json.loads(json.dumps(list(x), ensure_ascii=False,
                                     sort_keys=True, default=str))

    return {
        "prompt_sha": _sha(str(prompt).encode("utf-8")),
        "prompt_chars": len(str(prompt)),
        # ★**차례 그대로** — 이름표와 사진이 짝지어 간다
        "labels": [str(lbl) for lbl, _b in labeled_refs],
        "reference_shas": [_sha(b) for _lbl, b in labeled_refs],
        "roles": [str(r) for r in ref_roles],
        "role_metadata": _fold(ref_role_metadata),
        "attached": _fold(attached_meta),
        "shot": {k: str(shot.get(k) or "")
                 for k in ("project_id", "episode_id", "scene_index",
                           "shot_index", "still_id")},
        "capability": capability_lock(capability),
    }


def lock_of(*, outbound: Dict[str, Any],
            sidecar_required: Sequence[Tuple[str, str]]) -> Dict[str, Any]:
    """이 판을 **무엇으로 사는가**. ★`outbound_of` 가 낸 **한 벌**을 받는다.

    ★참조 **bytes 자체**·**이름표**·**역할 metadata**·**샷 좌표**·**프롬프트**
    가 다 들어간다 — 하나라도 바뀌면 다른 판이다.
    """
    return {
        "outbound": dict(outbound),
        "reference_count": len(outbound.get("reference_shas") or []),
        "sidecar_required": [list(x) for x in sidecar_required],
        "image_dispatch": APPROVED_IMAGE_DISPATCH,
    }


def assert_lock(now: Dict[str, Any], approved: Optional[Dict[str, Any]]
                ) -> None:
    """잠근 것과 지금이 같나. ★다르면 **provider 앞에서** 선다."""
    if approved is None:
        return
    if now == approved:
        return
    diff = [k for k in set(now) | set(approved)
            if now.get(k) != approved.get(k)]
    raise CanaryScopeMismatch(
        f"승인 뒤 바뀐 것: {sorted(diff)} — 사람이 다시 정해야 한다. "
        f"지금 {json.dumps({k: now.get(k) for k in diff}, ensure_ascii=False)[:300]}")


def assert_count_ok(capability: Dict[str, Any], n: int) -> Dict[str, Any]:
    """참조 장수가 provider 능력 안인가. ★**같은 method** 를 소비한다.

    Returns:
        기록할 것 — `{"observed": n, "declared_max": ...}`.
        ★`declared_max` 가 `None` 이면 「모른다」다. 상한을 **발명하지 않고**
        observed 를 남긴다.
    """
    from app.modules.pipeline.grounding_reference_bundle import (
        assert_reference_count)

    assert_reference_count(capability, n)
    return {"observed": n, "declared_max": capability.get("max_images"),
            "declared_min": capability.get("min_images"),
            "provider": capability.get("provider"),
            "model": capability.get("model")}


def scene_image_client() -> Any:
    """씬 이미지가 **실제로 쓰는** client. ★설정 사본을 안 만든다.

    ★실측 (2026-08-31): 씬 이미지는 `GeminiImageClient` 를 쓰고, cine 변환은
    `cine_provider`(grok|reve) 를 쓴다 — **다른 경로**다. 묶음은 씬 이미지
    경로에 실리므로 여기서 읽는다.
    """
    from app.modules.llm.gemini_image_client import GeminiImageClient

    return GeminiImageClient()


def preflight(*, prompt: str = "", labeled_refs, ref_roles,
              ref_role_metadata: Sequence[Any] = (),
              attached_meta: Sequence[Any] = (),
              shot: Optional[Dict[str, Any]] = None, sidecar_required,
              approved_lock: Optional[Dict[str, Any]] = None,
              client: Any = None) -> Dict[str, Any]:
    """굽기 **직전**의 문. ★아무것도 안 산다.

    ★`client` 를 주면 그것에게 묻는다 — **보내기 직전과 같은 instance** 를
    넘겨야 두 벌이 안 된다.

    Returns:
        `capability` 는 **이때 승인된 raw 값**이다. ★보내기 직전에 이것과
        견줘야 한다 — 그때 다시 읽은 값끼리 견주면 drift 를 못 잡는다
        (Codex BLOCK B 2026-08-31).
    """
    from app.modules.pipeline.grounding_reference_bundle import capability_of

    cap = capability_of(client if client is not None else scene_image_client())
    n = len(list(labeled_refs))
    count = assert_count_ok(cap, n)
    outbound = outbound_of(prompt=prompt, labeled_refs=labeled_refs,
                           ref_roles=ref_roles,
                           ref_role_metadata=ref_role_metadata,
                           attached_meta=attached_meta,
                           shot=dict(shot or {}), capability=cap)
    lock = lock_of(outbound=outbound, sidecar_required=sidecar_required)
    assert_lock(lock, approved_lock)
    return {"lock": lock, "count": count, "outbound": outbound,
            "capability": cap, "ok": True}


def main() -> int:
    print(__doc__)
    print("★이 파일은 문(preflight)이다 — 굽는 것은 canary runner 가 한다.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())


def capability_of_for(client):
    """★preflight 와 **같은 함수**를 부른다 — runner 가 다시 안 적게."""
    from app.modules.pipeline.grounding_reference_bundle import capability_of

    return capability_of(client)


# ─────────────────────────────────────────────────────────────────────
# ★★★대상 샷은 **구조 조건만으로 결정적으로** 고른다 (Codex 2026-08-31)
#
# > 사람에게 후보를 골라 달라고 하지 마십시오(HITL 금지).
# > 0개면 억지 후보를 만들거나 다른 원고 자산을 섞지 말고 provider 0 으로
# > 「eligible shot 없음」을 보고한 뒤, 별도 명시 canary scenario 로 갑니다.
#
# 조건 다섯 —
#   ①같은 project/episode 의 **실제** 샷
#   ②기존 base ref 가 있다
#   ③context+detail selected sidecar 둘이 **실제로 조립된다**
#   ④좌표가 `source=file` 이고 해시가 **전부** 맞는다
#   ⑤provider capability 안의 **exact N**
# ─────────────────────────────────────────────────────────────────────

#: 정본 정렬 키. ★복수면 이 차례의 **첫 행**이다 — 사람이 안 고른다.
SHOT_SORT_KEYS = ("project_id", "episode_id", "scene_index", "shot_index",
                  "still_id")


def _scene_detail_manifests(projects_dir):
    """production 과 **같은 자리**를 본다 (`lookup_render_prompt_card`)."""
    from pathlib import Path

    root = Path(projects_dir)
    if not root.is_dir():
        return []
    return sorted(root.glob(
        "*/checkpoints/episodes/*/scene_detail/manifest.json"))


def eligible_shots(*, projects_dir=None, capability=None,
                   load_bytes=None, assemble=None) -> Dict[str, Any]:
    """조건 일곱을 다 지나는 샷을 **전부** 찾는다. ★아무것도 안 산다.

    Args:
        assemble: `(candidate) -> plan|None`. **실제 production 조립**을 무료로
            태우는 callable. ★없으면 `chosen` 을 **안 낸다** — selector 가 센
            N 과 유료 runner 가 보내는 N 이 **같은 근거**여야 한다
            (Codex 2026-08-31). `len(required_refs)+len(plan)` 은 **셈**이지
            조립이 아니다.

    ★조건 ⓪ `status=completed` · `failed_count=0`. 앞 판은 이것을 안 봐서
     **부분 실패한 manifest** 도 sidecar 만 있으면 자격 샷이 됐다.
    """
    import hashlib
    import json as _json
    from pathlib import Path

    from app.core.config import settings
    from app.modules.pipeline import grounding_reference_bundle as gb

    projects_dir = projects_dir or settings.projects_dir
    cap = (capability if capability is not None
           else capability_of_for(scene_image_client()))

    def _read(coord):
        p = Path(str(coord.get("path") or ""))
        if not p.is_file():
            return b"", {}
        return p.read_bytes(), {"source": gb.SOURCE_FILE, "path": str(p)}

    load = load_bytes or _read
    seen, rejected = [], {}

    def _no(why):
        rejected[why] = rejected.get(why, 0) + 1

    for mf in _scene_detail_manifests(projects_dir):
        try:
            data = _json.loads(mf.read_text(encoding="utf-8"))
        except Exception as exc:                    # noqa: BLE001
            _no(f"manifest 를 못 읽었다: {type(exc).__name__}")
            continue
        # ★⓪ — 끝난 판인가. **fail-closed** 로 본다
        if str(data.get("status") or "") != "completed":
            _no(f"⓪status 가 completed 가 아님: {data.get('status')!r}")
            continue
        if int(data.get("failed_count") or 0) != 0:
            _no(f"⓪failed_count 가 0 이 아님: {data.get('failed_count')!r}")
            continue
        pid, eid = mf.parts[-6], mf.parts[-3]
        for row in (data.get("data") or {}).get("scenes") or []:
            rpc = row.get("render_prompt_card") or row
            # ★③ — sidecar 가 **실제로** 있나. 없으면 D 가 아직 inert 인 것
            try:
                members = gb.members_from_rpc(rpc)
            except Exception as exc:                # noqa: BLE001
                _no(f"sidecar 모양이 계약 밖: {type(exc).__name__}")
                continue
            if not members:
                _no("③sidecar 없음 (D 가 아직 inert)")
                continue
            picked = [m for m in members
                      if str(m.get("outcome")) == gb.OUTCOME_SELECTED]
            purposes = {str(m.get("purpose")) for m in picked}
            if not {"context", "detail"} <= purposes:
                _no("③context+detail 둘이 아님")
                continue
            # ★② — 기존 base ref
            req = ((rpc.get("asset_requirements") or {}).get("required_refs")
                   or [])
            if not req:
                _no("②기존 base ref 없음")
                continue
            # ★④ — 파일 좌표 · 해시 전부
            bad = False
            for m in picked:
                if str(m.get("source")) != gb.SOURCE_FILE:
                    _no("④file 좌표가 아님")
                    bad = True
                    break
                raw, _r = load(gb.coordinate_of(m))
                if not raw or hashlib.sha256(raw).hexdigest() != str(
                        m.get("content_sha256") or ""):
                    _no("④해시가 안 맞거나 못 읽음")
                    bad = True
                    break
            if bad:
                continue
            plan, _required = gb.plan_bundle(picked)
            n = len(req) + len(plan)
            # ★⑤ — provider 가 받는 장수 안
            try:
                gb.assert_reference_count(cap, n)
            except Exception:                       # noqa: BLE001
                _no(f"⑤capability 밖 (N={n})")
                continue
            seen.append({"project_id": pid, "episode_id": eid,
                         "scene_index": row.get("scene_index"),
                         "shot_index": row.get("_shot_index"),
                         "still_id": str(row.get("still_id") or ""),
                         "counted_reference_count": n})

    seen.sort(key=lambda r: tuple(str(r.get(k) or "") for k in SHOT_SORT_KEYS))
    out = {"eligible": seen, "chosen": None, "assembled": None,
           "manifests_read": len(_scene_detail_manifests(projects_dir)),
           "rejected_because": rejected}
    if not seen:
        return out
    if assemble is None:
        out["why_no_choice"] = (
            "⑥실제 조립을 안 태웠다 — 센 장수만으로는 유료 runner 가 보낼 "
            "장수와 **같은 근거**가 아니다. `assemble=` 을 넘겨야 고른다")
        return out
    # ★⑥ — **실제 production 조립**이 되는 첫 행을 고른다. 셈이 아니라 조립이다
    for cand in seen:
        try:
            got = assemble(cand)
        except Exception as exc:                    # noqa: BLE001
            _no(f"⑥조립이 섰다: {type(exc).__name__}: {exc}")
            continue
        if not got:
            _no("⑥조립이 빈손")
            continue
        made = len(got.get("labeled_refs") or ())
        if made != cand["counted_reference_count"]:
            _no(f"⑥센 장수 {cand['counted_reference_count']} ≠ 조립 {made}")
            continue
        out["chosen"] = {**cand, "reference_count": made}
        out["assembled"] = got
        return out
    return out