"""②⑥ 을 **Gemini 정순 · Grok 역순 2콜**로 (2026-08-29 사용자 지시).

> "2, 6번은 Gemini 정순, Grok 역순 으로 병렬로 (즉 두번만)"

재는 것 — 말이 아니라 **동작**으로:
  ① 콜이 **정확히 두 번**인가 (넷이 되면 지시 위반)
  ② 두 콜이 **겹치는가**
  ③ Grok 에게 **정말 뒤집어** 보여 주는가
  ④ 역순 응답이 **canonical 로 되돌아오는가**
  ⑤ 갈리면 **GG46 정규화·위반 합집합**이 도는가 (모델 척도가 달라도)
  ⑥ 한쪽 실패 → 생존 슬롯, 양쪽 실패 → raise
  ⑦ ★옛 `agreement` 이름을 **안 쓰는가** (뜻이 달라졌다)
"""
from __future__ import annotations

import threading
from typing import Any, Dict, List

import pytest

from app.modules.pipeline.multiroll_gemini import (
    CROSS_MODEL_ORDER_POLICY_VERSION,
    _judge_cross_model_order,
)

G = "gemini-pro"
X = "openrouter:x-ai/grok-4.6"
LABELS = ["A", "B"]


def _verdict(winner: str, scores: Dict[str, int],
             violations: Dict[str, List[str]] | None = None) -> Dict[str, Any]:
    order = sorted(scores, key=lambda k: -scores[k])
    return {
        "winner": winner,
        "ranking": order,
        "verdicts": [{"label": k, "score": scores[k], "verdict_ko": ""}
                     for k in LABELS],
        "readings": [{"label": k,
                      "hard_violations": (violations or {}).get(k, [])}
                     for k in LABELS],
        "all_candidates_fail": False,
    }


class _Spy:
    """`one(model, parts, suffix)` 를 흉내내며 **무엇을 몇 번** 받았는지 센다."""

    def __init__(self, answers: Dict[str, Dict[str, Any]],
                 gate: threading.Barrier | None = None):
        self.answers = answers
        self.calls: List[str] = []
        self.seen_order: Dict[str, List[str]] = {}
        self.gate = gate
        self.lock = threading.Lock()

    def __call__(self, model, parts, suffix):
        if self.gate is not None:
            try:
                self.gate.wait()
            except threading.BrokenBarrierError:
                pass
        # parts 에서 "Candidate X:" 뒤에 어느 후보 파일이 왔는지 읽는다
        shown: List[str] = []
        for i, p in enumerate(parts):
            t = p.get("text") or ""
            if t.startswith("Candidate "):
                nxt = parts[i + 1] if i + 1 < len(parts) else {}
                shown.append(str(nxt.get("_src") or "?"))
        with self.lock:
            self.calls.append(model)
            self.seen_order[model] = shown
        ans = self.answers.get(model)
        if isinstance(ans, Exception):
            raise ans
        return ans


@pytest.fixture(autouse=True)
def _fake_png(monkeypatch):
    """이미지 대신 출처를 실은 표시를 넣는다 — 순서를 눈으로 볼 수 있게."""
    monkeypatch.setattr(
        "app.modules.pipeline.multiroll_gemini.png_part",
        lambda p: {"type": "image", "_src": str(p)})
    monkeypatch.setattr(
        "app.modules.pipeline.multiroll_gemini.ref_parts", lambda refs: [])


def _run(spy) -> Dict[str, Any]:
    return _judge_cross_model_order(
        [G, X], spy, "HEAD", [], ["candA.png", "candB.png"], list(LABELS))


# ── ①②③ 콜 수 · 겹침 · 뒤집힘 ──────────────────────────────────

def test_exactly_two_calls_one_per_model():
    spy = _Spy({G: _verdict("A", {"A": 8, "B": 5}),
                X: _verdict("A", {"A": 9, "B": 4})})
    _run(spy)
    assert len(spy.calls) == 2, f"콜이 {len(spy.calls)}번 — 지시는 「두번만」"
    assert sorted(spy.calls) == sorted([G, X])


def test_the_two_calls_overlap():
    """★진짜로 동시에 살아 있어야 문이 열린다."""
    gate = threading.Barrier(2, timeout=3.0)
    spy = _Spy({G: _verdict("A", {"A": 8, "B": 5}),
                X: _verdict("A", {"A": 9, "B": 4})}, gate=gate)
    _run(spy)  # 순차면 여기서 타임아웃 → BrokenBarrier
    assert not gate.broken, "두 콜이 안 겹쳤다 — 병렬이 아니다"


def test_grok_really_sees_the_reversed_order():
    """★Gemini 는 정순, Grok 은 **뒤집어** 본다."""
    spy = _Spy({G: _verdict("A", {"A": 8, "B": 5}),
                X: _verdict("A", {"A": 9, "B": 4})})
    _run(spy)
    assert spy.seen_order[G] == ["candA.png", "candB.png"], "Gemini 가 정순이 아니다"
    assert spy.seen_order[X] == ["candB.png", "candA.png"], "Grok 이 역순이 아니다"


# ── ④ 역매핑 ────────────────────────────────────────────────────

def test_reverse_answer_is_mapped_back_to_canonical():
    """Grok 이 display 'A'(=canonical B)를 골랐으면 승자는 **B** 다."""
    spy = _Spy({
        G: _verdict("B", {"A": 3, "B": 9}),        # 정순 — B 를 고름
        X: _verdict("A", {"A": 9, "B": 3}),        # 역순 display A = canonical B
    })
    out = _run(spy)
    assert out["winner"] == "B", (
        "역매핑이 안 됐다 — 두 슬롯이 같은 그림을 골랐는데 갈린 것으로 읽힌다")
    cmo = out["cross_model_order"]
    assert cmo["route"] == "cross_slot_agree"
    assert cmo["slot_winner"] == {G: "B", X: "B"}


# ── ⑤ 갈리면 GG46 규칙 ─────────────────────────────────────────

def test_disagreement_uses_normalized_scores_and_violation_union():
    """★척도가 달라도 큰 척도가 지배하면 안 되고, 위반은 합집합이다."""
    spy = _Spy({
        # Gemini 0~10 척도 — A 를 조금 앞세움
        G: _verdict("A", {"A": 7, "B": 6}),
        # Grok 0~100 척도, 역순 display: display A = canonical B
        # → canonical 로는 B=90, A=20 이므로 B 를 크게 앞세움
        X: _verdict("A", {"A": 90, "B": 20},
                    violations={"A": ["B 쪽 하드위반"]}),
    })
    out = _run(spy)
    cmo = out["cross_model_order"]
    assert cmo["route"] == "cross_slot_combined"
    assert cmo["slot_winner_match"] is False
    # 정규화 합: A = 7/7 + 20/90 ≈ 1.22 · B = 6/7 + 90/90 ≈ 1.86
    # 그런데 canonical B 에 하드위반 1건 → 페널티가 붙는다.
    # 어느 쪽이 이기든 **원점수 합(97 vs 26)으로 정해지지 않았음**이 계약이다.
    assert out["ranking"][0] in ("A", "B")
    scores = {v["label"]: v["score"] for v in out["verdicts"]}
    assert scores["A"] != 97 and scores["B"] != 26, (
        "원점수를 그대로 실었다 — 큰 척도 모델이 판정을 지배한다")


# ── ⑥ 실패 갈래 ─────────────────────────────────────────────────

def test_one_slot_failure_survives_with_named_route():
    """★route 에 **모델 이름을 박지 않는다** (2026-08-29).

    종전 이름은 `single_grok_reverse` 였다. 그런데 둘째 슬롯이 grok 에서
    GPT-5.6 Sol 로 바뀌었고 또 바뀔 수 있다 — 「grok 이 골랐다」고 적힌
    기록이 실제로는 GPT 였으면 그건 결함보다 나쁜 거짓말이다. 어느
    모델이었는지는 `slots[].model` 이 SOT 다.
    """
    spy = _Spy({G: _verdict("A", {"A": 8, "B": 5}), X: RuntimeError("boom")})
    out = _run(spy)
    cmo = out["cross_model_order"]
    assert cmo["route"] == "single_forward"
    assert "grok" not in cmo["route"] and "gemini" not in cmo["route"], (
        "route 에 모델 이름이 박혔다 — 슬롯이 바뀌면 기록이 거짓이 된다")
    assert cmo["failed"] == [X]
    assert out["winner"] == "A"
    # 어느 모델이 어느 순서를 봤는지는 슬롯 기록에 그대로 남는다
    assert [s["model"] for s in cmo["slots"]] == [G, X]
    assert [s["order"] for s in cmo["slots"]] == ["forward", "reverse"]


def test_both_slots_failing_raises():
    spy = _Spy({G: RuntimeError("a"), X: RuntimeError("b")})
    with pytest.raises(RuntimeError):
        _run(spy)


# ── ⑦ 옛 이름을 안 쓴다 ────────────────────────────────────────

def test_the_old_agreement_name_is_not_reused():
    """★`agreement` 는 「같은 모델이 순서에 안 흔들렸다」였다.

    지금은 **모델도 순서도 다른 두 슬롯**의 승자 일치일 뿐이다. 이름을
    물려 쓰면 기록이 조용히 거짓말을 한다.
    """
    spy = _Spy({G: _verdict("A", {"A": 8, "B": 5}),
                X: _verdict("B", {"A": 5, "B": 8})})
    cmo = _run(spy)["cross_model_order"]
    assert "agreement" not in cmo, "옛 이름을 물려 썼다"
    assert "slot_winner_match" in cmo
    assert cmo["policy"] == CROSS_MODEL_ORDER_POLICY_VERSION


def test_slot_record_names_model_and_order():
    """기록이 **어느 모델이 어느 순서를 맡았는지**를 말한다."""
    spy = _Spy({G: _verdict("A", {"A": 8, "B": 5}),
                X: _verdict("A", {"A": 9, "B": 4})})
    slots = _run(spy)["cross_model_order"]["slots"]
    assert [s["model"] for s in slots] == [G, X], "슬롯 순서가 흔들린다"
    assert [s["order"] for s in slots] == ["forward", "reverse"]
    assert all("display_to_canonical" in s for s in slots)


# ── ⑧ owns_order 가 **dispatch 와 같은 조건**인가 ──────────────

def test_owns_order_is_false_when_dispatch_would_not_take_the_cross_path(
        monkeypatch):
    """★두 selector 가 갈리면 **정·역 비교가 조용히 사라진다.**

    dispatch 는 Qwen 갈래를 **먼저** 잡아간다. 그래서 models 에 Qwen 과
    OpenRouter 가 같이 실리면 실제로는 `_judge_gq`(정순 한 번)로 가는데,
    `owns_order` 를 `_gg46_equal` 로 매기면 True 가 되어 바깥이 flip 을
    건너뛴다. 오류는 안 난다 — 없어지기만 한다.

    오늘 `resolve_select_judge_models` 는 갈래마다 목록 하나만 돌려주어
    이 조합이 안 나온다. **그 우연이 계약을 대신하게 두지 않는다.**
    """
    import app.modules.pipeline.multiroll_gemini as mg

    def _fn(models):
        monkeypatch.setattr(mg, "resolve_select_judge_models", lambda: models)
        return mg.make_gemini_judge_fn(judge_sys="sys", judge_schema={})

    cross = _fn([mg.JUDGE_MODEL, X])
    assert cross.owns_order is True, "cross 갈래인데 순서를 안 가졌다고 한다"

    # Qwen 이 섞이면 dispatch 는 `_judge_gq` 로 간다 → 소유하지 않는다
    mixed = _fn([mg.JUDGE_MODEL, mg.QWEN_JUDGE_MODEL, X])
    assert mixed.owns_order is False, (
        "dispatch 는 Qwen 갈래로 가는데 순서를 소유한다고 말한다 — "
        "바깥이 flip 을 건너뛰어 정·역 비교가 통째로 사라진다")

    # Gemini 단독·G+Q 도 소유하지 않는다
    assert _fn([mg.JUDGE_MODEL, mg.QWEN_JUDGE_MODEL]).owns_order is False
    assert _fn([mg.JUDGE_MODEL]).owns_order is False


# ── ⑨ ⑥ 갈래가 「둘 다 못 쓴다」를 **끝점까지** 나르는가 ────────

def _run_fix_rejudge_owns_order(tmp_path, *, rejudge_all_fail: bool):
    """★조립부가 아니라 `run_multiroll_select` **끝점**에서 잰다.

    Codex BLOCK (2026-08-29): 이 갈래가 `all_candidates_fail` 을 버려서
    `_set_needs_reshoot` 가 초기 롤 선언으로 되돌아갔다. 초기 롤은
    통과했는데 원본·수정본 재판정에서 둘 다 못 쓰는 샷이 **재촬영 표시
    없이 정상 산출처럼** 남았다.

    그래서 초기 롤은 **통과**시키고(= `initial_roll_all_fail` 없음),
    재판정에서만 실패를 선언한다. 되돌아가면 그 실패가 사라진다.
    """
    from app.modules.pipeline.multiroll_select import run_multiroll_select

    def gen_fn(tag, prompt, labeled_refs, out_path):
        out_path.parent.mkdir(parents=True, exist_ok=True)
        body = b"FIX" if tag.endswith("_fix") else b"ROLL:" + tag.encode()
        out_path.write_bytes(b"\x89PNG\r\n\x1a\n" + body)
        return out_path

    def judge_fn(tag, prompt, labeled_refs, cand_paths, labels):
        # 초기 선정 — **통과**한다 (all_candidates_fail 없음)
        return _verdict("A", {lab: (9 if lab == "A" else 3) for lab in labels})

    def critique_fn(tag, prompt, labeled_refs, image_path):
        return {"issues": [{"issue_ko": "결함", "fix_en": "repair"}]}

    def rejudge_fn(tag, prompt, labeled_refs, cand_paths, labels):
        out = _verdict("B", {"A": 4, "B": 7})
        out["all_candidates_fail"] = rejudge_all_fail
        out["cross_model_order"] = {"route": "cross_slot_agree"}
        return out

    rejudge_fn.owns_order = True  # ★C 의 단일 호출 갈래로 보낸다

    return run_multiroll_select(
        tag="t1", prompt="P", labeled_refs=[],
        out_stem=tmp_path / "out" / "s1",
        gen_fn=gen_fn, judge_fn=judge_fn,
        critique_fn=critique_fn, fix_gen_fn=gen_fn,
        roll_count=2, critique_enabled=True,
        fix_head="H", fix_tail="T", fix_label="L",
        fix_rejudge_fn=rejudge_fn, record=None,
    )


def test_owns_order_rejudge_carries_all_fail_to_needs_reshoot(tmp_path):
    """둘 다 못 쓴다고 재판정하면 **재촬영 표시가 끝까지 간다.**"""
    _, record = _run_fix_rejudge_owns_order(tmp_path, rejudge_all_fail=True)
    assert record.get("initial_roll_all_fail") in (None, False), (
        "초기 롤이 실패로 잡혔다 — 그러면 되돌아간 경로도 통과해 "
        "이 시험이 결함을 못 잡는다")
    assert record["fix_rejudge"]["all_candidates_fail"] is True, (
        "재판정 선언이 기록에서 사라졌다")
    assert record.get("needs_reshoot") is True, (
        "실패한 최종 _sel 이 재촬영 표시 없이 정상 산출처럼 남는다")


def test_owns_order_rejudge_without_all_fail_does_not_flag(tmp_path):
    """양성 뒤집기 — 재판정이 통과면 재촬영 표시가 붙지 않는다."""
    _, record = _run_fix_rejudge_owns_order(tmp_path, rejudge_all_fail=False)
    assert record["fix_rejudge"]["all_candidates_fail"] is False
    assert "needs_reshoot" not in record