"""QK 판정 체계 (2026-08-12, 같은 날 밤 반전 재결정) — 구성·라우팅·OFF 불변.

사용자 확정 이력: "qwen 3.8 max main -> kimi k3 보조" → 카나리아 실측
(OpenRouter 경유 Kimi 평균 245s·Qwen 207s+비JSON 50%)으로 같은 날 밤
재결정 "gemini 를 보조로 하고 기존 코드에서 QWEN 의 비중을 반대로".

계약 요점:
  · ON: 선정 = [qwen-vlm(DashScope, 메인·취합), gemini(JUDGE_MODEL,
    보조·관찰)] 동시 — 합의 수학은 `_judge_gq` 공유(슬롯 순서=우선권),
    우선 채택 route 만 "qwen_priority".
    수정 = Gemini 관찰 → Qwen 취합 (GQ 2단 계약의 자리 반전, 팩 v9 도입판).
  · OFF: 기존 경로 그대로 — 반환값·지문 재료가 1비트도 안 움직여야 한다.
  · GQ 와 동시 ON = fail-closed (판정 체계는 하나만).
"""
from __future__ import annotations

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) -> Dict[str, Any]:
    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": [],
        "all_candidates_fail": False,
    }


def _qk_on(monkeypatch):
    monkeypatch.setattr(settings, "multiroll_qk_judge_enabled", True)
    monkeypatch.setattr(settings, "multiroll_gq_judge_enabled", False)
    monkeypatch.setattr(settings, "multiroll_gg46_judge_enabled", False)
    monkeypatch.setattr(settings, "dashscope_api_key", "k")


# ── resolve 층 ────────────────────────────────────────────────────────

def test_resolve_models_qk_on_returns_qwen_main_gemini_assist(monkeypatch):
    """반전 계약: 메인(models[0])=Qwen(DashScope 센티널), 보조=Gemini —
    기존 G+Q 목록의 정확한 역순이어야 한다(기존 코드 재활용이 계약)."""
    _qk_on(monkeypatch)
    models = mg.resolve_select_judge_models()
    assert models == [mg.QWEN_JUDGE_MODEL, mg.JUDGE_MODEL]
    # alias(지문 재료)도 QK 메인이어야 판정 교체가 산출 무효화로 이어진다
    assert mg.resolve_select_judge_model() == mg.QWEN_JUDGE_MODEL
    # physical 스탬프 — qwen 물리 모델+gemini 물리 모델 쌍
    phys = mg.resolve_select_judge_model_physical()
    assert settings.qwen_vlm_model in phys
    assert settings.gemini_text_model in phys


def test_resolve_qk_on_without_dashscope_key_fails_closed(monkeypatch):
    monkeypatch.setattr(settings, "multiroll_qk_judge_enabled", True)
    monkeypatch.setattr(settings, "multiroll_gq_judge_enabled", False)
    monkeypatch.setattr(settings, "multiroll_gg46_judge_enabled", False)
    monkeypatch.setattr(settings, "dashscope_api_key", "")
    with pytest.raises(AppError):
        mg.resolve_select_judge_models()


def test_resolve_qk_blank_physical_model_slots_fail_closed(monkeypatch):
    """물리 모델 슬롯 공백 preflight (Codex 반전 리뷰 BLOCK-1): 키만 보면
    qwen_vlm_model 공백이 통과해 빈 model 호출 실패를 _judge_gq 가 삼키고
    Gemini 단독(single_*)으로 확정된다 — 양방향 모두 첫 호출 전 차단."""
    _qk_on(monkeypatch)
    monkeypatch.setattr(settings, "qwen_vlm_model", "   ")
    with pytest.raises(AppError) as ei:
        mg.resolve_select_judge_models()
    assert "empty_model_slot" in getattr(ei.value, "code", "")
    monkeypatch.setattr(settings, "qwen_vlm_model", "qwen3.8-max")
    monkeypatch.setattr(settings, "gemini_text_model", "")
    with pytest.raises(AppError):
        mg.resolve_select_judge_models()


def test_resolve_qk_and_gq_both_on_fails_closed(monkeypatch):
    monkeypatch.setattr(settings, "multiroll_qk_judge_enabled", True)
    monkeypatch.setattr(settings, "multiroll_gq_judge_enabled", True)
    monkeypatch.setattr(settings, "multiroll_gg46_judge_enabled", False)
    monkeypatch.setattr(settings, "dashscope_api_key", "k")
    with pytest.raises(AppError) as ei:
        mg.resolve_select_judge_models()
    assert "judge_flags_conflict" in getattr(ei.value, "code", "")


def test_resolve_qk_off_is_byte_identical_legacy(monkeypatch):
    """QK OFF = 기존 경로 그대로 — 완료 산출 동결이 이 기본값에 걸려 있다."""
    monkeypatch.setattr(settings, "multiroll_qk_judge_enabled", False)
    monkeypatch.setattr(settings, "multiroll_gq_judge_enabled", False)
    monkeypatch.setattr(settings, "multiroll_gg46_judge_enabled", False)
    monkeypatch.setattr(settings, "anthropic_api_key", "")
    assert mg.resolve_select_judge_models() == [mg.JUDGE_MODEL]
    assert mg.resolve_select_judge_model() == mg.JUDGE_MODEL


# ── _judge_gq — 슬롯 순서=우선권 / route 이름 ─────────────────────────

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_priority_route_default_unchanged():
    """priority_route 기본값 = 기존 GQ 기록값 byte-identical."""
    g = _verdict({"A": 9, "B": 8}, "A")
    q = _verdict({"A": 9, "B": 10}, "B")  # gap 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["gq"]["route"] == "gemini_priority"


def test_judge_gq_qk_reversed_slots_qwen_priority():
    """QK 반전: models[0]=Qwen 이 우선 — 불일치·격차<문턱이면 Qwen 채택,
    격차는 보조(Gemini) 눈으로 잰다 (_judge_gq 슬롯 계약 재활용)."""
    q = _verdict({"A": 9, "B": 8}, "A")   # 메인(Qwen)
    g = _verdict({"A": 9, "B": 10}, "B")  # 보조(Gemini) gap 0.1 < 0.2
    one, calls = _one_factory({mg.QWEN_JUDGE_MODEL: q, mg.JUDGE_MODEL: g})
    res = mg._judge_gq([mg.QWEN_JUDGE_MODEL, mg.JUDGE_MODEL], one, [],
                       ["A", "B"], priority_route="qwen_priority")
    assert calls == [mg.QWEN_JUDGE_MODEL, mg.JUDGE_MODEL]
    assert res["winner"] == "A"  # 메인(Qwen) 우선
    assert res["gq"]["route"] == "qwen_priority"


# ── make_gemini_judge_fn 디스패치 — Qwen 센티널이 앞이면 QK ──────────

def test_judge_fn_qk_dispatch_and_route(monkeypatch, tmp_path):
    _qk_on(monkeypatch)
    q = _verdict({"A": 9, "B": 3}, "A")
    g = _verdict({"A": 7, "B": 6}, "A")

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

    seen = {"qwen": 0, "gemini_model": None, "gemini_fallback": "미전달"}

    def fake_qwen(tag, sys_p, parts, schema, **kw):
        seen["qwen"] += 1
        return dict(q)

    def fake_structured(tag, sys_p, parts, schema, **kw):
        seen["gemini_model"] = (kw.get("project_config") or {}).get(
            tag, {}).get("model")
        seen["gemini_fallback"] = kw.get("enable_fallback", "미전달")
        return dict(g)

    monkeypatch.setattr(qc, "ask_qwen_structured", fake_qwen)
    monkeypatch.setattr(llm_client, "call_structured", fake_structured)
    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"])
    # Qwen(DashScope) 1콜 + Gemini(call_structured, JUDGE_MODEL) 1콜
    assert seen["qwen"] == 1
    assert seen["gemini_model"] == mg.JUDGE_MODEL
    # Codex BLOCK-2: QK 반전의 Gemini 보조는 Gemini 로 봉인 — safety
    # 강등(GPT Tier3)이 Gemini 판정으로 오귀속되는 창 차단
    assert seen["gemini_fallback"] is False
    assert res["gq"]["route"] == "agree"
    assert res["winner"] == "A"


def test_judge_fn_gq_classic_unchanged(monkeypatch, tmp_path):
    """기존 G+Q(Gemini 앞) 라우팅은 반전 도입 후에도 byte-identical —
    gemini_priority 계열 그대로."""
    monkeypatch.setattr(settings, "multiroll_qk_judge_enabled", False)
    monkeypatch.setattr(settings, "multiroll_gq_judge_enabled", True)
    monkeypatch.setattr(settings, "multiroll_gg46_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

    seen = {"fallback": "미전달"}

    def fake_structured(*a, **k):
        seen["fallback"] = k.get("enable_fallback", "미전달")
        return dict(g)

    monkeypatch.setattr(qc, "ask_qwen_structured",
                        lambda *a, **k: dict(q))
    monkeypatch.setattr(llm_client, "call_structured", fake_structured)
    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["verdicts"] == g["verdicts"]  # Gemini(우선) 판정 그대로
    # 기존 GQ 는 fallback 기본값 그대로(미전달) — byte-identical 보존
    assert seen["fallback"] == "미전달"


# ── make_qk_critique_fn — Gemini 관찰 → Qwen 취합 (자리 반전) ─────────

def _wire_qk_critique(monkeypatch, observations, compose_result):
    """관찰(Gemini=call_structured)/취합(Qwen=ask_qwen_structured) 대역 —
    외부 API 응답 데이터일 뿐, 조기 반환·조립 로직은 실물이 돈다."""
    import app.modules.llm.llm_client as llm_client
    import app.modules.llm.qwen_vlm_client as qc

    seen = {"observe": 0, "observe_model": None, "compose": 0,
            "compose_parts": None, "observe_fallback": "미전달"}

    def fake_observe(tag, sys_p, parts, schema, **kw):
        seen["observe"] += 1
        seen["observe_model"] = (kw.get("project_config") or {}).get(
            tag, {}).get("model")
        seen["observe_fallback"] = kw.get("enable_fallback", "미전달")
        from jsonschema import validate

        payload = {"observations": observations}
        validate(payload, build_gq_observe_schema())
        return payload

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

    monkeypatch.setattr(llm_client, "call_structured", fake_observe)
    monkeypatch.setattr(qc, "ask_qwen_structured", fake_compose)
    return seen


def test_qk_critique_no_observations_skips_compose(monkeypatch):
    seen = _wire_qk_critique(monkeypatch, [], {"issues": []})
    fn = mg.make_qk_critique_fn(critique_schema={"type": "object"})
    out = fn("t", "prompt", [], __file__)
    # 관찰 0건 = 취합(Qwen) 미호출 — 유료 1콜 절약 계약 (GQ 동형)
    assert seen["observe"] == 1
    assert seen["observe_model"] == mg.JUDGE_MODEL
    # Codex BLOCK-2: 관찰자도 Gemini 로 봉인(fallback OFF) — GPT 강등
    # 관찰이 observer_model='gemini-pro' 로 영속되는 오귀속 차단
    assert seen["observe_fallback"] is False
    assert seen["compose"] == 0
    assert out == {"issues": [], "observer_observations": [],
                   "observer_model": mg.JUDGE_MODEL}


def test_qk_critique_composes_with_qwen_and_records_observer(monkeypatch):
    obs = [{"issue_ko": "왼손이 지지 없이 공중에 떠 있다",
            "severity": "major"}]
    seen = _wire_qk_critique(
        monkeypatch, obs, {"issues": [{"fix_en": "ground the hand"}]})
    fn = mg.make_qk_critique_fn(critique_schema={"type": "object"})
    out = fn("t", "prompt", [], __file__)
    # 관찰(Gemini) → 취합(Qwen) 순서
    assert seen["observe"] == 1 and seen["compose"] == 1
    compose_text = "\n".join(
        str(p.get("text") or "") for p in seen["compose_parts"]
        if p.get("type") == "text")
    assert "OBSERVATIONS" in compose_text
    assert out["issues"] == [{"fix_en": "ground the hand"}]
    assert out["observer_observations"] == obs
    assert out["observer_model"] == mg.JUDGE_MODEL


def test_qk_critique_severity_rebinds_to_observer(monkeypatch):
    """Codex R1 BLOCK-1: 심각도의 SOT=관찰자 — 취합이 major→critical 로
    승격해도 코드가 관찰 원값으로 되돌리고(유료 fix 재발동 창 제거),
    관찰에 없는 인덱스의 발명 이슈는 fail-closed 로 버린다."""
    obs = [
        {"issue_ko": "가구 소실", "severity": "major"},
        {"issue_ko": "다른 인물", "severity": "critical"},
    ]
    _wire_qk_critique(monkeypatch, obs, {"issues": [
        {"issue_ko": "가구 소실", "fix_en": "restore it",
         "severity": "critical", "observation_index": 0},   # 승격 시도
        {"issue_ko": "다른 인물", "fix_en": "match the reference",
         "severity": "critical", "observation_index": 1},   # 정합
        {"issue_ko": "발명 이슈", "fix_en": "invented",
         "severity": "critical", "observation_index": 7},   # 미등록
    ]})
    fn = mg.make_qk_critique_fn(critique_schema={"type": "object"})
    out = fn("t", "prompt", [], __file__)
    assert [i["severity"] for i in out["issues"]] == ["major", "critical"]
    assert out["compose_severity_rebound_count"] == 1
    assert [d["issue_ko"] for d in out["compose_dropped_unbound"]] == [
        "발명 이슈"]


def test_qk_critique_without_severity_keeps_legacy_shape(monkeypatch):
    """severity 를 안 쓰는 스키마(구 경로) 취합 반환은 재결속 블록을 타지
    않는다 — 기존 계약 그대로."""
    obs = [{"issue_ko": "관찰", "severity": "major"}]
    _wire_qk_critique(monkeypatch, obs, {"issues": [
        {"issue_ko": "관찰", "fix_en": "fix it"}]})
    fn = mg.make_qk_critique_fn(critique_schema={"type": "object"})
    out = fn("t", "prompt", [], __file__)
    assert out["issues"] == [{"issue_ko": "관찰", "fix_en": "fix it"}]
    assert "compose_dropped_unbound" not in out


def test_qk_critique_pack_is_v10_and_gq_pack_untouched():
    """QK 현행=v10 (2026-08-13 #106: 심각도 구조 전달+LOOK 축 2종 — v9
    STAGING 축 승계), GQ 는 v8 고정 — 어떤 QK 승격도 GQ 지문을 건드리지
    않음을 상수로 잠근다."""
    assert mg.QK_CRITIQUE_PACK_VERSION == "10"
    assert mg.GQ_CRITIQUE_PACK_VERSION == "8"


def test_qk_select_policy_version_reflects_sealed_contract():
    """정책 문자열은 실행 계약을 대변해야 한다 — v1(Kimi 보조)·v2(Gemini
    보조, GPT 강등 허용) 산출이 v3(Gemini 봉인, no_fallback) 산출로
    오독되지 않게 셋 다 갈라져 있어야 한다 (Codex 재확인 BLOCK)."""
    v = mg.QK_SELECT_POLICY_VERSION
    assert "v3" in v
    assert "qwen_main" in v
    assert "no_fallback" in v  # fallback 봉인 의미가 문자열에 잠김
    assert "v1" not in v and "v2" not in v