"""참조 묶음 **유료 canary** — provider 가 실제로 몇 장을 받나. ★`--live` 만 유료.

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

이 판은 **D 활성화가 아니다.** 「조립이 만든 참조가 provider 까지 **그대로**
가나」만 본다 —

    실제 입력 이미지 **수** · **차례** · **역할** · **metadata**
    장부 ↔ Opik ↔ provider raw **1:1**

## 왜 `generate_image` 를 **직접** 부르나

`generate_and_validate_scene` 은 moderation 최대 4회 + 보정 1회 길이 있어
**목적 밖 재생성**이 섞인다 (Codex). 이 canary 는 **장수·역할 전달**만 재므로
실제 조립 산출을 `GeminiImageClient.generate_image` 에 **바로** 넣는다.

## 문 세 겹

    ①`ImageCallBudget(cap=1)` 을 **호출 스레드에** 깐다
      ★`generate_image` 는 재시도마다 `reserve_current_call` 을 부른다 —
       두 번째 시도는 **예산에서 막힌다**. 「논리 1회」로는 못 막는다
    ②예산 문이 **장부·trace·provider 보다 앞**이다
    ③보내기 **직전**에 capability 를 **같은 callable** 로 다시 본다

★첫 raw 요청이 429·moderation·모름이면 **두 번째를 안 사고**
 `inconclusive` 로 끝낸다. **자동 재시도 없다.**
★이번 판 sidecar 좌표는 **`source=file` 만** — asset loader 가 아직 없다.

    python tools/grounding_audit/bundle_canary_runner.py --dry  <out_dir>
    python tools/grounding_audit/bundle_canary_runner.py --live <out_dir>  # ★유료
"""
from __future__ import annotations

import base64
import contextlib
import hashlib
import json
import sys
import uuid
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"))

from tools.grounding_audit import bundle_canary_preflight as pf  # noqa: E402

#: ★raw provider 호출 상한. **논리 1회가 아니라 raw 1회**다.
RAW_CALL_CAP = 1
#: ★이 판이 승인받은 **논리** 구매 횟수. 장부가 재개를 이 수로 잡는다.
APPROVED_LOGICAL = 1

#: 장부에 적히는 **끝난 갈래** 셋. ★사람에게 고치라고 요구하지 않는다.
JS_OK = "ok"
#: **나갔는데** 결과를 모른다 — 자동 재구매 0. 사람이 Opik 을 보면 된다.
JS_UNKNOWN = "unknown_not_retried"
#: **한 번도 안 나갔다** — 자리를 놓는다. 고치고 그대로 다시 돌리면 된다.
JS_NOT_DISPATCHED = "definitely_not_dispatched"
#: 이번 판이 허용하는 좌표 갈래. ★asset loader 가 아직 없다.
ALLOWED_SOURCES = ("file",)

TRACE_NAME = "grounding_bundle_canary"
TRACE_TAG = "op:bundle-canary"
TRACE_THREAD = "bundle-canary"


class CanaryRefused(RuntimeError):
    """문에 걸렸다. ★아무것도 안 사고 선다."""


def assert_sources_allowed(members: Sequence[Dict[str, Any]]) -> None:
    """이번 판은 **파일 좌표만**. ★asset 갈래는 loader 가 없다."""
    bad = [str(m.get("source") or "") for m in members
           if str(m.get("source") or "") not in ALLOWED_SOURCES]
    if bad:
        raise CanaryRefused(
            f"이번 canary 는 좌표 갈래 {ALLOWED_SOURCES} 만 받는다 — {bad} 가 "
            "왔다. `source=asset` loader 는 D cutover 에서 잇는다")


@contextlib.contextmanager
def provider_capture(into: Optional[Dict[str, Any]] = None):
    """provider 로 **정말 나간 것**을 본다. ★보고 흘려보낼 뿐, 안 바꾼다.

    Args:
        into: 적을 그릇. ★**부르는 쪽이 쥔다** — 안에서 새로 만들면 예외가
            났을 때 부르는 쪽이 그것을 **못 본다**. 앞 판이 그래서 실패한
            판의 관측을 통째로 잃었다 (Codex 재현 2026-08-31:
            `raw_counter=1` 인데 `result.provider_seen={}`).

    ★★「우리가 넘긴 것」과 「HTTP 몸통에 실린 것」은 다르다. 이 canary 의
    목적이 **장수·차례·역할 전달**이므로 **가장 바깥 문**에서 본다
    ([[feedback-test-the-exit-not-the-assembly]]).
    """
    import urllib.request

    seen: Dict[str, Any] = into if into is not None else {}
    seen.setdefault("calls", 0)
    seen.setdefault("image_shas", [])
    seen.setdefault("text_parts", [])
    real = urllib.request.urlopen

    def watching(req, *a, **kw):
        seen["calls"] += 1
        try:
            body = json.loads(bytes(getattr(req, "data", b"") or b"")
                              .decode("utf-8"))
            for c in body.get("contents") or []:
                for p in c.get("parts") or []:
                    inline = p.get("inlineData") or p.get("inline_data")
                    if isinstance(inline, dict) and inline.get("data"):
                        seen["image_shas"].append(hashlib.sha256(
                            base64.b64decode(inline["data"])).hexdigest())
                    elif p.get("text"):
                        seen["text_parts"].append(str(p["text"]))
        except Exception as exc:                    # noqa: BLE001
            # ★못 읽었다고 **「안 실렸다」로 적지 않는다**
            seen.setdefault("unread", []).append(f"{type(exc).__name__}: {exc}")
        return real(req, *a, **kw)

    urllib.request.urlopen = watching
    try:
        yield seen
    finally:
        urllib.request.urlopen = real


def canonical_identity(pre: Dict[str, Any]) -> str:
    """이 구매의 **신원**. ★`prompt_sha + N` 은 신원이 아니다 (Codex BLOCK C).

    나가는 **한 벌 전부** + sidecar 요구 + raw 상한을 접는다 — 이름표나
    역할만 바뀐 판이 **같은 신원으로 되쓰이면** 안 된다.
    """
    blob = json.dumps({"outbound": pre["lock"]["outbound"],
                       "sidecar_required": pre["lock"]["sidecar_required"],
                       "image_dispatch": pre["lock"]["image_dispatch"],
                       "raw_cap": RAW_CALL_CAP},
                      ensure_ascii=False, sort_keys=True, default=str)
    return hashlib.sha256(blob.encode("utf-8")).hexdigest()

def send_once(client: Any, *, prompt: str,
              labeled_refs: Sequence[Tuple[str, bytes]],
              approved_capability: Dict[str, Any],
              capture: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    """**한 번만** 보낸다. ★예산이 raw 호출을 센다.

    Args:
        approved_capability: **preflight 때 승인된** 값. ★지금 값을 다시 읽어
            넘기면 안 된다 — 그러면 current↔current 를 견주게 되어 preflight
            뒤 drift 가 **같은 새 값이면 그냥 통과**한다 (Codex BLOCK B).
        capture: 관측을 적을 그릇. ★**실패해도 여기 남는다** — 예외 갈래에서
            새 dict 로 덮으면 「아무것도 못 봄」이 된다 (Codex 2026-08-31).
    """
    from app.core.image_call_budget import (ImageCallBudget, install_budget,
                                            uninstall_budget)
    from app.modules.pipeline.grounding_reference_bundle import (
        assert_reference_count, capability_of)

    seen: Dict[str, Any] = capture if capture is not None else {}
    # ★③보내기 직전 재확인 — **승인된 것**과 **지금 것**을 견준다
    now = capability_of(client)
    if now != approved_capability:
        raise CanaryRefused(
            f"preflight 뒤 capability 가 바뀌었다: 승인 {approved_capability} "
            f"→ 지금 {now}. 승인 밖이므로 안 보낸다")
    assert_reference_count(now, len(list(labeled_refs)))

    budget = ImageCallBudget(cap=RAW_CALL_CAP)
    install_budget(budget)
    try:
        with provider_capture(seen):
            png, ms = client.generate_image(
                prompt=prompt,
                labeled_references=[(str(l), b) for l, b in labeled_refs])
        return {"ok": True, "bytes": len(png or b""), "elapsed_ms": ms,
                "budget": budget.snapshot(), "provider_seen": dict(seen)}
    except Exception as exc:                        # noqa: BLE001
        # ★★**두 번째를 안 산다.** 429·moderation·모름 어느 것이든
        #  `inconclusive` 로 끝낸다 — 자동 재시도 없다.
        #  ★관측은 **그대로 싣는다** — 실패한 판일수록 기록이 중요하다.
        return {"ok": False, "error": f"{type(exc).__name__}: {exc}",
                "budget": budget.snapshot(), "provider_seen": dict(seen),
                "verdict": "inconclusive"}
    finally:
        uninstall_budget()


def run(out_dir: Path, *, live: bool, plan: Dict[str, Any],
        approved_lock: Optional[Dict[str, Any]] = None,
        client: Any = None) -> Dict[str, Any]:
    """조립 산출을 받아 문 넷을 지나 **한 번** 보낸다.

    Args:
        plan: `{prompt, labeled_refs, ref_roles, ref_role_metadata,
            attached_meta, shot, sidecar_required, members}` —
            **실제 조립이 낸 것**. ★`attached_meta` 를 빼면 그 칸이 **빈 채로
            승인**된다 (Codex BLOCK A 재현).

    ★문 넷 —
        ①승인 잠금(`preflight`) ②부모 trace ③장부 재개(무중복) ④예산 cap=1

    ★②가 ③보다 **앞**이다. trace 가 안 열려 끝난 판이 장부에 자리를 남기면,
    다음 판이 막혀 **사람에게 장부를 고치라**고 요구하게 된다 — 자동화 정책과
    반대다 (Codex 2026-08-31).
    """
    from app.modules.llm.gemini_image_client import GeminiImageClient
    from app.modules.llm.opik_trace import open_trace
    from app.modules.pipeline.grounding_chunk_journal import (ChunkJournal,
                                                              buy_or_reuse)
    from tools.grounding_audit import cc_runner as rr

    out_dir.mkdir(parents=True, exist_ok=True)
    assert_sources_allowed(plan.get("members") or ())

    client = client if client is not None else GeminiImageClient()
    # ★①문 — 상한·잠금이 **장부·trace·provider 보다 앞**이다.
    #  ★`attached_meta` 를 **반드시** 넘긴다 (BLOCK A)
    pre = pf.preflight(
        prompt=plan["prompt"], labeled_refs=plan["labeled_refs"],
        ref_roles=plan["ref_roles"],
        ref_role_metadata=plan["ref_role_metadata"],
        attached_meta=plan["attached_meta"],
        shot=plan["shot"], sidecar_required=plan["sidecar_required"],
        approved_lock=approved_lock, client=client)
    approved_cap = pre["capability"]          # ★BLOCK B — **이것**을 보존한다

    jr = ChunkJournal(out_dir / "_canary_journal.json",
                      contract={"kind": "bundle_canary",
                                "raw_cap": RAW_CALL_CAP,
                                "lock": pre["lock"]})
    if jr.contract_drifted():
        raise CanaryRefused(
            "장부의 계약이 지금 계약과 다르다 — 이어 쓰지 않는다")
    ident = canonical_identity(pre)
    run_id = uuid.uuid4().hex[:16]

    if not live:
        got = {"ok": None, "verdict": "dry", "budget": None,
               "provider_seen": {"calls": 0, "image_shas": []}}
    else:
        jr.assert_no_uncertain()
        # ★★「나갔는데 결과를 모른다」가 남아 있으면 **여기서 끝낸다** —
        #  자동 재구매 0, 그리고 **사람에게 요구하지 않는다**. 그대로 두면
        #  `buy_or_reuse` 가 60초 기다렸다 사람을 부른다 (Codex 2026-08-31).
        prev = jr.entries.get(ident) or {}
        if prev.get("status") == JS_UNKNOWN and prev.get("epoch") == jr.epoch:
            got = {"ok": False, "verdict": "inconclusive",
                   "journal_status": JS_UNKNOWN, "budget": None,
                   "provider_seen": (prev.get("provider_seen") or {}),
                   "error": prev.get("why"), "not_retried": True}
            rec = {"run_id": run_id, "live": live, "identity": ident,
                   "preflight": pre, "result": got,
                   "journal": {"bought": jr.bought(), "reused": jr.reused()},
                   "acceptance": acceptance_of(pre, got, plan)}
            (out_dir / "canary.json").write_text(
                json.dumps(rec, ensure_ascii=False, indent=1, default=str),
                encoding="utf-8")
            return rec
        before = jr.bought()
        # ★②문 — 부모 trace 가 안 열리면 **자리도 안 잡고** 선다
        with open_trace(name=TRACE_NAME, tags=[TRACE_TAG],
                        metadata={rr.ID_META_KEY: ident,
                                  rr.RUN_META_KEY: run_id,
                                  "shot": pre["lock"]["outbound"]["shot"],
                                  "reference_count":
                                      pre["lock"]["reference_count"]},
                        thread_id=TRACE_THREAD,
                        input_data={"identity": ident}) as tr:
            trace_id = getattr(tr, "uid", None) if tr is not None else None
            if not trace_id:
                raise CanaryRefused(
                    "부모 Opik trace 를 못 열었다(또는 `uid` 가 없다) — 무엇을 "
                    "샀는지 못 되짚는다. **자리도 안 잡고** provider 앞에서 선다")
            # ★③문 — production helper 가 ①멈춤 ②자리잡기 ③구매를 한다
            got = buy_or_reuse(
                jr, ident, cap=APPROVED_LOGICAL,
                send=lambda: _buy_once(jr, client, plan, pre, ident, run_id,
                                       approved_cap, trace_id))
        if jr.bought() == before:
            got = {**got, "reused": True}
        else:
            got = _record(jr, ident, got, run_id, trace_id)

    rec = {"run_id": run_id, "live": live, "identity": ident,
           "preflight": pre, "result": got,
           "journal": {"bought": jr.bought(), "reused": jr.reused()},
           "acceptance": acceptance_of(pre, got, plan)}
    (out_dir / "canary.json").write_text(
        json.dumps(rec, ensure_ascii=False, indent=1, default=str),
        encoding="utf-8")
    if got.get("journal_status") == JS_NOT_DISPATCHED:
        raise CanaryRefused(
            f"한 번도 안 나갔다 — {got.get('error')}. 자리를 놓았으니 고친 뒤 "
            "그대로 다시 돌리면 된다 (사람이 장부를 고칠 일 없다)")
    return rec


def _record(jr, ident, got: Dict[str, Any], run_id, trace_id
            ) -> Dict[str, Any]:
    """장부에 **끝난 갈래**를 적는다. ★바깥 helper 가 덮은 것을 바로잡는다.

    ★`buy_or_reuse` 는 돌아온 것을 무조건 `ok` 로 적는다. 실패한 판을 그대로
    두면 다음 판이 그것을 **되쓴다**. 세 갈래로 갈라 적는다 —

        ok                       샀고 답을 받았다
        unknown_not_retried      **나갔는데** 결과를 모른다 — 자동 재구매 0
        definitely_not_dispatched **한 번도 안 나갔다** — 자리를 놓는다
    """
    st = got.get("journal_status") or JS_OK
    meta = {"run_id": run_id, "trace_id": trace_id,
            "provider_seen": (got.get("provider_seen") or {}),
            "why": got.get("error")}
    if st == JS_NOT_DISPATCHED:
        # ★안 샀다 — 자리를 놓아 **다음 판이 그냥 돌게** 한다.
        #  ★`buy_or_reuse` 가 방금 `ok` 로 적어 놨으므로 **끝난 갈래를 먼저
        #   적고** 놓는다. `release` 의 「산 것은 안 놓는다」 가드는 그대로 둔다 —
        #   가드를 느슨하게 하면 진짜 구매를 놓는 길이 열린다.
        jr.put(ident, None, status=st, meta=meta)
        jr.release(ident)
        return got
    jr.put(ident, got if st == JS_OK else None, status=st, meta=meta)
    return got


def _buy_once(jr, client, plan, pre, ident, run_id, approved_cap,
              trace_id) -> Dict[str, Any]:
    """자리를 잡은 뒤 **한 번** 보낸다. ★여기서는 **안 터진다**.

    ★★터뜨리면 바깥 `buy_or_reuse` 의 except 가 「샀는지 모른다」를 **다시
    써서** run_id·trace_id·까닭·관측을 **지운다** (Codex 2026-08-31). 그래서
    갈래를 여기서 정하고 **분류된 결과**를 돌려준다.

    ★부모 trace 는 자리를 잡기 **전에** 확인한다 — 그래야 trace 가 안 열려
    끝난 판이 장부에 자리를 남기지 않는다.
    """
    seen: Dict[str, Any] = {}
    try:
        got = send_once(client, prompt=plan["prompt"],
                        labeled_refs=plan["labeled_refs"],
                        approved_capability=approved_cap, capture=seen)
    except BaseException as exc:                    # noqa: BLE001
        got = {"ok": False, "error": f"{type(exc).__name__}: {exc}",
               "budget": None, "provider_seen": dict(seen)}
    got["trace_id"] = trace_id
    got["run_id"] = run_id
    # ★갈래는 **관측**이 정한다 — 한 번도 안 나갔나, 나갔는데 모르나
    calls = int((got.get("provider_seen") or {}).get("calls") or 0)
    if got.get("ok"):
        got["journal_status"] = JS_OK
    elif calls == 0:
        got["journal_status"] = JS_NOT_DISPATCHED
    else:
        got["journal_status"] = JS_UNKNOWN
    return got


def acceptance_of(pre: Dict[str, Any], got: Dict[str, Any],
                  plan: Dict[str, Any]) -> Dict[str, Any]:
    """**기계 계약**만 본다 — 눈검사는 진단이다.

        실제 입력 N장 · 각 role · 누락 0 · 예산 used=1 / denied=0

    ★★`missing` 은 **관측한 것**으로 센다. 앞 판은 상수 0 이었다 —
    그건 재는 척만 하는 것이다 (Codex BLOCK C).
    """
    ob = pre["lock"]["outbound"]
    b = got.get("budget") or {}
    seen = got.get("provider_seen") or {}
    want = list(ob["reference_shas"])
    sent = list(seen.get("image_shas") or [])
    text = "\n".join(seen.get("text_parts") or [])
    missing = [s[:12] for s in want if s not in sent]
    extra = [s[:12] for s in sent if s not in want]
    labels_absent = [l for l in ob["labels"] if l and l not in text]
    order_ok = sent == want
    return {
        "reference_count": pre["lock"]["reference_count"],
        "roles": ob["roles"],
        "labels": ob["labels"],
        "provider_raw_calls": seen.get("calls"),
        "provider_images": len(sent),
        "missing": missing,
        "extra": extra,
        "order_matches": order_ok,
        "labels_absent_from_text": labels_absent,
        "budget_used": b.get("used"),
        "budget_denied": b.get("denied"),
        "raw_cap": RAW_CALL_CAP,
        "passed": (bool(got.get("ok")) and b.get("used") == 1
                   and b.get("denied") == 0 and seen.get("calls") == 1
                   and not missing and not extra and order_ok
                   and not labels_absent),
    }


def main() -> int:
    if len(sys.argv) < 3 or sys.argv[1] not in ("--dry", "--live"):
        print(__doc__)
        return 2
    print("★조립 산출(plan)을 넘겨 `run()` 을 부르는 자리다 — "
          "고정 샷을 고르는 것은 다음 단계다.")
    return 0


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