"""G+G46 판정 체계 (2026-08-13 사용자 확정 "Gemini + Grok 4.6 을 50:50으로,
수정사항을 취합해서 다시 Grok 으로 수정하고 재평가") — 구성·동등 라우팅·
양쪽 관찰 critique·fix gen 분리·OFF 불변.

계약 요점:
  · ON: 선정 = [gemini(JUDGE_MODEL), openrouter:grok-4.6] 동시 **동등** —
    합의 수학은 `_judge_gq` 공유하되 우선권 갈래가 없다: 일치=채택(agree),
    불일치=combined 합산(격차 무관). 수정 = 양쪽 관찰 → Gemini 취합 →
    fix i2i 는 Grok(별도 gen 주입) → fix 재판정도 같은 동등 이중.
  · OFF: 기존 경로 그대로 — 반환값·지문 재료가 1비트도 안 움직여야 한다.
  · GQ/QK 와 동시 ON = fail-closed (판정 체계는 하나만).
"""
from __future__ import annotations

from pathlib import Path
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

GROK_SENTINEL = mg.OPENROUTER_JUDGE_PREFIX + "x-ai/grok-4.6"


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 _gg46_on(monkeypatch):
    monkeypatch.setattr(settings, "multiroll_gg46_judge_enabled", True)
    monkeypatch.setattr(settings, "multiroll_qk_judge_enabled", False)
    monkeypatch.setattr(settings, "multiroll_gq_judge_enabled", False)
    monkeypatch.setattr(settings, "openrouter_api_key", "k")
    monkeypatch.setattr(settings, "grok_judge_model", "x-ai/grok-4.6")


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

def test_resolve_models_select_pair_is_gemini_plus_sol(monkeypatch):
    """★2026-08-29 사용자 지시 — **고르는 자리는 Gemini + GPT Sol**.

    수정 관찰은 Gemini+Grok 그대로다(`make_gg46_critique_fn` 이 자기
    목록을 따로 만든다). 여기서 재는 것은 **선정** 슬롯 하나뿐이다.
    models[0]=Gemini 는 우선권이 아니라 agree 시 shape 제공 슬롯이고,
    physical 은 두 물리 모델 쌍이라 어느 한쪽 교체도 지문이 잡는다.
    """
    _gg46_on(monkeypatch)
    models = mg.resolve_select_judge_models()
    assert models == [mg.JUDGE_MODEL, mg.SELECT_JUDGE_MODEL_2]
    assert GROK_SENTINEL not in models, (
        "선정 슬롯에 grok 이 남았다 — 수정 관찰과 섞였다")
    assert mg.resolve_select_judge_model() == mg.JUDGE_MODEL
    phys = mg.resolve_select_judge_model_physical()
    assert settings.gemini_text_model in phys
    assert settings.openai_model in phys, (
        "물리 지문에 둘째 심판이 안 실렸다 — 판정자를 바꿔도 완주 산출이 "
        "stale 이 안 된다")


def test_resolve_select_without_openai_key_fails_closed(monkeypatch):
    """둘째 심판이 GPT 이므로 **OpenAI 키**가 없으면 막는다.

    ★키 확인은 `settings.openai_api_key`(1차 슬롯 필드)가 아니라
    `has_openai_key()` 로 한다 — 보조 슬롯·환경변수만 있는 정상 구성을
    「키 없음」으로 막던 결함이 `openai_keys.py:486` 에 기록돼 있다.
    """
    _gg46_on(monkeypatch)
    monkeypatch.setattr(
        "app.core.openai_keys.has_openai_key", lambda: False)
    with pytest.raises(AppError) as ei:
        mg.resolve_select_judge_models()
    assert "select_judge_unconfigured" in getattr(ei.value, "code", "")


def test_resolve_select_no_longer_requires_openrouter(monkeypatch):
    """★선정은 grok 을 안 쓴다 — 안 쓰는 열쇠를 요구하지 않는다.

    종전에는 OPENROUTER_API_KEY 가 비면 **선정까지** 422 였다. 수정
    단계가 꺼져 있으면 그 열쇠를 아무도 안 쓰는데도 그랬다. 확인은
    실제 소비처(`make_gg46_critique_fn`)로 옮겼다.
    """
    _gg46_on(monkeypatch)
    monkeypatch.setattr(settings, "openrouter_api_key", "")
    assert mg.resolve_select_judge_models() == [
        mg.JUDGE_MODEL, mg.SELECT_JUDGE_MODEL_2]


def test_resolve_blank_select_model_slots_fail_closed(monkeypatch):
    """두 물리 슬롯(Gemini·OpenAI) **각각** 빈 값이면 막는다."""
    _gg46_on(monkeypatch)
    monkeypatch.setattr(settings, "openai_model", "   ")
    with pytest.raises(AppError) as ei:
        mg.resolve_select_judge_models()
    assert "empty_model_slot" in getattr(ei.value, "code", "")
    monkeypatch.setattr(settings, "openai_model", "gpt-5.6-sol")
    monkeypatch.setattr(settings, "gemini_text_model", "")
    with pytest.raises(AppError):
        mg.resolve_select_judge_models()


@pytest.mark.parametrize("other_flag", [
    "multiroll_qk_judge_enabled", "multiroll_gq_judge_enabled"])
def test_resolve_gg46_conflicts_fail_closed(monkeypatch, other_flag):
    """판정 체계는 하나만 — GG46 은 QK·GQ 어느 쪽과도 동시 ON 금지."""
    _gg46_on(monkeypatch)
    monkeypatch.setattr(settings, other_flag, True)
    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_gg46_off_is_byte_identical_legacy(monkeypatch):
    """GG46 OFF = 기존 경로 그대로 — 완료 산출 동결이 이 기본값에 걸려 있다."""
    monkeypatch.setattr(settings, "multiroll_gg46_judge_enabled", False)
    monkeypatch.setattr(settings, "multiroll_qk_judge_enabled", False)
    monkeypatch.setattr(settings, "multiroll_gq_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 동등 모드 — 불일치=combined (우선 채택 갈래 없음) ───────

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_equal_agree_adopts_first_slot_shape():
    g = _verdict({"A": 9, "B": 3}, "A")
    x = _verdict({"A": 7, "B": 6}, "A")
    one, calls = _one_factory({mg.JUDGE_MODEL: g, GROK_SENTINEL: x})
    res = mg._judge_gq([mg.JUDGE_MODEL, GROK_SENTINEL], one, [],
                       ["A", "B"], equal_disagree_combined=True)
    assert calls == [mg.JUDGE_MODEL, GROK_SENTINEL]
    assert res["gq"]["route"] == "agree"
    assert res["verdicts"] == g["verdicts"]  # shape 제공 슬롯=models[0]
    assert res["gq"]["per_model_winner"] == {
        mg.JUDGE_MODEL: "A", GROK_SENTINEL: "A"}


def test_judge_gq_equal_small_gap_still_combined():
    """★동등 계약의 핵심: 불일치·격차<문턱(기존이면 우선 채택 구간)도
    combined 합산 — 동등 체계에는 격차 문턱이 정할 우선 심판이 없다."""
    g = _verdict({"A": 9, "B": 8}, "A")
    x = _verdict({"A": 9, "B": 10}, "B")  # 보조 눈 gap 0.1 < 0.2
    one, _ = _one_factory({mg.JUDGE_MODEL: g, GROK_SENTINEL: x})
    res = mg._judge_gq([mg.JUDGE_MODEL, GROK_SENTINEL], one, [],
                       ["A", "B"], equal_disagree_combined=True)
    assert res["gq"]["route"] == "combined"
    assert res["gq"]["per_model_winner"] == {
        mg.JUDGE_MODEL: "A", GROK_SENTINEL: "B"}
    # 합산 근거가 기록에 남는다 (combine_select_verdicts 관례)
    assert res["dual"]["models"] == [mg.JUDGE_MODEL, GROK_SENTINEL]
    # (Codex R1 BLOCK-1) verdicts 점수=합산값(adjusted×1000 정수) —
    # 실소비자(gemini_select·flip 합산)가 합산 승자를 그대로 뽑는 전제.
    # G: A=1.0/B=0.889, X: A=0.9/B=1.0 → adj A=1.9, B=1.889
    scores = {v["label"]: v["score"] for v in res["verdicts"]}
    assert scores == {"A": 1900, "B": 1889}
    from app.modules.pipeline.multiroll_select import gemini_select

    _totals, picked = gemini_select(res)
    assert picked == res["winner"]  # 소비자 선정 == 합산 승자 (일관)


def test_gg46_equal_combined_selected_follows_sum_not_gemini_top(
        monkeypatch, tmp_path):
    """(Codex R1 BLOCK-1 회귀 — 실호출) Codex 재현 수치: Gemini A=10/B=9,
    Grok A=1/B=10 → 합산 B 가 이긴다. 종전 shape 는 verdicts 에 Gemini
    원점수를 실어 run_multiroll_select 의 gemini_select 가 A 를 다시
    골랐다 — selected 까지 B 인 것을 잠근다."""
    from app.modules.pipeline.multiroll_select import run_multiroll_select

    _gg46_on(monkeypatch)
    g = _verdict({"A": 10, "B": 9}, "A")
    # ★2026-08-29: Grok 은 이제 **역순**을 본다 — 그 응답은 display 값이다.
    #  display A = canonical B 이므로, canonical 로 「B 를 크게 앞세우는」
    #  응답을 만들려면 display 에서 **A** 를 앞세워야 한다.
    #  display {A:10, B:1} · winner A  →  canonical {A:1, B:10} · winner B
    #  그러면 종전 시험과 **같은 상황**이 된다: Gemini top=A 인데 합산은 B.
    x = _verdict({"A": 10, "B": 1}, "A")

    import app.modules.llm.llm_client as llm_client

    # ★2026-08-29: 둘째 심판이 GPT 라 **두 슬롯 다 `call_structured`** 를
    #  탄다. 종전처럼 openrouter 대역만 놓으면 두 슬롯이 같은 응답을 받아
    #  agree 로 빠지고, 이 시험이 재려던 combined 갈래를 안 태운다.
    #  슬롯은 tag 접미사로 가른다 — 접미사는 **조립부와 같은 식**으로
    #  만든다(둘째 심판 이름이 바뀌면 손으로 박은 값이 조용히 안 맞는다).
    def _by_slot(tag, *a, **k):
        _sfx = "_" + mg.SELECT_JUDGE_MODEL_2.replace("-", "")
        return dict(x) if tag.endswith(_sfx) else dict(g)

    monkeypatch.setattr(llm_client, "call_structured", _by_slot)
    judge_fn = mg.make_gemini_judge_fn(
        judge_sys="JUDGE", judge_schema={"type": "object"},
        step_tag="t_judge")

    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" + tag.encode())
        return out_path

    _sel, record = 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=None, fix_gen_fn=None,
        roll_count=2, critique_enabled=False,
        fix_head="H", fix_tail="T", fix_label="L",
        record=None,
    )
    # ★계약은 그대로다 — **합산 승자가 실제 선정**이고 Gemini top 이 아니다.
    #  자리만 옮겼다: `gq` → `cross_model_order` (2026-08-29).
    assert record["cross_model_order"]["route"] == "cross_slot_combined"
    assert record["selected"] == "B"          # 합산 승자가 실제 선정
    # 합산 근거 durable — 슬롯별 승자가 **canonical 기준**으로 남는다
    assert record["cross_model_order"]["slot_winner"] == {
        mg.JUDGE_MODEL: "A", mg.SELECT_JUDGE_MODEL_2: "B"}


# ── fix 재평가 — 동등 합산이 최종 _sel 까지 지배 (Codex R1 BLOCK-1) ──

def _make_equal_rejudge_fn(g_scores: Dict[str, int],
                           x_scores: Dict[str, int]):
    """경로(원본/수정본)로 점수를 매기는 동등 재판정 — 정·역 라벨 스왑에
    안전(라벨이 아니라 실제 후보 파일로 판단). 실물 _judge_gq 수학이 돈다."""
    def fn(tag, prompt, labeled_refs, cand_paths, labels):
        def kind(p):
            return "fix" if str(p).endswith("_fix.png") else "orig"

        def one(model, parts, suffix):
            table = g_scores if model == mg.JUDGE_MODEL else x_scores
            scores = {lab: table[kind(p)]
                      for lab, p in zip(labels, cand_paths)}
            winner = max(scores, key=lambda lab: scores[lab])
            return _verdict(scores, winner)

        return mg._judge_gq(
            [mg.JUDGE_MODEL, GROK_SENTINEL], one, [], list(labels),
            equal_disagree_combined=True)
    return fn


def _run_fix_rejudge_case(tmp_path, g_scores, x_scores):
    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"FIXBYTES" 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):
        return _verdict({lab: (9 if lab == "A" else 3) for lab in labels},
                        "A")

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

    sel, record = 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=_make_equal_rejudge_fn(g_scores, x_scores),
        record=None,
    )
    return sel, record


def test_gg46_fix_rejudge_equal_sum_fix_wins(tmp_path):
    """Gemini 는 원본 선호(9:8)·Grok 은 수정본 강선호(1:10) — 동등 합산
    (원본 1.1 vs 수정본 1.889)이 수정본을 최종 _sel 로 확정해야 한다.
    종전 shape 면 Gemini 점수만 남아 원본이 이겼다."""
    sel, record = _run_fix_rejudge_case(
        tmp_path, g_scores={"orig": 9, "fix": 8},
        x_scores={"orig": 1, "fix": 10})
    assert record["fix_rejudge"]["winner"] == "B"
    assert sel.read_bytes().endswith(b"FIXBYTES")  # 최종 bytes=수정본
    # provenance 헬퍼도 같은 승부를 본다 (BLOCK-2 단일 판정)
    from app.modules.pipeline.still_recipe import (
        effective_prompt_used,
        fix_stage_won,
        winner_exact_multiroll_tag,
    )

    assert fix_stage_won(record) is True
    assert winner_exact_multiroll_tag("t1", record) == "t1_fix"
    assert effective_prompt_used(record, "BASE") == record["fix_prompt"]


def test_gg46_fix_rejudge_fix_loses_keeps_roll_provenance(tmp_path):
    """둘 다 원본 선호 → 재판정 A(원본) 승 — 최종 bytes=선정 롤이고
    provenance 도 롤 호출·롤 프롬프트·fix_applied=False 여야 한다
    (Codex R1 BLOCK-2: 종전엔 fix_prompt 존재만으로 _fix 에 링크)."""
    sel, record = _run_fix_rejudge_case(
        tmp_path, g_scores={"orig": 10, "fix": 2},
        x_scores={"orig": 9, "fix": 3})
    assert record["fix_rejudge"]["winner"] == "A"
    assert b"ROLL:" in sel.read_bytes()            # 최종 bytes=선정 롤
    from app.modules.pipeline.still_recipe import (
        effective_prompt_used,
        fix_stage_won,
        winner_exact_multiroll_tag,
    )

    assert fix_stage_won(record) is False
    sel_lab = str(record["selected"]).lower()
    assert winner_exact_multiroll_tag("t1", record) == f"t1_{sel_lab}"
    assert effective_prompt_used(record, "BASE") == "BASE"  # 롤 프롬프트 층


# ── provenance 헬퍼 행렬 (Codex R1 BLOCK-2) ──────────────────────────

def test_fix_stage_won_matrix():
    from app.modules.pipeline.still_recipe import fix_stage_won

    assert fix_stage_won({}) is False                       # 수리 없음
    assert fix_stage_won({"fix_skipped": True}) is False    # 게이트 스킵
    assert fix_stage_won({"fix_prompt": "F"}) is True       # legacy 무재평가
    assert fix_stage_won(
        {"repair_mode": "regenerate", "regen_prompt": "R"}) is True
    assert fix_stage_won(
        {"fix_prompt": "F", "fix_rejudge": {"winner": "B"}}) is True
    assert fix_stage_won(
        {"fix_prompt": "F", "fix_rejudge": {"winner": "A"}}) is False


def test_winner_exact_tag_respects_rejudge_and_regen():
    from app.modules.pipeline.still_recipe import winner_exact_multiroll_tag

    won = {"fix_prompt": "F", "fix_rejudge": {"winner": "B"},
           "selected": "A"}
    lost = {"fix_prompt": "F", "fix_rejudge": {"winner": "A"},
            "selected": "A"}
    regen = {"repair_mode": "regenerate", "regen_prompt": "R",
             "fix_rejudge": {"winner": "B"}, "selected": "C"}
    assert winner_exact_multiroll_tag("t1", won) == "t1_fix"
    assert winner_exact_multiroll_tag("t1", lost) == "t1_a"
    assert winner_exact_multiroll_tag("t1", regen) == "t1_regen"
    assert winner_exact_multiroll_tag("t1", {"selected": ""}) == ""


def test_effective_prompt_used_fix_won_returns_repair_prompt():
    from app.modules.pipeline.still_recipe import effective_prompt_used

    won = {"fix_prompt": "REPAIR", "fix_rejudge": {"winner": "B"},
           "selected": "A", "roll_prompts": {"A": "ROLLP"}}
    lost = {"fix_prompt": "REPAIR", "fix_rejudge": {"winner": "A"},
            "selected": "A", "roll_prompts": {"A": "ROLLP"}}
    assert effective_prompt_used(won, "BASE") == "REPAIR"
    assert effective_prompt_used(lost, "BASE") == "ROLLP"  # 변형 모드 층 유지


def test_judge_gq_equal_large_gap_combined_unchanged():
    g = _verdict({"A": 9, "B": 2}, "A")
    x = _verdict({"A": 1, "B": 10}, "B")  # gap 0.9 ≥ 0.2
    one, _ = _one_factory({mg.JUDGE_MODEL: g, GROK_SENTINEL: x})
    res = mg._judge_gq([mg.JUDGE_MODEL, GROK_SENTINEL], one, [],
                       ["A", "B"], equal_disagree_combined=True)
    assert res["gq"]["route"] == "combined"


def test_judge_gq_equal_one_side_down_single_route():
    g = _verdict({"A": 9, "B": 3}, "A")
    one, _ = _one_factory({mg.JUDGE_MODEL: g}, fails={GROK_SENTINEL})
    res = mg._judge_gq([mg.JUDGE_MODEL, GROK_SENTINEL], one, [],
                       ["A", "B"], equal_disagree_combined=True)
    assert res["gq"]["route"] == f"single_{mg.JUDGE_MODEL}"


def test_judge_gq_default_priority_math_unchanged():
    """equal 파라미터 기본값(False) = 기존 GQ/QK 수학 byte-identical —
    불일치·격차<문턱은 여전히 우선 채택(gemini_priority)."""
    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"


# ── make_gemini_judge_fn 디스패치 — Gemini 앞+OpenRouter 동승=동등 ────

def test_judge_fn_gg46_dispatch_equal_route_and_sealed_gemini(
        monkeypatch, tmp_path):
    """디스패치 전 경로 검증: ①grok 은 openrouter 클라이언트로(모델 문자열
    prefix 제거, max_tokens=8000 파일럿 구성) ②Gemini 판정은 봉인
    (enable_fallback=False — GPT 강등 오귀속 차단, QK BLOCK-2 동류)
    ③불일치·소격차인데 combined (동등 계약이 디스패치로 이어짐)."""
    _gg46_on(monkeypatch)
    g = _verdict({"A": 9, "B": 8}, "A")
    # ★2026-08-29: Grok 은 **역순**을 보므로 이 응답은 display 값이다.
    #  display {A:10, B:9} · winner A  →  canonical {A:9, B:10} · winner B.
    #  그래야 종전 시험과 **같은 상황**(두 심판이 갈림)이 되어 combined 로 간다.
    x = _verdict({"A": 10, "B": 9}, "A")

    import app.modules.llm.llm_client as llm_client
    import app.modules.llm.openrouter_vlm_client as orc

    seen: Dict[str, Any] = {}

    def fake_structured(tag, sys_p, parts, schema, **kw):
        # ★접미사는 **조립부와 같은 식**으로 만든다 — 손으로 박으면 둘째
        #  심판 이름이 바뀔 때(`gpt` → `gpt-high`) 조용히 안 맞는다.
        _sfx = "_" + mg.SELECT_JUDGE_MODEL_2.replace("-", "")
        slot = "gpt" if tag.endswith(_sfx) else "gemini"
        seen[f"{slot}_fallback"] = kw.get("enable_fallback", "미전달")
        seen[f"{slot}_retries"] = kw.get("num_retries", "미전달")
        return dict(x) if slot == "gpt" else dict(g)

    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"])
    # ★**두 슬롯 다** 봉인돼야 한다 (2026-08-29 Codex BLOCK-4).
    #  둘째가 OpenRouter 였을 때는 그 갈래가 Router 를 안 타서 Gemini 만
    #  봉인해도 「정확히 두 번」이 성립했다. GPT alias 는 Router 를 타므로
    #  Tier2/3 강등이 열리고, `enable_fallback=False` 만으로는 Router
    #  재시도(기본 3회)가 안 막힌다 — `num_retries=0` 이 같이 가야 한다.
    for slot in ("gemini", "gpt"):
        assert seen[f"{slot}_fallback"] is False, (
            f"{slot} 슬롯이 안 봉인됐다 — 강등 판정이 이 슬롯 이름으로 "
            "기록되면 오귀속이다")
        assert seen[f"{slot}_retries"] == 0, (
            f"{slot} 슬롯에 num_retries=0 이 안 갔다 — 한 alias 가 최대 "
            "4번 나가고 usage_sink 는 마지막만 본다")
    # ★결합 기록의 **자리와 이름이 바뀌었다** (2026-08-29 사용자 지시).
    #  `gq.route="combined"` → `cross_model_order.route="cross_slot_combined"`.
    #  종전은 「한 방향 안에서 두 모델」이었고 지금은 「Gemini 정순 · Grok
    #  역순」이라 **뜻이 다르다** — 이름을 물려 쓰면 기록이 거짓말을 한다.
    cmo = res["cross_model_order"]
    assert cmo["route"] == "cross_slot_combined"
    assert set(cmo["models"]) == {mg.JUDGE_MODEL, mg.SELECT_JUDGE_MODEL_2}
    assert "gq" not in res, "옛 `gq` 키를 아직 쓴다"


# ── make_gg46_critique_fn — 양쪽 관찰 → Gemini 취합 ──────────────────

def _wire_gg46_critique(monkeypatch, gemini_obs, grok_obs, compose_result,
                        grok_raises: bool = False,
                        gemini_observe_raises: bool = False):
    """관찰 2종/취합 대역 — 외부 API 응답 데이터일 뿐, 병합·조기 반환·
    재결속 로직은 실물이 돈다. Gemini 호출은 tag 로 관찰/취합을 가른다."""
    import app.modules.llm.llm_client as llm_client
    import app.modules.llm.openrouter_vlm_client as orc

    seen: Dict[str, Any] = {
        "observe_g": 0, "observe_x": 0, "compose": 0,
        "compose_parts": None, "fallbacks": [], "or_max_tokens": None,
    }

    def fake_structured(tag, sys_p, parts, schema, **kw):
        seen["fallbacks"].append(kw.get("enable_fallback", "미전달"))
        # ★[2026-09-08] 둘째 관찰자가 grok(OpenRouter) → gpt-6-astra(high) 로
        #  바뀌어 **둘 다 이 함수로 온다.** 태그로 가른다.
        if tag.endswith("_observe_2"):
            seen["observe_x"] += 1
            if grok_raises:
                raise RuntimeError("second observer down")
            return {"observations": list(grok_obs)}
        if tag.endswith("_observe"):
            seen["observe_g"] += 1
            if gemini_observe_raises:
                raise RuntimeError("gemini observer down")
            from jsonschema import validate

            payload = {"observations": list(gemini_obs)}
            validate(payload, build_gq_observe_schema())
            return payload
        assert tag.endswith("_compose")
        seen["compose"] += 1
        seen["compose_parts"] = parts
        return dict(compose_result)

    def fake_openrouter(tag, sys_p, parts, schema, *, model,
                        max_tokens=None, **kw):
        # ★이제 여기로 오면 **결함**이다 — 둘째 관찰자는 Router 로 가야 한다.
        seen["or_called"] = True
        raise AssertionError("관찰이 OpenRouter 로 나갔다")

    monkeypatch.setattr(llm_client, "call_structured", fake_structured)
    monkeypatch.setattr(orc, "ask_openrouter_structured", fake_openrouter)
    return seen


def test_gg46_critique_merges_both_observers_slot_order(monkeypatch):
    _gg46_on(monkeypatch)
    g_obs = [{"issue_ko": "왼손 부유", "severity": "critical"}]
    x_obs = [{"issue_ko": "좌석 역방향 착석", "severity": "critical"}]
    seen = _wire_gg46_critique(
        monkeypatch, g_obs, x_obs, {"issues": []})
    fn = mg.make_gg46_critique_fn(critique_schema={"type": "object"})
    out = fn("t", "prompt", [], __file__)
    assert seen["observe_g"] == 1 and seen["observe_x"] == 1
    assert seen["compose"] == 1
    assert "or_called" not in seen, "관찰이 OpenRouter 로 나갔다"
    # 병합 순서=모델 슬롯 순서(Gemini 먼저) — 인덱스 계약의 전제
    assert out["observer_observations"] == g_obs + x_obs
    assert out["observer_models"] == [mg.JUDGE_MODEL, mg.SELECT_JUDGE_MODEL_2]
    assert out["observer_counts"] == {
        mg.JUDGE_MODEL: 1, mg.SELECT_JUDGE_MODEL_2: 1}
    assert "observer_failed" not in out
    compose_text = "\n".join(
        str(p.get("text") or "") for p in seen["compose_parts"]
        if p.get("type") == "text")
    assert "TWO independent" in compose_text
    # 관찰자·취합자 Gemini 호출 전부 봉인
    # ★관찰자 둘 + 취합 하나 = 세 호출이 모두 Router 를 탄다(종전에는 둘째
    #  관찰자가 OpenRouter 라 두 개였다). 셋 다 봉인돼야 한다 — 강등되면
    #  기록의 모델 이름이 거짓이 된다.
    assert seen["fallbacks"] == [False, False, False]


def test_gg46_critique_no_observations_skips_compose(monkeypatch):
    _gg46_on(monkeypatch)
    seen = _wire_gg46_critique(monkeypatch, [], [], {"issues": []})
    fn = mg.make_gg46_critique_fn(critique_schema={"type": "object"})
    out = fn("t", "prompt", [], __file__)
    assert seen["compose"] == 0  # 관찰 0건 = 취합 미호출 (유료 1콜 절약)
    assert out["issues"] == []
    assert out["observer_observations"] == []


def test_gg46_critique_one_observer_down_survivor_and_record(monkeypatch):
    """관찰자 한쪽 실패 = 생존 관찰로 진행 + observer_failed 기록 —
    판정 single_* route 관례 동형 (기록이 정직한 강등)."""
    _gg46_on(monkeypatch)
    g_obs = [{"issue_ko": "왼손 부유", "severity": "critical"}]
    seen = _wire_gg46_critique(
        monkeypatch, g_obs, [], {"issues": []}, grok_raises=True)
    fn = mg.make_gg46_critique_fn(critique_schema={"type": "object"})
    out = fn("t", "prompt", [], __file__)
    assert seen["compose"] == 1
    assert out["observer_observations"] == g_obs
    assert out["observer_failed"] == [mg.SELECT_JUDGE_MODEL_2]
    assert out["observer_counts"][mg.SELECT_JUDGE_MODEL_2] == 0


def test_gg46_critique_both_observers_down_raises(monkeypatch):
    _gg46_on(monkeypatch)
    _wire_gg46_critique(monkeypatch, [], [], {"issues": []},
                        grok_raises=True, gemini_observe_raises=True)
    fn = mg.make_gg46_critique_fn(critique_schema={"type": "object"})
    with pytest.raises(RuntimeError):
        fn("t", "prompt", [], __file__)


def test_gg46_critique_severity_rebinds_across_merged_index(monkeypatch):
    """severity SOT=관찰자, 인덱스는 **병합 목록** 기준 — grok 관찰(병합
    인덱스 1)의 승격 시도를 원값으로 되돌리고 미등록 인덱스는 버린다."""
    _gg46_on(monkeypatch)
    g_obs = [{"issue_ko": "가구 소실", "severity": "major"}]
    x_obs = [{"issue_ko": "좌석 역방향", "severity": "major"}]
    _wire_gg46_critique(monkeypatch, g_obs, x_obs, {"issues": [
        {"issue_ko": "좌석 역방향", "fix_en": "seat them forward",
         "severity": "critical", "observation_index": 1},   # 승격 시도
        {"issue_ko": "발명 이슈", "fix_en": "invented",
         "severity": "critical", "observation_index": 9},   # 미등록
    ]})
    fn = mg.make_gg46_critique_fn(critique_schema={"type": "object"})
    out = fn("t", "prompt", [], __file__)
    assert [i["severity"] for i in out["issues"]] == ["major"]
    assert out["compose_severity_rebound_count"] == 1
    assert [d["issue_ko"] for d in out["compose_dropped_unbound"]] == [
        "발명 이슈"]


# ── 팩·정책 상수 — 신설이 기존 지문을 건드리지 않는다 ─────────────────

def test_gg46_pack_v12_and_existing_packs_untouched():
    assert mg.GG46_CRITIQUE_PACK_VERSION == "12"
    assert mg.QK_CRITIQUE_PACK_VERSION == "10"
    assert mg.GQ_CRITIQUE_PACK_VERSION == "8"
    assert mg.resolve_judge_pack_version("11") == "11.202608132045"
    assert mg.resolve_judge_pack_version("12") == "12.202608141305"


def test_gg46_pack_stems_copy_contract():
    """v11 = v10 사본 + 취합 스템만 2관찰자 개정 / v12 = v11 승계 +
    **표기 정책 축만** 관찰 스템 개정(2026-08-14) — 사본 계약이 살아야
    '어느 층이 바뀌었나'를 바이트로 되짚을 수 있다."""
    base = Path(__file__).resolve().parents[3] / "prompts" / "_base" \
        / "multiroll_judge"
    v10 = base / "10.202608131121"
    v11 = base / "11.202608132045"
    v12 = base / "12.202608141305"
    for stem in ("gq_observe_sys.md", "fix_rejudge_header.md"):
        assert (v11 / stem).read_bytes() == (v10 / stem).read_bytes()
    compose11 = (v11 / "gq_compose_sys.md").read_text("utf-8")
    assert compose11 != (v10 / "gq_compose_sys.md").read_text("utf-8")
    assert "TWO independent" in compose11
    assert "at most ONCE" in compose11
    # v12: 취합·재평가 헤더는 사본, 관찰만 표기 축 개정
    assert (v12 / "gq_compose_sys.md").read_bytes() == \
        (v11 / "gq_compose_sys.md").read_bytes()
    assert (v12 / "fix_rejudge_header.md").read_bytes() == \
        (v11 / "fix_rejudge_header.md").read_bytes()
    obs12 = (v12 / "gq_observe_sys.md").read_text("utf-8")
    assert obs12 != (v11 / "gq_observe_sys.md").read_text("utf-8")
    assert "caption or subtitle" in obs12          # 자막 굽기 critical 열거
    assert "is legitimate" in obs12                # 실물 원어 표기 정당화
    assert "no captions, subtitles" in obs12       # 배제 목록 축소판
    # judge_still 은 v7 바이트 사본 — 판정 계약 불변
    v7 = base / "7.202608071100"
    assert (v12 / "judge_still.md").read_bytes() == \
        (v7 / "judge_still.md").read_bytes()
    # fix_tail 은 표기 축 개정 — 실물 표기 보존·오버레이만 금지
    tail12 = (v12 / "fix_tail.md").read_text("utf-8")
    assert "stays exactly as the" in tail12
    assert "overlay text" in tail12


def test_gg46_select_policy_string_reflects_contract():
    v = mg.GG46_SELECT_POLICY_VERSION
    # ★v3 (2026-08-29) = 합산 계약이 바뀌었다: 승자가 같아도 합산을 태우고
    #  페널티가 후보당 한 번이다. 어느 롤이 뽑히는지가 바뀌므로 v2 산출
    #  지문이 무효화돼야 한다.
    assert "v3" in v and "v2" not in v
    assert "always_combine" in v and "binary_penalty" in v, (
        "합산 계약이 문자열에 안 적혔다")
    assert "equal" in v            # 우선권 없는 동등 이중
    assert "no_fallback" in v      # Gemini 판정 봉인 의미가 문자열에 잠김
    # ★`combined` 는 뺐다 — v3 부터 **승자가 같아도 합산을 태우므로**
    #  「불일치일 때만 합산」이라는 옛 뜻이 더는 맞지 않는다.
    for name in ("grok", "gemini", "gpt", "sol", "qwen", "kimi", "opus"):
        assert name not in v.lower(), f"모델 이름 {name!r} 이 박혔다"


# ── run_branch_select — fix gen 분리 주입 ─────────────────────────────

def _png(path: Path) -> Path:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(b"\x89PNG\r\n\x1a\n" + b"0" * 8)
    return path


class _Records:
    def __init__(self):
        self.data: Dict[str, Any] = {}

    def save(self):
        pass


def _branch_kwargs(tmp_path, make_fix_gen_fn=None):
    from app.modules.pipeline.multiroll_select import run_multiroll_select

    roll_calls: List[str] = []
    fix_calls: List[str] = []

    def roll_gen(tag, prompt, labeled_refs, out_path):
        roll_calls.append(tag)
        return _png(out_path)

    def fix_gen(tag, prompt, labeled_refs, out_path):
        fix_calls.append(tag)
        return _png(out_path)

    def judge_fn(tag, prompt, labeled_refs, cand_paths, labels):
        return _verdict({lab: (9 if lab == "A" else 3) for lab in labels},
                        "A")

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

    kwargs = dict(
        branch_tag="t1",
        branch_refs=[],
        rec_key="t1",
        out_stem=tmp_path / "out" / "s1",
        prompt="P",
        records=_Records(),
        make_gen_fn=lambda bt: roll_gen,
        judge_fn=judge_fn,
        critique_fn=critique_fn,
        roll_count=2,
        critique_enabled=True,
        judge_texts={"judge_sys": "J", "critique_sys": "C",
                     "fix_head": "H", "fix_tail": "T", "fix_label": "L"},
        extra_fingerprint={},
        run_fn=run_multiroll_select,
    )
    if make_fix_gen_fn is not None:
        kwargs["make_fix_gen_fn"] = lambda bt: fix_gen
    return kwargs, roll_calls, fix_calls


def test_run_branch_select_fix_gen_split_when_provided(tmp_path):
    """make_fix_gen_fn 제공 시 fix i2i 만 별도 gen 으로 — 롤은 기존 gen.
    G+G46 의 '수정은 Grok' 배선이 이 이음새 하나에 걸려 있다."""
    from app.modules.pipeline.still_recipe import run_branch_select

    kwargs, roll_calls, fix_calls = _branch_kwargs(
        tmp_path, make_fix_gen_fn=True)
    run_branch_select(**kwargs)
    assert all(not t.endswith("_fix") for t in roll_calls)
    assert len(roll_calls) == 2          # a/b 롤은 기존 gen
    assert [t for t in fix_calls] == ["t1_fix"]  # fix 만 분리 gen


def test_run_branch_select_fix_gen_default_reuses_roll_gen(tmp_path):
    """미제공(기본) = 기존 계약 그대로 — fix 도 롤 gen 재사용 byte-identical."""
    from app.modules.pipeline.still_recipe import run_branch_select

    kwargs, roll_calls, fix_calls = _branch_kwargs(tmp_path)
    run_branch_select(**kwargs)
    assert fix_calls == []
    assert [t for t in roll_calls if t.endswith("_fix")] == ["t1_fix"]