"""cine 검증을 **같은 모델 두 번 → 두 모델 각 한 번** (2026-08-27, #92 4단계).

사용자 지시: 「gpt sol vlm 은 성능이 안좋아 … 무조건 gemini 3.1 pro 와 grok
최신 모델 둘을 사용해야해」.

## 왜 콜 수를 안 늘리나

종전에는 `gpt` **한 모델에게 같은 그림을 두 번** 물었다. 회차가 둘인 이유는
「둘 다 파탄이라 해야 기각」이라는 합의 강도였는데, **같은 모델을 두 번
부르면 그 둘이 독립이 아니다** — 한 모델의 버릇이 두 표로 센다.

모델을 갈면 **같은 콜 수로 더 독립적인 두 의견**을 얻는다.
2모델×2회=4콜은 근거 없는 중복이다(Codex 판정).

## 합산 규칙은 **안 바꿨다**

    둘 다 파탄     → broken (기각)
    엇갈림         → split  (기각 안 함)
    한쪽이라도 실패 → complete=False → **기각 권한 없음**

이미 그 모양이었다. 바꾼 것은 「누가 말하나」뿐이다.
"""
from __future__ import annotations

import ast
import pathlib

import pytest

import app.modules.pipeline.cine_verify as cv


def test_the_pair_is_what_the_user_asked_for():
    assert cv.CINE_VERIFY_MODELS == ("gemini-pro", "grok")
    assert "gpt" not in cv.CINE_VERIFY_MODELS, "물린 모델이 남아 있다"


def test_two_rounds_means_two_models_not_two_calls_to_one():
    assert cv.verify_models(2) == ("gemini-pro", "grok")
    assert len(set(cv.verify_models(2))) == 2, "같은 모델을 두 번 부른다"


def test_rounds_cannot_drop_a_model(monkeypatch):
    """★★**내가 여기서 계약 위반을 정답으로 못박았다** (2026-08-27 Codex
    BLOCK).

    앞 판은 `verify_models(1) == ("gemini-pro",)` 를 기대값으로 썼다.
    그러면 공개 설정 `STILL_CINE_VERIFY_ROUNDS=1` 하나로

        · grok 이 **조용히** 빠지고 (「무조건 둘」 위반)
        · Gemini 한 번의 broken 만으로 최종본이 기각된다
          (「둘 다 동의할 때만 기각」 위반)

    검사를 끄는 것은 `still_cine_verify_enabled` 플래그가 한다 — 회차가
    아니다.
    """
    for r in (0, 1, 2, 3, None, "이상한값"):
        assert cv.verify_models(r) == cv.CINE_VERIFY_MODELS, (
            f"회차 {r!r} 로 짝이 줄었다")


def test_a_stale_rounds_setting_does_not_rejudge(monkeypatch):
    """★회차는 이 계약에서 아무 뜻이 없다 — 접으면 쓰지도 않는 설정을
    바꿀 때 **전량 재판정**이 난다."""
    assert cv.verify_contract_sha(1) == cv.verify_contract_sha(2)


def test_one_model_alone_cannot_reject(monkeypatch):
    """★★설정을 어떻게 만져도 **한 모델의 파탄만으로는 기각 못 한다.**

    위 시험들은 함수를 보고, 이 시험은 **나가는 판정**을 본다.
    """
    from app.core.config import settings

    cats = _cats()
    a, b = cv.CINE_VERIFY_MODELS
    _wire(monkeypatch, {a: {c: (c != cats[0]) for c in cats},
                        b: {c: True for c in cats}})
    monkeypatch.setattr(settings, "still_cine_verify_rounds", 1,
                        raising=False)
    out = cv.verify_cine_result(result_png=b"x", rounds=1)
    assert out["models"] == list(cv.CINE_VERIFY_MODELS), (
        "회차 1 로 grok 이 빠졌다")
    assert out["broken"] == [], "★한 모델만 파탄이라 했는데 기각한다"
    assert cats[0] in out["split"]


def test_the_contract_sha_folds_the_whole_pair(monkeypatch):
    """★한 모델만 접으면 짝의 나머지를 갈아도 지문이 안 움직인다 —
    옛 판정이 「지금 계약」으로 통한다."""
    before = cv.verify_contract_sha(2)
    monkeypatch.setattr(cv, "CINE_VERIFY_MODELS", ("gemini-pro", "딴것"))
    assert cv.verify_contract_sha(2) != before, (
        "짝의 둘째를 갈았는데 계약 지문이 그대로다")


def test_the_physical_names_are_folded_too(monkeypatch):
    """물리 모델 판이 바뀌면 다시 판정해야 한다."""
    from app.core.config import settings

    before = cv.verify_contract_sha(2)
    monkeypatch.setattr(settings, "grok_judge_model", "x-ai/grok-9.9")
    assert cv.verify_contract_sha(2) != before, (
        "grok 물리 판이 바뀌었는데 지문이 그대로다")


def test_each_model_gets_its_own_physical_name():
    got = cv.resolve_verify_model_physical()
    assert "," in got, "짝 전체가 안 들어갔다"
    assert cv.resolve_verify_model_physical("grok") in got
    assert cv.resolve_verify_model_physical("gemini-pro") in got


# ── 나가는 호출 ──────────────────────────────────────────────────────

def test_the_contract_lives_in_one_place_only():
    """★★**같은 계약이 두 벌이면 한 벌에 봉인을 빠뜨린다** (2026-08-27).

    종전에는 `cine_verify` 가 `call_structured` 를 직접 돌아, 「두 모델 각
    한 번 · fallback 봉인 · Router 재시도 봉인 · 모델별 usage」라는 같은
    계약이 `dual_vlm` 과 **두 벌** 있었다. 둘째 벌에 `num_retries` 가
    빠져 Codex 가 **두 번** BLOCK 했다.

    한 벌로 합쳤다 — 봉인을 넣는 것을 잊을 자리가 없어진다.
    """
    src = (pathlib.Path(__file__).resolve().parents[2] / "app" / "modules"
           / "pipeline" / "cine_verify.py").read_text(encoding="utf-8")
    direct = [n for n in ast.walk(ast.parse(src))
              if isinstance(n, ast.Call)
              and getattr(n.func, "id", "") == "call_structured"]
    assert not direct, (
        f"판정이 다시 직접 호출로 갈라졌다 (line "
        f"{[n.lineno for n in direct]}) — 봉인이 한 벌에만 남는다")
    assert "ask_both" in src, "단일 계약(dual_vlm)을 안 쓴다"


def _sound_answer() -> str:
    """schema 가 요구하는 **온전한** 답 — `minItems` 가 범주 수와 같다."""
    import json

    return json.dumps({"checks": [
        {"category": c, "sound": True, "what_is_impossible_ko": ""}
        for c, _ in cv.DEFECTS]})


def test_router_retries_are_sealed_at_the_endpoint(monkeypatch):
    """★★**「총 2콜」이 되려면 여기도 봉인해야 한다** (2026-08-27 Codex
    재리뷰).

    이 자리는 `dual_vlm.ask_both` 를 안 쓰고 `call_structured` 를 직접
    부른다 — `enable_fallback=False` 만으로는 Router 의 `num_retries=3` 이
    그대로라 **한 모델이 최대 4번 전송**될 수 있다.

    ★인자가 아니라 **`_completion` 이 받은 kwargs** 를 본다. 앞 판의 PR
     설명이 「총 2콜」을 주장했는데 코드가 안 받쳐 줬다.
    """
    import app.modules.llm.llm_client as lc

    seen = []

    def _fake_completion(binding, model, kwargs):
        seen.append((model, kwargs.get("num_retries")))
        return type("R", (), {
            "choices": [type("C", (), {"message": type("M", (), {
                "content": _sound_answer()})()})()],
            "usage": None, "model": f"phys/{model}"})()

    monkeypatch.setattr(lc, "_completion", _fake_completion)
    monkeypatch.setattr(lc, "_get_router_binding", lambda: object())
    monkeypatch.setattr(cv, "load_prompt", lambda *a, **k: "sys")
    from app.core.config import settings

    monkeypatch.setattr(settings, "openrouter_api_key", "sk", raising=False)
    monkeypatch.setattr(settings, "grok_judge_model", "x", raising=False)

    cv._ask_all(b"x", step_tag="t", models=cv.CINE_VERIFY_MODELS,
                project_config=None, opik_metadata=None)
    assert len(seen) == len(cv.CINE_VERIFY_MODELS)
    for model, n in seen:
        assert n == 0, (
            f"{model}: Router 재시도가 열려 있다(num_retries={n!r}) — "
            "「각 한 번」이 실제로는 최대 네 번이다")


def test_usage_reaches_the_result(monkeypatch):
    """모델별 토큰·비용·물리 모델이 산출까지 온다."""
    cats = _cats()
    _wire(monkeypatch, {m: {c: True for c in cats}
                        for m in cv.CINE_VERIFY_MODELS})
    out = cv.verify_cine_result(result_png=b"x", rounds=2)
    for m in cv.CINE_VERIFY_MODELS:
        assert out["model_usage"][m]["physical_model"] == f"phys/{m}"


def test_each_model_gets_its_own_step_tag(monkeypatch):
    """★기록에서 두 모델이 안 갈리면 「어느 쪽이 파탄이라 했나」를 못 읽는다."""
    seen = []

    def _fake(step, sys_p, parts, schema, **kw):
        seen.append((step, (kw.get("project_config") or {})
                     .get(step, {}).get("model")))
        return {"checks": []}

    import app.modules.llm.llm_client as lc

    monkeypatch.setattr(lc, "call_structured", _fake)
    monkeypatch.setattr(cv, "load_prompt", lambda *a, **k: "sys")
    cv._ask_all(b"x", step_tag="t", models=cv.CINE_VERIFY_MODELS,
                project_config=None, opik_metadata=None)
    assert seen == [(f"t__{m}", m) for m in cv.CINE_VERIFY_MODELS], seen


# ── 합산은 안 바꿨다 ─────────────────────────────────────────────────

def _wire(monkeypatch, answers):
    """모델별 답을 정해 준다. `answers[alias] = {category: sound}`.

    ★대역을 **`dual_vlm` 경계**에 둔다 — `cine_verify` 가 그 계약을 실제로
     쓰는지까지 재려면 그 아래를 흉내 내야 한다.
    """
    from app.modules.llm.dual_vlm import DualResult, OneCall

    def _fake_all(judged_png, *, step_tag, models, project_config,
                  opik_metadata):
        calls = []
        for m in models:
            a = answers[m]
            if isinstance(a, Exception):
                calls.append(OneCall(alias=m, ok=False,
                                     error=f"{type(a).__name__}: {a}",
                                     usage={"physical_model": f"phys/{m}"}))
                continue
            calls.append(OneCall(
                alias=m, ok=True,
                payload={"checks": [
                    {"category": cat, "sound": ok,
                     "what_is_impossible_ko": "" if ok else "이상"}
                    for cat, ok in a.items()]},
                usage={"physical_model": f"phys/{m}"}))
        return DualResult(calls=calls)

    monkeypatch.setattr(cv, "_ask_all", _fake_all)


def _cats():
    return [c for c, _ in cv.DEFECTS]


def test_both_broken_rejects(monkeypatch):
    cats = _cats()
    _wire(monkeypatch, {m: {c: (c != cats[0]) for c in cats}
                        for m in cv.CINE_VERIFY_MODELS})
    out = cv.verify_cine_result(result_png=b"x", rounds=2)
    assert out["complete"] is True
    assert cats[0] in out["broken"], "둘 다 파탄이라 했는데 기각 안 한다"
    assert out["ok"] is False


def test_disagreement_is_split_not_reject(monkeypatch):
    cats = _cats()
    a, b = cv.CINE_VERIFY_MODELS
    _wire(monkeypatch, {a: {c: (c != cats[0]) for c in cats},
                        b: {c: True for c in cats}})
    out = cv.verify_cine_result(result_png=b"x", rounds=2)
    assert out["broken"] == [], "엇갈렸는데 기각한다"
    assert cats[0] in out["split"]
    assert out["ok"] is True


def test_one_model_dead_means_no_reject_authority(monkeypatch):
    cats = _cats()
    a, b = cv.CINE_VERIFY_MODELS
    _wire(monkeypatch, {a: {c: (c != cats[0]) for c in cats},
                        b: RuntimeError("dead")})
    out = cv.verify_cine_result(result_png=b"x", rounds=2)
    assert out["complete"] is False, "한쪽이 죽었는데 완주로 본다"
    assert out["broken"] == [], "★기각 권한이 없는데 기각했다"
    assert out["ok"] is True


def test_the_dead_model_does_not_stop_the_other(monkeypatch):
    """★한쪽이 죽었다고 나머지를 안 물으면 「못 물어봤다」와 「둘 다
    나쁘다」가 구분이 안 된다."""
    cats = _cats()
    a, b = cv.CINE_VERIFY_MODELS
    _wire(monkeypatch, {a: RuntimeError("dead"),
                        b: {c: True for c in cats}})
    out = cv.verify_cine_result(result_png=b"x", rounds=2)
    assert out["rounds"] == 1, "죽은 쪽 뒤로 안 물었다"
    assert out["complete"] is False


# ── 기록 ─────────────────────────────────────────────────────────────

def test_the_result_says_who_judged(monkeypatch):
    cats = _cats()
    _wire(monkeypatch, {m: {c: True for c in cats}
                        for m in cv.CINE_VERIFY_MODELS})
    out = cv.verify_cine_result(result_png=b"x", rounds=2)
    assert out["models"] == list(cv.CINE_VERIFY_MODELS), (
        "누가 판정했는지 산출에 없다 — 설정만 보고는 모른다")
    assert set(out["model_usage"]) == set(cv.CINE_VERIFY_MODELS)
    for m in cv.CINE_VERIFY_MODELS:
        assert out["model_usage"][m]["physical_model"] == f"phys/{m}"


def test_a_dead_model_still_appears_in_the_record(monkeypatch):
    cats = _cats()
    a, b = cv.CINE_VERIFY_MODELS
    _wire(monkeypatch, {a: {c: True for c in cats}, b: RuntimeError("dead")})
    out = cv.verify_cine_result(result_png=b"x", rounds=2)
    assert b in out["model_usage"], "죽은 모델이 기록에서 사라졌다"
    assert b in str(out.get("error") or ""), "어느 모델이 죽었는지 안 적었다"


if __name__ == "__main__":
    raise SystemExit(pytest.main([__file__]))