"""G+Q 동시 판정 체계 (2026-08-10 설계) — 합의 규칙·분기·OFF 불변.

설계 SOT: docs/superpowers/specs/2026-08-10-dual-judge-selection-redesign-design.md
계약 요점:
  · ON: 선정 = [gemini-pro, qwen-vlm] 동시 → winner 일치/격차<0.2 = Gemini
    채택, 이상 = 합산 합의. 수정 = Qwen 관찰 → Gemini 취합(계약 불변).
  · OFF: 기존 경로 그대로 — 반환값·지문 재료가 1비트도 안 움직여야 한다
    (완료 256샷 동결이 이 기본값에 걸려 있다).
"""
from __future__ import annotations

import json
from typing import Any, Dict, List

import pytest

import app.modules.pipeline.multiroll_gemini as mg
from app.core.config import settings
from app.core.errors import AppError
from app.modules.pipeline.multiroll_select import build_gq_observe_schema


def _verdict(labels_scores: Dict[str, int], winner: str,
             acf: bool = False) -> Dict[str, Any]:
    """프로덕션 판정 스키마 shape 의 데이터 fixture (로직 복제 아님)."""
    labs = list(labels_scores)
    return {
        "winner": winner,
        "ranking": sorted(labs, key=lambda x: -labels_scores[x]),
        "verdicts": [
            {"label": lab, "score": sc, "verdict_ko": f"{lab} 판정"}
            for lab, sc in labels_scores.items()
        ],
        "readings": [
            {"label": lab, "direction": "-", "built_space": "-",
             "entities": "-", "physics": "-", "hard_violations": []}
            for lab in labs
        ],
        "all_candidates_fail": acf,
    }


# ── _q_gap ────────────────────────────────────────────────────────────

def test_q_gap_normalized_by_top():
    q = _verdict({"A": 10, "B": 5}, "A")
    assert mg._q_gap(q, g_winner="B", q_winner="A") == pytest.approx(0.5)


def test_q_gap_all_zero_scores_means_no_opposition():
    # 전 후보 0점 = 반대 근거 없음 → 0.0 → Gemini 채택으로 떨어진다.
    q = _verdict({"A": 0, "B": 0}, "A")
    assert mg._q_gap(q, "B", "A") == 0.0


# ── _judge_gq 갈래 ────────────────────────────────────────────────────

def _one_factory(results: Dict[str, Any], fails: set = frozenset()):
    calls: List[str] = []

    def one(model, parts, tag_suffix):
        calls.append(model)
        if model in fails:
            raise RuntimeError(f"{model} down")
        return results[model]

    return one, calls


def test_judge_gq_agree_adopts_gemini():
    g = _verdict({"A": 9, "B": 3}, "A")
    q = _verdict({"A": 7, "B": 6}, "A")
    one, calls = _one_factory({mg.JUDGE_MODEL: g, mg.QWEN_JUDGE_MODEL: q})
    res = mg._judge_gq(
        [mg.JUDGE_MODEL, mg.QWEN_JUDGE_MODEL], one, [], ["A", "B"])
    # 동시 판정 — margin-skip 없이 둘 다 불렀어야 한다
    assert calls == [mg.JUDGE_MODEL, mg.QWEN_JUDGE_MODEL]
    assert res["winner"] == "A"
    assert res["gq"]["route"] == "agree"
    assert res["verdicts"] == g["verdicts"]  # Gemini 판정 그대로


def test_judge_gq_small_gap_gemini_priority():
    g = _verdict({"A": 9, "B": 8}, "A")
    q = _verdict({"A": 9, "B": 10}, "B")  # gap = (10-9)/10 = 0.1 < 0.2
    one, _ = _one_factory({mg.JUDGE_MODEL: g, mg.QWEN_JUDGE_MODEL: q})
    res = mg._judge_gq(
        [mg.JUDGE_MODEL, mg.QWEN_JUDGE_MODEL], one, [], ["A", "B"])
    assert res["winner"] == "A"  # Gemini 우선
    assert res["gq"]["route"] == "gemini_priority"
    assert res["gq"]["gap"] == pytest.approx(0.1)
    assert res["gq"]["per_model_winner"] == {
        mg.JUDGE_MODEL: "A", mg.QWEN_JUDGE_MODEL: "B"}


def test_judge_gq_large_gap_goes_combined():
    g = _verdict({"A": 9, "B": 8}, "A")
    q = _verdict({"A": 2, "B": 10}, "B")  # gap = 0.8 ≥ 0.2
    one, _ = _one_factory({mg.JUDGE_MODEL: g, mg.QWEN_JUDGE_MODEL: q})
    res = mg._judge_gq(
        [mg.JUDGE_MODEL, mg.QWEN_JUDGE_MODEL], one, [], ["A", "B"])
    assert res["gq"]["route"] == "combined"
    # 합산: A = 9/9 + 2/10 = 1.2, B = 8/9 + 10/10 ≈ 1.889 → B 승
    assert res["winner"] == "B"
    assert res["gq"]["gap"] == pytest.approx(0.8)
    assert "dual" in res  # combine_select_verdicts 산출물이 그대로 실린다


def test_judge_gq_one_side_down_uses_survivor():
    q = _verdict({"A": 5, "B": 7}, "B")
    one, _ = _one_factory(
        {mg.QWEN_JUDGE_MODEL: q}, fails={mg.JUDGE_MODEL})
    res = mg._judge_gq(
        [mg.JUDGE_MODEL, mg.QWEN_JUDGE_MODEL], one, [], ["A", "B"])
    assert res["winner"] == "B"
    assert res["gq"]["route"] == f"single_{mg.QWEN_JUDGE_MODEL}"


def test_judge_gq_all_down_raises_with_cause():
    one, _ = _one_factory(
        {}, fails={mg.JUDGE_MODEL, mg.QWEN_JUDGE_MODEL})
    with pytest.raises(RuntimeError) as ei:
        mg._judge_gq(
            [mg.JUDGE_MODEL, mg.QWEN_JUDGE_MODEL], one, [], ["A", "B"])
    assert ei.value.__cause__ is not None  # 원인 삼키지 않기


# ── resolve 층 — ON 구성 / OFF 불변 ──────────────────────────────────

def test_resolve_models_on_returns_gq(monkeypatch):
    monkeypatch.setattr(settings, "multiroll_gq_judge_enabled", True)
    monkeypatch.setattr(settings, "multiroll_gg46_judge_enabled", False)
    # 환경 핀 — 운영 .env 의 QK ON 과 상호배타 fail-closed 충돌 방지
    monkeypatch.setattr(settings, "multiroll_qk_judge_enabled", False)
    monkeypatch.setattr(settings, "dashscope_api_key", "k")
    assert mg.resolve_select_judge_models() == [
        mg.JUDGE_MODEL, mg.QWEN_JUDGE_MODEL]
    assert mg.resolve_select_judge_model() == mg.JUDGE_MODEL


def test_resolve_models_on_without_key_fails_closed(monkeypatch):
    monkeypatch.setattr(settings, "multiroll_gq_judge_enabled", True)
    monkeypatch.setattr(settings, "multiroll_gg46_judge_enabled", False)
    # 환경 핀 — 운영 .env 의 QK ON 과 상호배타 fail-closed 충돌 방지
    monkeypatch.setattr(settings, "multiroll_qk_judge_enabled", False)
    monkeypatch.setattr(settings, "dashscope_api_key", "")
    with pytest.raises(AppError):
        mg.resolve_select_judge_models()
    # physical 도 resolve 경유라 같은 자리에서 막힌다(지문 계산 fail-closed)
    with pytest.raises(AppError):
        mg.resolve_select_judge_model_physical()


def test_resolve_physical_on_is_gemini_plus_qwen(monkeypatch):
    monkeypatch.setattr(settings, "multiroll_gq_judge_enabled", True)
    monkeypatch.setattr(settings, "multiroll_gg46_judge_enabled", False)
    # 환경 핀 — 운영 .env 의 QK ON 과 상호배타 fail-closed 충돌 방지
    monkeypatch.setattr(settings, "multiroll_qk_judge_enabled", False)
    monkeypatch.setattr(settings, "dashscope_api_key", "k")
    phys = mg.resolve_select_judge_model_physical()
    assert phys == (
        f"{settings.gemini_text_model}+{settings.qwen_vlm_model}")


def test_resolve_off_keeps_legacy_paths(monkeypatch):
    """OFF = 기존 경로 byte-identical — 완료 산출 동결이 여기 걸려 있다."""
    monkeypatch.setattr(settings, "multiroll_gq_judge_enabled", False)
    monkeypatch.setattr(settings, "multiroll_gg46_judge_enabled", False)
    monkeypatch.setattr(settings, "multiroll_qk_judge_enabled", False)
    # anthropic 키 있음 + dual ON = Opus+Sol (기존 기본)
    monkeypatch.setattr(settings, "anthropic_api_key", "k")
    monkeypatch.setattr(settings, "multiroll_dual_select_judge_enabled", True)
    assert mg.resolve_select_judge_models() == [
        mg.SELECT_JUDGE_MODEL, mg.SELECT_JUDGE_MODEL_2]
    assert mg.resolve_select_judge_model() == mg.SELECT_JUDGE_MODEL
    assert mg.resolve_select_judge_model_physical() == (
        f"{settings.anthropic_judge_model}+{settings.openai_model}")
    # anthropic 키 없음 = gemini 단독 (기존 fail-open)
    monkeypatch.setattr(settings, "anthropic_api_key", "")
    assert mg.resolve_select_judge_models() == [mg.JUDGE_MODEL]
    assert mg.resolve_select_judge_model() == mg.JUDGE_MODEL
    assert mg.resolve_select_judge_model_physical() == (
        settings.gemini_text_model)


# ── make_gemini_judge_fn 디스패치 — Qwen 이 끼면 _judge_gq 로 ─────────

def test_judge_fn_dispatches_to_gq_when_qwen_present(
        monkeypatch, tmp_path):
    monkeypatch.setattr(settings, "multiroll_gq_judge_enabled", True)
    monkeypatch.setattr(settings, "multiroll_gg46_judge_enabled", False)
    # 환경 핀 — 운영 .env 의 QK ON 과 상호배타 fail-closed 충돌 방지
    monkeypatch.setattr(settings, "multiroll_qk_judge_enabled", False)
    monkeypatch.setattr(settings, "dashscope_api_key", "k")
    g = _verdict({"A": 9, "B": 3}, "A")
    q = _verdict({"A": 7, "B": 6}, "A")

    import app.modules.llm.llm_client as llm_client
    import app.modules.llm.qwen_vlm_client as qc

    def fake_structured(tag, sys_p, parts, schema, **kw):
        return dict(g)

    def fake_qwen(tag, sys_p, parts, schema, **kw):
        return dict(q)

    monkeypatch.setattr(llm_client, "call_structured", fake_structured)
    monkeypatch.setattr(qc, "ask_qwen_structured", fake_qwen)
    judge_fn = mg.make_gemini_judge_fn(
        judge_sys="JUDGE", judge_schema={"type": "object"},
        step_tag="t_judge")
    png = tmp_path / "a.png"
    png.write_bytes(b"\x89PNG\r\n\x1a\n" + b"0" * 16)
    res = judge_fn("t", "prompt", [], [png, png], ["A", "B"])
    assert res["gq"]["route"] == "agree"
    assert res["winner"] == "A"


# ── make_gq_critique_fn — 관찰 0건 조기 반환·취합 경로 ────────────────

def _wire_critique(monkeypatch, observations, compose_result):
    """관찰/취합 대역을 걸고 critique_fn 을 만든다. 대역은 외부 API 응답
    데이터일 뿐 — 파싱·조립·조기 반환 로직은 실물이 돈다."""
    import app.modules.llm.llm_client as llm_client
    import app.modules.llm.qwen_vlm_client as qc

    seen = {"observe": 0, "compose": 0, "compose_parts": None}

    def fake_qwen(tag, sys_p, parts, schema, **kw):
        seen["observe"] += 1
        # 관찰 스키마로 검증해 대역 데이터가 실계약과 맞는지 고정
        from jsonschema import validate
        payload = {"observations": observations}
        validate(payload, build_gq_observe_schema())
        return payload

    def fake_structured(tag, sys_p, parts, schema, **kw):
        seen["compose"] += 1
        seen["compose_parts"] = parts
        return dict(compose_result)

    monkeypatch.setattr(qc, "ask_qwen_structured", fake_qwen)
    monkeypatch.setattr(llm_client, "call_structured", fake_structured)
    fn = mg.make_gq_critique_fn(
        critique_schema={"type": "object"}, step_tag="t_crit")
    return fn, seen


def test_gq_critique_no_observation_skips_compose(monkeypatch, tmp_path):
    fn, seen = _wire_critique(monkeypatch, [], {"issues": [{"x": 1}]})
    png = tmp_path / "s.png"
    png.write_bytes(b"\x89PNG\r\n\x1a\n" + b"0" * 16)
    out = fn("t", "prompt", [], png)
    assert out == {"issues": [], "qwen_observations": []}
    assert seen["observe"] == 1
    assert seen["compose"] == 0  # 유료 취합 호출 생략


def test_gq_critique_composes_and_carries_observations(
        monkeypatch, tmp_path):
    obs = [{"issue_ko": "총구가 상대를 향하지 않음", "severity": "critical"}]
    compose = {"issues": [
        {"issue_ko": "총구 방향", "fix_en": "Aim the muzzle at ..."}]}
    fn, seen = _wire_critique(monkeypatch, obs, compose)
    png = tmp_path / "s.png"
    png.write_bytes(b"\x89PNG\r\n\x1a\n" + b"0" * 16)
    out = fn("t", "prompt", [], png)
    assert out["issues"] == compose["issues"]
    assert out["qwen_observations"] == obs  # 기각 추적용 원본 병기
    assert seen["compose"] == 1
    # 취합 입력에 관찰 목록이 실려야 한다
    texts = [p.get("text", "") for p in seen["compose_parts"]
             if p.get("type") == "text"]
    assert any("OBSERVATIONS" in t for t in texts)
    assert any("총구가 상대를 향하지 않음" in t for t in texts)


def test_gq_pack_stems_load_from_disk():
    """v8 팩 스템 2개가 실제로 로드된다 — 스템 부재는 여기서 잡힌다."""
    from app.modules.prompt_loader import load_prompt

    resolved = mg.resolve_judge_pack_version(mg.GQ_CRITIQUE_PACK_VERSION)
    observe = load_prompt(mg.JUDGE_MODULE, "gq_observe_sys", version=resolved)
    compose = load_prompt(mg.JUDGE_MODULE, "gq_compose_sys", version=resolved)
    assert "OBSERVER" in observe and "repair" in observe
    assert "COMPOSER" in compose and "fix_en" in compose
    # v8 은 추가-스템 전용 — v7 판정 계약 디렉토리는 불변이어야 한다
    assert mg.judge_pack_content_hash(
        mg.STILL_JUDGE_PACK_VERSION) != mg.judge_pack_content_hash(
        mg.GQ_CRITIQUE_PACK_VERSION)


def test_run_multiroll_select_preserves_gq_in_durable_record(tmp_path):
    """Codex BLOCK-3 회귀 — 실호출: _judge_gq 가 얹은 gq(route·격차·양쪽
    승자)가 durable record 에 남아야 한 심판 장애(single_*)·합의 경로가
    평범한 판정과 구분된다."""
    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)
        out_path.write_bytes(b"\x89PNG\r\n\x1a\n" + b"0" * 8)
        return out_path

    gq_meta = {"route": f"single_{mg.QWEN_JUDGE_MODEL}",
               "models": [mg.QWEN_JUDGE_MODEL]}

    def judge_fn(tag, prompt, labeled_refs, cand_paths, labels):
        return {**_verdict({"A": 5, "B": 9}, "B"), "gq": dict(gq_meta)}

    (tmp_path / "ref.png").write_bytes(b"ref")
    _sel, record = run_multiroll_select(
        tag="t1", prompt="P",
        labeled_refs=[("REF", tmp_path / "ref.png")],
        out_stem=tmp_path / "out" / "s1",
        gen_fn=gen_fn, judge_fn=judge_fn,
        critique_fn=None, fix_gen_fn=None,
        roll_count=2, critique_enabled=False,
        fix_head="H", fix_tail="T", fix_label="L",
        record=None,
    )
    assert record["gq"] == gq_meta
    assert record["selected"] == "B"


def test_observe_schema_shape():
    s = build_gq_observe_schema()
    # 관찰자 계약: 수정문·수정 가능성 필드가 없어야 한다(취합자 몫)
    item = s["properties"]["observations"]["items"]["properties"]
    assert set(item) == {"issue_ko", "severity"}
    assert json.dumps(s)  # 직렬화 가능(스키마 동봉 경로)