"""유료 canary 의 **raw 1회 문**. ★유료 0 — 바깥 호출은 하나도 안 나간다.

Codex BLOCK 3 (2026-08-31) —

> 「logical dispatch 1」은 raw provider gate 가 아닙니다.
> `GeminiImageClient.generate_image` 는 재시도마다 `reserve_current_call` 을
> 부르고(`gemini_image_client:304-306`), `generate_and_validate_scene` 에는
> moderation ≤4 + 보정 1 이 더 있습니다. 실제 `build_scene_attached_refs`
> 산출을 **직접** `generate_image` 에 먹이고 `ImageCallBudget(cap=1)` 을
> 호출 스레드에 깐 뒤, 첫 raw 실패는 **재시도 없이** inconclusive 로 끝내십시오.

★★그래서 이 시험은 **진짜 `generate_image` 를 부른다.** 재시도 loop 가
거기 있으므로, 흉내낸 client 로는 「두 번째를 안 산다」를 못 잰다
([[feedback-test-the-exit-not-the-assembly]]).
"""
from __future__ import annotations

import base64
import io
import json
import urllib.error
import urllib.request

import pytest

from tools.grounding_audit import bundle_canary_runner as rn

PNG = b"\x89PNG\r\n\x1a\n" + b"0" * 64


def _plan(**over):
    """조립이 낸 모양. ★`source=file` 만 — asset loader 는 아직 없다."""
    p = {
        "prompt": "한 컷",
        "labeled_refs": [("배경판", b"plate"), ("상세", b"det")],
        "ref_roles": ["background_chain_ref", "grounding_part_detail_ref"],
        "ref_role_metadata": [{"pipeline_role": "background_render"},
                              {"purposes": ["detail"]}],
        "attached_meta": [("background", "L01B01"),
                          ("background", "grounding:LP01:detail:y")],
        "shot": {"project_id": "p", "episode_id": "e", "scene_index": 8,
                 "shot_index": 4, "still_id": "s"},
        "sidecar_required": [("background", "grounding:LP01:detail:y")],
        "members": [{"source": "file", "path": "/x/a.png"}],
    }
    p.update(over)
    return p


class _Counter:
    """urlopen 을 **몇 번** 불렀나. ★진짜 raw 자리를 센다."""

    def __init__(self, behave):
        self.n = 0
        self._behave = behave

    def __call__(self, req, timeout=None):
        self.n += 1
        return self._behave(self.n)


def _ok(_n):
    body = json.dumps({"candidates": [{"content": {"parts": [
        {"inlineData": {"data": base64.b64encode(PNG).decode(),
                        "mimeType": "image/png"}}]}}]}).encode()

    class _R:
        def read(self):
            return body

        def __enter__(self):
            return self

        def __exit__(self, *a):
            return False

    return _R()


def _rate_limited(_n):
    raise urllib.error.HTTPError("u", 429, "too many", {},
                                 io.BytesIO(b'{"error":"quota"}'))


@pytest.fixture
def no_sleep(monkeypatch):
    """재시도 대기는 안 잰다 — **횟수**를 잰다."""
    import time as _t

    monkeypatch.setattr(_t, "sleep", lambda *_a, **_k: None)


@pytest.fixture
def retries_on(monkeypatch):
    """★재시도를 **켠다**. production 은 켜져 있다 — 그 자리를 재야 한다."""
    from app.modules.llm import gemini_image_client as gic

    monkeypatch.setattr(gic._settings, "llm_max_retries", 3, raising=False)
    assert gic._settings.llm_max_retries >= 1


@pytest.fixture
def client(monkeypatch):
    from app.modules.llm.gemini_image_client import GeminiImageClient

    monkeypatch.setenv("GEMINI_API_KEY", "테스트용-가짜-키")
    c = GeminiImageClient()
    monkeypatch.setattr(c, "_get_api_key", lambda *a, **k: "가짜")
    return c


def _wire(monkeypatch, behave):
    counter = _Counter(behave)
    monkeypatch.setattr(urllib.request, "urlopen", counter)
    return counter


class TestOnlyOneRawCallCanHappen:
    def test_a_success_uses_exactly_one(self, client, monkeypatch):
        got = _wire(monkeypatch, _ok)
        r = rn.send_once(client, prompt="한 컷",
                         labeled_refs=[("배경판", b"a"), ("상세", b"b")],
                         approved_capability=rn.pf.capability_of_for(client))
        assert r["ok"] is True
        assert got.n == 1, f"★raw 를 {got.n}번 불렀다"
        assert r["budget"]["used"] == 1 and r["budget"]["denied"] == 0

    def test_the_second_raw_attempt_is_denied(self, client, monkeypatch,
                                              no_sleep, retries_on):
        """★★429 를 받아도 **두 번째를 안 산다**.

        ★이것이 「논리 1회」로는 못 막던 자리다 — 진짜 재시도 loop 가
        `reserve_current_call` 을 다시 부르고, 예산이 거기서 막는다.

        ★★`retries_on` 이 **꼭 있어야 한다.** 시험 conftest 는 재시도를 0 으로
        두는데, 유료 canary 는 **production 설정**(`llm_max_retries`)으로 돈다.
        0 으로 재면 「두 번째를 안 샀다」가 **예산 덕분인지 설정 덕분인지**
        못 가른다 — 그건 결함을 정답으로 못박는 시험이다.
        """
        got = _wire(monkeypatch, _rate_limited)
        r = rn.send_once(client, prompt="한 컷",
                         labeled_refs=[("배경판", b"a")],
                         approved_capability=rn.pf.capability_of_for(client))
        assert r["ok"] is False
        assert r["verdict"] == "inconclusive", "★실패를 합격으로 세지 않는다"
        assert got.n == 1, f"★raw 를 {got.n}번 불렀다 — 두 번째를 샀다"
        assert r["budget"]["used"] == 1 and r["budget"]["denied"] == 1

    def test_the_cap_is_one(self):
        assert rn.RAW_CALL_CAP == 1

    def test_the_budget_is_gone_afterwards(self, client, monkeypatch):
        """★예산은 이 판에만 깔린다 — 다른 일이 물려받지 않는다."""
        from app.core.image_call_budget import get_current_budget

        _wire(monkeypatch, _ok)
        rn.send_once(client, prompt="x", labeled_refs=[("a", b"a")],
                     approved_capability=rn.pf.capability_of_for(client))
        assert get_current_budget() is None


class TestItRefusesBeforeSpending:
    def test_an_asset_coordinate_is_refused(self):
        """★이번 판은 파일 좌표만 — asset loader 가 아직 없다."""
        with pytest.raises(rn.CanaryRefused):
            rn.assert_sources_allowed([{"source": "asset", "asset_id": "x"}])

    def test_a_file_coordinate_passes(self):
        rn.assert_sources_allowed([{"source": "file", "path": "/x"}])

    def test_a_changed_capability_stops_before_the_call(self, client,
                                                        monkeypatch):
        """★★보내기 **직전**에 다시 묻는다 — preflight 뒤에 바뀔 수 있다."""
        got = _wire(monkeypatch, _ok)
        with pytest.raises(rn.CanaryRefused):
            rn.send_once(client, prompt="x", labeled_refs=[("a", b"a")],
                         approved_capability={"max_images": 999,
                                              "min_images": 1})
        assert got.n == 0, "★거절했는데 이미 샀다"

    def test_too_many_references_stop_before_the_call(self, monkeypatch):
        """★provider 가 상한을 밝혔으면 그것과 대조한다."""
        from app.modules.llm.reve_image_client import ReveImageClient
        from app.modules.pipeline.grounding_reference_bundle import (
            ReferenceCountRefused, capability_of)

        got = _wire(monkeypatch, _ok)
        c = ReveImageClient.__new__(ReveImageClient)
        c._model = "reve-image-1.0"        # capability 가 읽는 유일한 칸
        cap = capability_of(c)
        assert cap["max_images"] == 1, f"★reve 상한이 {cap}"
        with pytest.raises(ReferenceCountRefused):
            rn.send_once(c, prompt="x",
                         labeled_refs=[("a", b"a"), ("b", b"b")],
                         approved_capability=cap)
        assert got.n == 0


class TestTheAcceptanceCountsWhatHappened:
    def _pre(self, client):
        p = _plan()
        return rn.pf.preflight(
            prompt=p["prompt"], labeled_refs=p["labeled_refs"],
            ref_roles=p["ref_roles"],
            ref_role_metadata=p["ref_role_metadata"],
            attached_meta=p["attached_meta"], shot=p["shot"],
            sidecar_required=p["sidecar_required"], client=client)

    def test_a_failed_send_is_not_a_pass(self, client):
        pre = self._pre(client)
        got = {"ok": False, "verdict": "inconclusive",
               "budget": {"cap": 1, "used": 1, "denied": 1}}
        assert rn.acceptance_of(pre, got, _plan())["passed"] is False

    def test_a_dry_run_is_not_a_pass(self, client):
        pre = self._pre(client)
        acc = rn.acceptance_of(pre, {"ok": None, "budget": None}, _plan())
        assert acc["passed"] is False, "★안 산 판을 합격으로 세지 않는다"

    def _seen(self, pre):
        return {"calls": 1,
                "image_shas": list(pre["lock"]["outbound"]["reference_shas"]),
                "text_parts": ["\n".join(pre["lock"]["outbound"]["labels"])]}

    def test_a_pass_needs_used_one_and_denied_zero(self, client):
        pre = self._pre(client)
        got = {"ok": True, "budget": {"cap": 1, "used": 1, "denied": 0},
               "provider_seen": self._seen(pre)}
        acc = rn.acceptance_of(pre, got, _plan())
        assert acc["passed"] is True
        assert acc["roles"] == _plan()["ref_roles"]
        assert acc["labels"] == [l for l, _b in _plan()["labeled_refs"]]

    def test_a_denied_extra_call_is_not_a_pass(self, client):
        """★성공했어도 **두 번째를 시도한 판**은 합격이 아니다."""
        pre = self._pre(client)
        got = {"ok": True, "budget": {"cap": 1, "used": 1, "denied": 1},
               "provider_seen": self._seen(pre)}
        assert rn.acceptance_of(pre, got, _plan())["passed"] is False


class TestTheRunnerDoesNotUseTheValidatingPath:
    def test_it_calls_generate_image_not_the_scene_wrapper(self):
        """★`generate_and_validate_scene` 은 moderation ≤4 + 보정 1 이 더 있다.

        ★글자가 아니라 **AST 로** 본다 — 설명을 적은 주석이 걸리면 안 된다
        ([[feedback-my-guard-caught-its-own-explanation]]).
        """
        import ast
        import inspect

        tree = ast.parse(inspect.getsource(rn))
        called = {n.func.attr for n in ast.walk(tree)
                  if isinstance(n, ast.Call)
                  and isinstance(n.func, ast.Attribute)}
        assert "generate_image" in called
        assert "generate_and_validate_scene" not in called


# ─────────────────────────────────────────────────────────────────────
# ★★★유료 **진입점**(`run`)에서 잰다 (Codex BLOCK A·B·C, 2026-08-31)
#
# 무료 unit 은 초록인데 `run()` wrapper 가 새 계약을 **우회**하고 있었다 —
# 또 조립만 잰 것이다. 여기서는 `run()` 을 공개 끝점으로 부른다.
# ─────────────────────────────────────────────────────────────────────


@pytest.fixture
def trace_on(monkeypatch):
    """부모 trace 가 열린 척한다. ★안 열리면 사지 않는 계약은 그대로다."""
    import contextlib

    from app.modules.llm import opik_trace

    class _H:
        uid = "trace-uid-테스트"

    @contextlib.contextmanager
    def fake(*a, **k):
        yield _H()

    monkeypatch.setattr(opik_trace, "open_trace", fake)


def _run(tmp_path, client, monkeypatch, *, live=True, **over):
    return rn.run(tmp_path, live=live, plan=_plan(**over), client=client,
                  **{k: v for k, v in over.items() if k == "approved_lock"})


class TestTheActualRunLocksWhatGoesOut:
    def test_a_changed_attached_meta_stops_the_run(self, client, monkeypatch,
                                                   trace_on, tmp_path):
        """★★BLOCK A — `run` 이 `attached_meta` 를 문에 안 넘기고 있었다.

        Codex 재현: `actual_plan_attached=[('background','must-be-locked')]`
        인데 `locked_attached=[]`. 칸을 갖춰도 **유료 진입점이 그 칸을 비운
        채 승인**되면 아무 소용이 없다.
        """
        got = _wire(monkeypatch, _ok)
        first = rn.run(tmp_path, live=False, plan=_plan(), client=client)
        approved = first["preflight"]["lock"]
        assert approved["outbound"]["attached"], "★잠금에 attached 가 비었다"

        changed = _plan()["attached_meta"][:1] + [("background", "다른값")]
        with pytest.raises(rn.pf.CanaryScopeMismatch):
            rn.run(tmp_path, live=True, plan=_plan(attached_meta=changed),
                   client=client, approved_lock=approved)
        assert got.n == 0, "★승인 밖인데 provider 를 불렀다"

    def test_the_same_plan_passes_the_lock(self, client, monkeypatch,
                                           trace_on, tmp_path):
        _wire(monkeypatch, _ok)
        first = rn.run(tmp_path, live=False, plan=_plan(), client=client)
        again = rn.run(tmp_path, live=False, plan=_plan(), client=client,
                       approved_lock=first["preflight"]["lock"])
        assert again["identity"] == first["identity"]


class TestTheCapabilityIsComparedToTheApprovedOne:
    def test_a_drifting_capability_stops_before_the_call(self, monkeypatch,
                                                         trace_on, tmp_path):
        """★★BLOCK B — 앞 판은 **지금 값끼리** 견줘 drift 를 놓쳤다.

        preflight 뒤에 바뀌는 client 를 넣는다. 승인된 값과 견주면 선다.
        """
        from app.modules.llm.gemini_image_client import GeminiImageClient

        got = _wire(monkeypatch, _ok)
        c = GeminiImageClient()
        monkeypatch.setattr(c, "_get_api_key", lambda *a, **k: "가짜")
        seq = {"n": 0}
        base = dict(c.reference_input_capability())

        def drifting():
            seq["n"] += 1
            if seq["n"] == 1:
                return dict(base)
            return {**base, "max_images": 7}   # ★preflight 뒤에 바뀌었다

        monkeypatch.setattr(c, "reference_input_capability", drifting)
        with pytest.raises(rn.CanaryRefused):
            rn.run(tmp_path, live=True, plan=_plan(), client=c)
        assert got.n == 0, "★바뀐 뒤인데 provider 를 불렀다"


class TestTheSamePlanIsNotBoughtTwice:
    def test_a_second_run_reuses_and_does_not_buy(self, client, monkeypatch,
                                                  trace_on, tmp_path):
        """★★BLOCK C — Codex 재현: 같은 plan·같은 out_dir 로 두 번 부르니
        `send_once_calls=2` 였다. 재실행·크래시에 중복 구매가 열려 있었다.
        """
        got = _wire(monkeypatch, _ok)
        a = rn.run(tmp_path, live=True, plan=_plan(), client=client)
        assert a["result"]["ok"] is True and got.n == 1
        b = rn.run(tmp_path, live=True, plan=_plan(), client=client)
        assert got.n == 1, f"★두 번째 판이 또 샀다 (raw {got.n}회)"
        assert b["result"].get("reused") is True
        assert b["identity"] == a["identity"]
        assert b["journal"]["bought"] == 1

    def test_an_unknown_outcome_stops_the_next_run(self, client, monkeypatch,
                                                   trace_on, tmp_path):
        """★「샀는지 모른다」가 남아 있으면 **자동으로 다시 안 산다**."""
        from app.modules.pipeline.grounding_chunk_journal import (
            NeedsHumanDecision)

        got = _wire(monkeypatch, _ok)
        first = rn.run(tmp_path, live=True, plan=_plan(), client=client)
        jp = tmp_path / "_canary_journal.json"
        j = json.loads(jp.read_text(encoding="utf-8"))
        for e in j["calls"]:
            if e["identity"] == first["identity"]:
                e["status"] = "uncertain"
        jp.write_text(json.dumps(j, ensure_ascii=False), encoding="utf-8")

        with pytest.raises(NeedsHumanDecision):
            rn.run(tmp_path, live=True, plan=_plan(), client=client)
        assert got.n == 1, "★모르는 판이 남았는데 또 샀다"

    def test_the_identity_is_the_whole_outbound(self, client, tmp_path):
        """★`prompt_sha + N` 은 신원이 아니다 — 이름표만 바뀌어도 다른 판."""
        a = rn.run(tmp_path, live=False, plan=_plan(), client=client)
        changed = [("다른 이름표", b"plate"), ("상세", b"det")]
        b = rn.run(tmp_path, live=False, plan=_plan(labeled_refs=changed),
                   client=client)
        assert a["identity"] != b["identity"]


class TestTheAcceptanceMeasuresWhatTheProviderSaw:
    def test_it_counts_the_images_that_actually_went(self, client,
                                                     monkeypatch, trace_on,
                                                     tmp_path):
        """★★`missing` 이 상수 0 이면 재는 척만 하는 것이다 (Codex)."""
        _wire(monkeypatch, _ok)
        r = rn.run(tmp_path, live=True, plan=_plan(), client=client)
        acc = r["acceptance"]
        assert acc["provider_raw_calls"] == 1
        assert acc["provider_images"] == 2, f"★{acc['provider_images']}장 갔다"
        assert acc["missing"] == [] and acc["extra"] == []
        assert acc["order_matches"] is True
        assert acc["labels_absent_from_text"] == []
        assert acc["passed"] is True

    def test_a_dropped_image_is_seen(self, client, monkeypatch, trace_on,
                                     tmp_path):
        """★provider 가 한 장을 못 받으면 **합격이 아니다**."""
        got = _wire(monkeypatch, _ok)
        real = client.generate_image

        def leaky(*, prompt, labeled_references, **kw):
            # ★조용히 마지막 한 장을 흘리는 client — 이런 결함을 잡아야 한다
            return real(prompt=prompt,
                        labeled_references=list(labeled_references)[:-1], **kw)

        monkeypatch.setattr(client, "generate_image", leaky)
        r = rn.run(tmp_path, live=True, plan=_plan(), client=client)
        assert got.n == 1
        acc = r["acceptance"]
        assert acc["provider_images"] == 1
        assert len(acc["missing"]) == 1
        assert acc["passed"] is False


# ─────────────────────────────────────────────────────────────────────
# ★★★실패한 판의 **기록**과 **다음 판** (Codex 2026-08-31 재리뷰)
#
# > 돈 문은 막았지만 감사 기록이 「아무것도 못 봄」이 됩니다.
# > 사람에게 장부를 고치라고 요구하지 마십시오.
# ─────────────────────────────────────────────────────────────────────


class TestAFailedCallStillLeavesItsRecord:
    def test_the_observation_survives_a_rate_limit(self, client, monkeypatch,
                                                   trace_on, no_sleep,
                                                   retries_on, tmp_path):
        """★★Codex 재현: `raw_counter=1` 인데 `provider_seen={}` 이었다.

        ★끝점은 URL counter 가 아니라 **얼어붙은 결과**의
        `provider_seen.calls` 다 — 그것이 감사에 남는 값이다.
        """
        got = _wire(monkeypatch, _rate_limited)
        r = rn.run(tmp_path, live=True, plan=_plan(), client=client)
        seen = r["result"]["provider_seen"]
        assert got.n == 1
        assert seen.get("calls") == 1, f"★관측을 잃었다: {seen}"
        assert len(seen.get("image_shas") or []) == 2, "★실린 사진을 잃었다"
        assert r["result"]["budget"]["used"] == 1
        assert r["result"]["budget"]["denied"] == 1
        assert r["result"]["journal_status"] == rn.JS_UNKNOWN

    def test_the_journal_keeps_the_identity_and_reason(self, client,
                                                       monkeypatch, trace_on,
                                                       no_sleep, retries_on,
                                                       tmp_path):
        """★바깥 helper 가 run_id·trace_id·까닭을 **덮지 않는다**."""
        _wire(monkeypatch, _rate_limited)
        r = rn.run(tmp_path, live=True, plan=_plan(), client=client)
        from app.modules.pipeline.grounding_chunk_journal import ChunkJournal
        # ★`calls` 는 append-only 사건 로그다 (2026-09-03) — 첫 줄(helper 의 ok)이
        #  아니라 **유효 뷰**(마지막 시도)를 읽는다. 로그의 옛 줄은 그대로 남는다.
        row = ChunkJournal(tmp_path / "_canary_journal.json").entries[r["identity"]]
        assert row["status"] == rn.JS_UNKNOWN
        assert row["run_id"] == r["run_id"]
        assert row["trace_id"] == "trace-uid-테스트"
        assert row["why"], "★까닭이 지워졌다"
        assert (row["provider_seen"] or {}).get("calls") == 1


class TestNobodyIsAskedToFixTheJournal:
    def test_an_unknown_outcome_ends_the_next_run_without_buying(
            self, client, monkeypatch, trace_on, no_sleep, retries_on,
            tmp_path):
        """★★앞 판이 「나갔는데 모른다」면 다음 판은 **바로 끝난다**.

        ★60초 기다렸다 사람을 부르지 않는다 — 자동화 정책과 반대다.
        """
        got = _wire(monkeypatch, _rate_limited)
        first = rn.run(tmp_path, live=True, plan=_plan(), client=client)
        assert first["result"]["journal_status"] == rn.JS_UNKNOWN

        second = rn.run(tmp_path, live=True, plan=_plan(), client=client)
        assert got.n == 1, "★모르는 판이 남았는데 또 샀다"
        assert second["result"]["not_retried"] is True
        assert second["acceptance"]["passed"] is False

    def test_a_trace_failure_leaves_no_reservation(self, client, monkeypatch,
                                                   tmp_path):
        """★★trace 가 안 열리면 **자리도 안 잡는다** — 다음 판이 그냥 돈다."""
        import contextlib

        from app.modules.llm import opik_trace

        got = _wire(monkeypatch, _ok)

        @contextlib.contextmanager
        def dead(*a, **k):
            yield None

        monkeypatch.setattr(opik_trace, "open_trace", dead)
        with pytest.raises(rn.CanaryRefused):
            rn.run(tmp_path, live=True, plan=_plan(), client=client)
        assert got.n == 0
        from app.modules.pipeline.grounding_chunk_journal import ChunkJournal
        jp = tmp_path / "_canary_journal.json"
        # ★유효 뷰가 비어야 한다 — 로그(`calls`)에는 놓은 줄이 history 로 남는다
        left = ChunkJournal(jp).entries if jp.exists() else {}
        assert left == {}, f"★안 나갔는데 자리가 남았다: {left}"

    def test_after_the_trace_comes_back_the_same_plan_just_runs(
            self, client, monkeypatch, trace_on, tmp_path):
        """★고친 뒤 **그대로 다시 돌리면 된다** — 사람이 장부를 안 고친다."""
        got = _wire(monkeypatch, _ok)
        r = rn.run(tmp_path, live=True, plan=_plan(), client=client)
        assert got.n == 1 and r["acceptance"]["passed"] is True

    def test_a_capability_drift_releases_its_slot(self, monkeypatch, trace_on,
                                                  tmp_path):
        """★자리를 잡은 뒤 승인 밖이 되어 **안 나간** 판도 자리를 놓는다."""
        from app.modules.llm.gemini_image_client import GeminiImageClient

        got = _wire(monkeypatch, _ok)
        c = GeminiImageClient()
        monkeypatch.setattr(c, "_get_api_key", lambda *a, **k: "가짜")
        base = dict(c.reference_input_capability())
        seq = {"n": 0}

        def drifting():
            seq["n"] += 1
            # preflight 가 첫 번째, 보내기 직전이 두 번째다
            return dict(base) if seq["n"] == 1 else {**base, "max_images": 7}

        monkeypatch.setattr(c, "reference_input_capability", drifting)
        with pytest.raises(rn.CanaryRefused):
            rn.run(tmp_path, live=True, plan=_plan(), client=c)
        assert got.n == 0
        from app.modules.pipeline.grounding_chunk_journal import ChunkJournal
        left = ChunkJournal(tmp_path / "_canary_journal.json").entries
        assert left == {}, f"★안 나갔는데 자리가 남았다: {left}"