"""canary ② — **이미지 생성 직전 payload** 에 참조가 어떻게 실리나.

    python tools/grounding_audit/canary_payload_probe.py \
        --project-id … --episode-id … --out artifact/<날짜_설명>

★★★이 도구는 **이미지 provider 를 안 부른다.** 인자로도 안 받는다.
 새 검색 0 · 새 받기 0 · VLM 0 · 최종 이미지 생성 0 — 재는 것은 **이미 산
 것이 어떻게 실리나**뿐이다 (Codex 승인 범위 2026-09-02).

★production 함수만 지난다 —

    central_cp_with_reviews   사람 판정을 얹은 중앙 CP (읽기 전용 투영)
    write_for_shot            그 컷의 sidecar 를 RPC 에 적는다
    attach_from_rpc           **좌표로 사진을 읽어 payload 에 넣는 그 자리**

재는 것: 샷별 참조 **수 · role · 순서 · bytes/SHA · subject 신원**.
"""
from __future__ import annotations

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

BACKEND = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(BACKEND))

PROBE_CONTRACT_VERSION = "1.202609021000"


class VisibilityMissing(RuntimeError):
    """샷별 가시성의 SOT 가 없다. ★**내가 만들어 넣지 않는다**."""


class ProbeRefused(RuntimeError):
    """이 판이 재려는 것과 다른 것을 잴 뻔했다. ★안 재고 선다."""


def shots_from(cp: Optional[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """샷별 **가시 목록**. ★`shot_director` 가 SOT 다 — 씬 단위로 안 내린다.

    ★★★없으면 **선다** (Codex 조건 3 · 2026-09-02): 「visibility 의 답을
    따르라」인데 그 답이 없으면 내가 지어내는 수밖에 없다. 그것은 재는 것이
    아니라 만드는 것이다.
    """
    scenes = ((cp or {}).get("data") or {}).get("scenes")
    if not scenes:
        raise VisibilityMissing(
            "`shot_director` 체크포인트가 없다 — 샷별 가시성의 SOT 다. "
            "씬 단위 `present_entity_ids` 로 대신하면 그것은 **다른 것을 "
            "재는 것**이고, 내가 목록을 만들어 넣으면 답을 지어내는 것이다")
    out: List[Dict[str, Any]] = []
    for sc in scenes:
        for sh in (sc.get("shots") or ()):
            out.append({
                "scene_index": sc.get("scene_index"),
                "shot_index": sh.get("shot_index"),
                "visible": list(sh.get("visible_entity_ids") or ()),
            })
    if not out:
        raise VisibilityMissing("`shot_director` 에 샷이 하나도 없다")
    return out


def measure_shot(central_cp: Dict[str, Any], visible: Sequence[str], *,
                 root: Path) -> Dict[str, Any]:
    """이 컷에서 **payload 에 실제로 들어가는 것**. ★production 경계 그대로.

    Returns:
        `{written, refs: [{order, sha256, bytes, role, subjects, purposes}]}`
    """
    from app.modules.pipeline import grounding_reference_bundle as rb
    from app.modules.pipeline import grounding_sidecar_writer as sw

    # ★production 과 **같은 helper** (2026-09-02: probe 가 제 손으로 sha 를 세어
    #  production 의 `content_sha256` 칸 결함을 못 봤다 — 조립부 측정)
    def _sha_of(row):
        return sw.row_content_sha256(row, root=root)

    def _coord_of(row):
        return sw.row_file_coordinate(row, root=root)

    rpc: Dict[str, Any] = {}
    written = sw.write_for_shot(rpc, central_cp, list(visible),
                                content_sha_of=_sha_of,
                                coordinate_of=_coord_of)

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

    labeled: List[Any] = []
    meta: List[Any] = []
    roles: List[Any] = []
    role_meta: List[Any] = []
    added = rb.attach_from_rpc(labeled, meta, roles, role_meta, rpc,
                               load_bytes=_load, where="probe")
    # ★붙는 쪽이 넘긴 목록이 아니라 **RPC 에 적힌 것**과 대조한다
    rb.assert_from_rpc(rpc, meta)

    # ★네 목록의 **실제 모양**을 그대로 읽는다 (production 이 넣는 대로):
    #   labeled_refs[i]       = (label, bytes)
    #   attached_meta[i]      = (ref_kind, meta_value)
    #   ref_roles[i]          = role
    #   ref_role_metadata[i]  = {subject_final_ids, purposes, content_sha256…}
    #  ★★모양을 짐작해서 `.get` 을 부르면 tuple 에서 죽는다(실제로 죽었다).
    refs = []
    for i, blob in enumerate(labeled):
        label, raw = ("", b"")
        if isinstance(blob, (tuple, list)) and len(blob) == 2:
            label, raw = str(blob[0] or ""), (blob[1] or b"")
        elif isinstance(blob, (bytes, bytearray)):
            raw = bytes(blob)
        rm = role_meta[i] if i < len(role_meta) else {}
        kind = ""
        if i < len(meta) and isinstance(meta[i], (tuple, list)) \
                and meta[i]:
            kind = str(meta[i][0] or "")
        refs.append({
            "order": i,
            "label": label,
            "ref_kind": kind,
            "bytes": len(raw),
            "sha256": hashlib.sha256(bytes(raw)).hexdigest() if raw else "",
            "role": roles[i] if i < len(roles) else None,
            "subjects": list((rm or {}).get("subject_final_ids") or ()),
            "purposes": list((rm or {}).get("purposes") or ()),
            # ★sidecar 가 **적은** 지문 — 위 `sha256` 은 **실린 bytes** 의 것
            "declared_sha256": str((rm or {}).get("content_sha256") or ""),
        })
    return {"written": int(written), "attached": int(added), "refs": refs}


def _subject_triples(items) -> List[Dict[str, Any]]:
    """감사 줄 → `{research_subject_id, final_id, purpose, fidelity, outcome}`.

    ★감사 줄에는 purpose 가 없어 subject id 의 `#` 뒤를 **그대로 옮긴다**
    (뜻 해석이 아니라 production 이 발급한 id 의 꼴이다). 없으면 빈 칸.
    """
    out: List[Dict[str, Any]] = []
    for a in items or ():
        sid = str(a.get("research_subject_id") or "")
        out.append({"research_subject_id": sid,
                    "final_id": a.get("final_id"),
                    "purpose": sid.split("#", 1)[1] if "#" in sid else "",
                    "fidelity": a.get("fidelity"),
                    "outcome": a.get("outcome")})
    return out


def rows_by_subject(cp: Dict[str, Any]) -> Dict[str, Dict[str, Any]]:
    """중앙 CP 의 줄을 **subject 단위**로 적는다.

    ★같은 `final_id` 가 purpose 별로 여러 줄이다(`LP01#detail` 확인됨,
    `LP01#context` 건너뜀). final_id 로 키를 잡으면 뒤 줄이 앞 줄을 **덮어**
    확인된 것이 「미확인」으로 읽힌다 — canary ② 에서 실제로 그랬다.
    """
    from app.modules.pipeline import reference_acquisition as ra

    out: Dict[str, Dict[str, Any]] = {}
    for r in ((cp.get("data") or {}).get("rows") or ()):
        led = r.get("ledger_row") or {}
        fid = str(led.get("final_id") or "")
        if not fid:
            continue
        purpose = str(led.get("purpose") or "")
        sid = str(r.get("research_subject_id") or led.get("research_subject_id")
                  or f"{fid}#{purpose}")
        out[sid] = {
            "final_id": fid,
            "purpose": purpose,
            "owner_type": str(led.get("owner_type") or ""),
            "outcome": r.get("outcome"),
            "fidelity": ra.fidelity_of(r),
            "usable": ra.usable_as_reference(r),
        }
    return out


def run(*, project_id: str, episode_id: str, db: Any,
        root: Optional[Path] = None) -> Dict[str, Any]:
    """샷마다 재고 **한 판의 산출**로 낸다. ★provider 는 안 부른다."""
    from app.core.config import settings
    from app.core.steps.episode_reference_policy_step import (
        EpisodeReferencePolicyStep as Policy)
    from app.modules.pipeline import grounding_fidelity_review as fr
    from app.modules.pipeline import reference_acquisition as ra

    base = Path(root or Path(settings.projects_dir).parent)
    s = Policy.__new__(Policy)
    s.project_id, s.episode_id, s.db = project_id, episode_id, db
    s.project_config = {"grounding_mode": "v2_chunk"}

    cp = fr.central_cp_with_reviews(s, required=True)
    director = s._load_prev_checkpoint("shot_director")
    shots = shots_from(director)

    per_shot = []
    for one in shots:
        got = measure_shot(cp, one["visible"], root=base)
        per_shot.append({**one, **got})

    # ★정책 lane 은 **따로** 센다 — 두 lane 에 겹쳐 붙으면 안 된다
    forced, blocked, audit = s._central_forced_short_ids()

    seen_rows = rows_by_subject(cp)
    sidecar_lane = sorted({sid for sh in per_shot for ref in sh["refs"]
                           for sid in ref["subjects"]})
    return {
        "contract": PROBE_CONTRACT_VERSION,
        "project_id": project_id, "episode_id": episode_id,
        "reviews_digest": fr.reviews_digest_for(s),
        "rows": seen_rows,
        "shots": per_shot,
        "sidecar_lane": sidecar_lane,
        "policy_lane": sorted(forced),
        "policy_blocked": sorted(blocked),
        # ★final_id 로만 펴면 `LP01#context unavailable` 과 `LP01#detail
        #  selected` 가 같은 LP01 로 보인다 — subject 단위로 남긴다
        "fidelity_unverified": _subject_triples(
            audit.get("fidelity_unverified")),
        "fidelity_rejected": _subject_triples(audit.get("fidelity_rejected")),
        "reference_unavailable": _subject_triples(
            audit.get("reference_unavailable")),
        "★means": ("이미지 생성 **직전** payload 에 실리는 것만 쟀다. "
                   "provider 는 안 불렀다"),
    }


def verdicts(got: Dict[str, Any]) -> List[Dict[str, Any]]:
    """받아들일 축들. ★어긋남·미확정·합격을 **갈라** 센다."""
    rows = got["rows"]
    sidecar = set(got["sidecar_lane"])
    policy = set(got["policy_lane"])
    # ★final_id 하나에 줄이 여럿(purpose 별). 「확인된 줄이 **하나라도**
    #  있나」로 갈라야 건너뛴 context 줄이 확인된 detail 줄을 안 가린다.
    owner_of: Dict[str, str] = {}
    usable: set = set()
    for key, r in rows.items():
        fid = str(r.get("final_id") or key)
        owner_of.setdefault(fid, r["owner_type"])
        if r["usable"]:
            usable.add(fid)
    bg_usable = {fid for fid in usable
                 if owner_of[fid] in ("location", "location_part")}
    prop_usable = {fid for fid in usable if owner_of[fid] == "prop"}
    not_usable = set(owner_of) - usable
    # ★참조는 **그 subject 가 보이는 샷**에만 붙는다. 확인됐어도 shot_director
    #  가 어느 샷에도 보이게 두지 않은 것은 실리지 않는 게 맞다 — 그것은
    #  production visibility 의 데이터 답이지 결함이 아니다(억지로 붙이지 않는다).
    #  다만 그 목록은 **따로 적는다** — 「안 보였다」와 「못 실었다」를 한 칸에
    #  넣지 않는다.
    visible_any = {str(v) for sh in got["shots"]
                   for v in (sh.get("visible") or ())}
    bg_expected = bg_usable & visible_any
    out = [
        {"축": "확인된 배경 갈래 중 샷에 보이는 것은 sidecar 에 실린다",
         "기대": sorted(bg_expected), "실제": sorted(sidecar & bg_usable),
         "통과": bg_expected == (sidecar & bg_usable) and bool(bg_expected),
         "확인됐지만 어느 샷에도 안 보여 안 실림": sorted(bg_usable - visible_any)},
        {"축": "확인 안 된 것은 payload 에 **0**",
         "기대": [], "실제": sorted(sidecar & not_usable),
         "통과": not (sidecar & not_usable)},
        {"축": "prop 은 정책 lane 으로만",
         "기대": sorted(prop_usable), "실제": sorted(policy),
         "통과": policy == prop_usable},
        {"축": "두 lane 에 겹쳐 붙지 않는다",
         "기대": [], "실제": sorted(sidecar & policy),
         "통과": not (sidecar & policy)},
    ]
    # ★같은 사진을 쓰는 것끼리 **합쳐지지 않는다**
    by_sha: Dict[str, List[str]] = {}
    for sh in got["shots"]:
        for ref in sh["refs"]:
            by_sha.setdefault(ref["sha256"], []).extend(ref["subjects"])
    leaked = sorted({s for names in by_sha.values() for s in names
                     if s in not_usable})
    out.append({"축": "같은 bytes 를 쓰는 거절/미확인이 **새지 않는다**",
                "기대": [], "실제": leaked, "통과": not leaked})
    return out


def main() -> int:
    import argparse

    ap = argparse.ArgumentParser(add_help=True)
    ap.add_argument("--project-id", required=True)
    ap.add_argument("--episode-id", required=True)
    ap.add_argument("--database-url", required=True)
    ap.add_argument("--projects-dir", required=True)
    ap.add_argument("--out", required=True)
    a = ap.parse_args()

    import os

    os.environ["DATABASE_URL"] = a.database_url
    os.environ["PROJECTS_DIR"] = a.projects_dir
    from sqlalchemy import create_engine
    from sqlalchemy.orm import Session

    from app.core.config import settings

    settings.projects_dir = a.projects_dir
    eng = create_engine(a.database_url)
    with Session(eng) as db:
        got = run(project_id=a.project_id, episode_id=a.episode_id, db=db,
                  root=Path(a.projects_dir).parent)
    got["verdicts"] = verdicts(got)
    out = Path(a.out)
    out.mkdir(parents=True, exist_ok=True)
    (out / "payload_probe.json").write_text(
        json.dumps(got, ensure_ascii=False, indent=1), encoding="utf-8")

    print(f"■ canary ② · 판 {PROBE_CONTRACT_VERSION} · 판정 지문 "
          f"{got['reviews_digest']}")
    print(f"  샷 {len(got['shots'])}개 · sidecar {got['sidecar_lane']} · "
          f"정책 {got['policy_lane']}")
    bad = [v for v in got["verdicts"] if not v["통과"]]
    for v in got["verdicts"]:
        print(f"  {'○' if v['통과'] else '✗'} {v['축']} — 기대 {v['기대']} · "
              f"실제 {v['실제']}")
    print(f"  적었다: {out / 'payload_probe.json'}")
    return 1 if bad else 0


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