"""도면을 두 모델로 읽는 판 (2026-08-27, #92 6단계).

## 이 계약은 **실측으로 정했다**

도면 8장·마커 111개 실측(`artifact/20260827_floor_readback_agreement/`,
$0.7332):

    ID 집합    111/111  한쪽만 본 것 0건
    kind      111/111  (100%)
    셀 정확     41/111  (**37%**)
    셀 인접     65/111  (59%)   → exact+adjacent = 96%
    셀 어긋남    5/111  (5%)

★`fp_l02_dining_hall` 은 14개 중 **1개만** 정확히 같았다. `exact` 만
 통과시키는 계약을 넣었으면 **그 도면 하나로 파이프라인이 섰다.**

그래서 이 판은 **아무것도 버리지 않는다.** 하류가 쓰는 값은 첫 모델 것
그대로이고, 보태는 것은 `readback["agreement"]` 등급 한 칸뿐이다.
"""
from __future__ import annotations

import pytest


def _rb(*markers):
    return {"status": "ok", "fp_id": "fp_x", "grid_size": [10, 10],
            "observed_markers": [
                {"number": n, "row": r, "col": c, "kind": k}
                for n, r, c, k in markers]}


# ─────────────────────────────────────────────────────────────────────
# 등급 매기기
# ─────────────────────────────────────────────────────────────────────


def test_same_cell_is_exact():
    from app.modules.pipeline.floor_plan_vlm_provider import compare_readbacks

    got = compare_readbacks(_rb((1, 3, 4, "base_structural_unit")),
                            _rb((1, 3, 4, "base_structural_unit")))
    assert got["markers"] == {"1": "exact"}


def test_one_cell_away_is_adjacent():
    """★실측의 59% 가 여기다 — 이걸 `conflict` 로 보면 대부분이 막힌다."""
    from app.modules.pipeline.floor_plan_vlm_provider import compare_readbacks

    got = compare_readbacks(_rb((1, 3, 4, "k")), _rb((1, 4, 5, "k")))
    assert got["markers"] == {"1": "adjacent"}


def test_two_cells_away_is_conflict():
    from app.modules.pipeline.floor_plan_vlm_provider import compare_readbacks

    got = compare_readbacks(_rb((1, 3, 4, "k")), _rb((1, 3, 6, "k")))
    assert got["markers"] == {"1": "conflict"}


def test_a_different_kind_is_its_own_grade():
    """★kind 는 실측 100% 일치였다 — 어긋나면 셀보다 큰 이야기다."""
    from app.modules.pipeline.floor_plan_vlm_provider import compare_readbacks

    got = compare_readbacks(_rb((1, 3, 4, "unit")), _rb((1, 3, 4, "opening")))
    assert got["markers"] == {"1": "kind_conflict"}


def test_markers_only_one_side_saw_are_listed_not_dropped():
    """★실측에서 0건이었지만 **버리지 않는다.**

    한쪽만 본 마커를 없애면 그 마커를 쓰는 하류가 통째로 빈다.
    목록으로 남기고 판단은 나중에 한다.
    """
    from app.modules.pipeline.floor_plan_vlm_provider import compare_readbacks

    got = compare_readbacks(_rb((1, 0, 0, "k"), (2, 1, 1, "k")),
                            _rb((1, 0, 0, "k")))
    assert got["only_primary"] == [2]
    assert got["only_secondary"] == []
    assert "2" not in got["markers"]


def test_the_counts_add_up():
    from app.modules.pipeline.floor_plan_vlm_provider import compare_readbacks

    got = compare_readbacks(
        _rb((1, 0, 0, "k"), (2, 0, 0, "k"), (3, 0, 0, "k")),
        _rb((1, 0, 0, "k"), (2, 0, 1, "k"), (3, 5, 5, "k")))
    assert got["counts"] == {"exact": 1, "adjacent": 1, "conflict": 1}


# ─────────────────────────────────────────────────────────────────────
# provider — ★아무것도 버리지 않는다
# ─────────────────────────────────────────────────────────────────────


def _patch(monkeypatch, calls, second=None, boom=None):
    """★`_read_with_alias` 를 대역으로 잡는다 — **한 단 더 안쪽**이다.

    첫 판은 `litellm_vlm_provider` 를 stub 했는데, 그 함수가 **OpenAI
    전용**이라 grok 은 애초에 못 탄다는 것을 시험이 못 봤다
    (Codex BLOCK-1). 지금은 Router 를 타는 `_read_with_alias` 를 잡고,
    **그 함수가 실제로 어느 경로를 쓰는지는 아래 별도 시험이 본다.**
    """
    import app.modules.pipeline.floor_plan_vlm_provider as fp

    def _fake(alias, **kw):
        calls.append(alias)
        if len(calls) == 1:
            return _rb((1, 3, 4, "k"))
        if boom:
            raise RuntimeError(boom)
        return second if second is not None else _rb((1, 3, 4, "k"))

    monkeypatch.setattr(fp, "_read_with_alias", _fake)
    return fp


def test_the_value_downstream_uses_is_the_first_models(monkeypatch):
    """★핵심 — 하류가 쓰는 `observed_markers` 는 **첫 모델 것 그대로**다."""
    calls = []
    fp = _patch(monkeypatch, calls, second=_rb((1, 9, 9, "k")))
    out = fp.dual_vlm_provider(dossier={}, fp_image_path="x.png")

    assert out["observed_markers"] == [
        {"number": 1, "row": 3, "col": 4, "kind": "k"}]
    assert out["agreement"]["markers"] == {"1": "conflict"}
    assert calls == ["gemini-pro", "grok"]


def test_a_second_model_failure_does_not_lose_the_first(monkeypatch):
    """★fail-open — 두 번째가 죽어도 첫 판을 버리지 않는다."""
    calls = []
    fp = _patch(monkeypatch, calls, boom="503 upstream")
    out = fp.dual_vlm_provider(dossier={}, fp_image_path="x.png")

    assert out["observed_markers"], "첫 판이 사라졌다"
    assert "markers" not in out["agreement"]
    rows = out["agreement"]["attempts"]
    assert [r["ok"] for r in rows] == [True, False]
    assert "503" in rows[1]["error"]


def test_the_first_model_failure_is_not_swallowed(monkeypatch):
    """★반대편 — **첫 모델 실패는 삼키지 않는다.**

    그건 도면 읽기가 통째로 실패한 것이고, 기존 단일 경로도 그때 예외를
    올린다(재시도는 스텝 몫). 여기서 삼키면 빈 산출이 정상처럼 흐른다.
    """
    import pytest as _pt
    import app.modules.pipeline.floor_plan_vlm_provider as fp

    def _boom(alias, **kw):
        raise RuntimeError("첫 판이 죽었다")

    monkeypatch.setattr(fp, "_read_with_alias", _boom)
    with _pt.raises(Exception):
        fp.dual_vlm_provider(dossier={}, fp_image_path="x.png")


def test_every_attempt_is_recorded_with_its_physical_model(monkeypatch):
    """★Codex BLOCK-2 — **모델별 한 줄**이 남아야 호출 수를 셀 수 있다.

    ★물리 모델도 같이 남긴다. 나중에 어느 쌍의 37/96/5% 인지 역사
     기록에서 복원해야 하는데 `config_hash` 는 불투명 해시다.
    """
    calls = []
    fp = _patch(monkeypatch, calls)
    out = fp.dual_vlm_provider(dossier={}, fp_image_path="x.png")

    rows = out["agreement"]["attempts"]
    assert len(rows) == 2
    assert [r["alias"] for r in rows] == ["gemini-pro", "grok"]
    for r in rows:
        assert r["physical_model"] and r["physical_model"] != r["alias"], (
            "물리 모델이 안 남았다 — 별칭만으로는 어느 모델이었는지 모른다")


def test_the_step_counter_adds_up_both_calls(monkeypatch):
    """★끝점 — 스텝이 세는 수가 **실제 호출 수**인가.

    wrapper 진입만 세면 두 모델이 다 성공해도 기록은 1이다 —
    「도면당 두 번 산다」와 실제가 반대가 된다(Codex BLOCK-2).
    """
    import inspect
    from app.core.steps import floor_plan_geometry_readback_step as st

    src = inspect.getsource(st)
    assert 'out.get("agreement") or {}).get("attempts")' in src, (
        "카운터가 attempts 를 안 본다 — 두 번 사고 한 번으로 적는다")

    assert "_counter[0] += 1" in src, (
        "★진입 즉시 안 센다 — 예외가 나면 그 유료 왕복이 기록에서 "
        "사라진다(이 저장소가 여러 번 세운 계약)")
    assert "len(rows) - 1" in src, (
        "두 번째부터를 더해야 한다 — 첫 번은 진입에서 이미 셌다")


def _count(out=None, boom=False) -> int:
    """스텝 클로저를 **그대로** 흉내 낸 셈.

    ★진입 즉시 1, 성공하면 `attempts` 의 **두 번째부터**를 더한다.
    """
    counter = [0]

    def base(**kw):
        if boom:
            raise RuntimeError("provider 가 죽었다")
        return out

    def counted(**kw):
        counter[0] += 1
        got = base(**kw)
        if isinstance(got, dict):
            rows = (got.get("agreement") or {}).get("attempts")
            if isinstance(rows, list) and len(rows) > 1:
                counter[0] += len(rows) - 1
        return got

    try:
        counted()
    except Exception:  # noqa: BLE001 — 셈만 본다
        pass
    return counter[0]


def test_both_models_count_as_two():
    assert _count({"agreement": {"attempts": [{"a": 1}, {"b": 2}]}}) == 2


def test_a_single_provider_still_counts_as_one():
    """반대편 — 단일 경로는 그대로여야 한다(byte-identical)."""
    assert _count({"status": "ok", "observed_markers": []}) == 1


def test_a_failed_paid_roundtrip_still_counts():
    """★**예외가 나도 센다** — 그 왕복은 이미 산 것이다.

    이 계약을 깜빡하고 「성공 뒤에 세기」로 옮겼다가 기존 스텝 시험
    셋을 깼다(2026-08-27). **지적의 범위를 넘기면 원래 결함이
    되살아난다.**
    """
    assert _count(boom=True) == 1


def test_a_second_model_failure_still_counts_as_two():
    """두 번째가 죽어도 **진입은 했다** — fail-open 이라 attempts 는 2줄."""
    assert _count({"agreement": {
        "attempts": [{"ok": True}, {"ok": False}]}}) == 2


def test_the_read_goes_through_the_router_not_the_openai_only_helper():
    """★BLOCK-1 의 핵심 — **어느 경로로 나가는가.**

    `litellm_vlm_provider` 는 OpenAI 전용이다:

        · `has_openai_key()` 를 필수로 본다
        · `litellm.supports_response_schema(model)` 이 False 면 거부하는데
          **`openrouter/x-ai/grok-4.6` 은 False 다**(무료 조회로 확인)

    그래서 두 번째 모델이 한 번도 안 불렸을 것이다. 실측 도구가 쓴
    `call_structured` 경로로 옮겼다.
    """
    import ast
    import inspect
    from app.modules.pipeline import floor_plan_vlm_provider as fp

    def _called(fn) -> set:
        """그 함수가 **부르는** 이름들 — ★docstring·주석은 안 센다.

        ★첫 판은 `"litellm_vlm_provider" not in src` 로 훑었는데,
         **그 이름을 docstring 에 적어 놨다는 이유로** 빨강이 됐다.
         오늘 두 번째 substring 함정이다.
        """
        tree = ast.parse(inspect.getsource(fn).lstrip())
        out = set()
        for n in ast.walk(tree):
            if isinstance(n, ast.Call):
                f = n.func
                out.add(f.id if isinstance(f, ast.Name)
                        else f.attr if isinstance(f, ast.Attribute) else "")
        return out

    read_calls = _called(fp._read_with_alias)
    assert "call_structured" in read_calls, "Router 를 안 탄다"
    assert "litellm_vlm_provider" not in read_calls, (
        "OpenAI 전용 헬퍼를 다시 부른다 — grok 이 못 탄다")

    src = inspect.getsource(fp._read_with_alias)
    assert "enable_fallback=False" in src and "num_retries=0" in src

    assert "litellm_vlm_provider" not in _called(fp.dual_vlm_provider)


def test_it_reuses_the_production_assembly_not_a_new_one():
    """★문안을 새로 지으면 「나가는 것을 쟀다」고 말할 수 없다.

    ★**AST 로 호출을 본다.** 첫 판은 `name in src` 로 훑었는데,
     **그 이름들을 docstring 에 적어 놨기 때문에** 조립을 통째로
     지어낸 판도 초록이었다. 오늘 세 번째 substring 함정이다 —
     이름이 문서에 있다는 것은 코드가 그것을 **부른다**는 뜻이 아니다.
    """
    import ast
    import inspect
    from app.modules.pipeline import floor_plan_vlm_provider as fp

    tree = ast.parse(inspect.getsource(fp._read_with_alias).lstrip())
    called = set()
    for n in ast.walk(tree):
        if isinstance(n, ast.Call):
            f = n.func
            called.add(f.id if isinstance(f, ast.Name)
                       else f.attr if isinstance(f, ast.Attribute) else "")
    for name in ("_build_dossier_facts", "_build_messages",
                 "_build_response_format", "validate_provider_output"):
        assert name in called, f"{name} 을 **안 부른다** — 조립이 두 벌이 된다"


# ─────────────────────────────────────────────────────────────────────
# 배선 · 지문
# ─────────────────────────────────────────────────────────────────────


def test_the_step_picks_the_dual_provider_only_when_the_flag_is_on(
        monkeypatch):
    """★만들어 놓고 안 부르면 없는 것과 같다 — 스텝이 실제로 고르는가."""
    from app.core.config import settings
    from app.core.steps.floor_plan_geometry_readback_step import (
        FloorPlanGeometryReadbackStep as S)
    from app.modules.pipeline.floor_plan_vlm_provider import (
        dual_vlm_provider, gemini_vlm_provider)

    step = S.__new__(S)
    step._vlm_provider_override = None
    monkeypatch.setattr(
        settings, "floor_plan_vlm_readback_real_provider_enabled", True)

    monkeypatch.setattr(
        settings, "floor_plan_vlm_readback_dual_enabled", False)
    # ★#92: 단일 경로도 **gemini** 다 — Sol 은 어디에도 없다.
    assert step._resolve_vlm_provider() is gemini_vlm_provider

    monkeypatch.setattr(
        settings, "floor_plan_vlm_readback_dual_enabled", True)
    assert step._resolve_vlm_provider() is dual_vlm_provider


def test_the_flag_moves_the_step_fingerprint(monkeypatch):
    """★안 접으면 완주 CP 가 옛 산출을 그대로 재사용한다.

    ★**두 칸이 겹쳐 지킨다** (2026-08-27 양성 대조로 확인). 지문에는
     플래그 칸과 모델 스탬프 칸이 둘 다 있어서, **한 칸만 지워도** 다른
     칸이 해시를 움직여 이 시험이 초록이다.

     그게 결함은 아니다 — 계약(「플래그가 켜지면 지문이 움직인다」)은
     어느 한 칸만 있어도 지켜진다. 다만 **이 시험 하나로 두 칸을 다
     지킨다고 읽으면 안 된다.** 모델 이름 쪽은 아래
     `test_the_physical_models_are_in_the_stamp_too` 가 따로 본다.
     (실측: 플래그 칸만 지움=초록, 모델 스탬프만 지움=빨강,
      둘 다 지움=빨강.)
    """
    from app.core.config import settings
    from app.core.steps.floor_plan_geometry_readback_step import (
        FloorPlanGeometryReadbackStep as S)

    step = S.__new__(S)
    monkeypatch.setattr(
        settings, "floor_plan_vlm_readback_dual_enabled", False)
    off = step._config_hash()
    monkeypatch.setattr(
        settings, "floor_plan_vlm_readback_dual_enabled", True)
    assert step._config_hash() != off


def test_the_physical_models_are_in_the_stamp_too(monkeypatch):
    """★별칭만 접으면 **설정의 실제 모델이 바뀌어도** 지문이 안 움직인다."""
    from app.core.config import settings
    from app.core.steps.floor_plan_geometry_readback_step import (
        FloorPlanGeometryReadbackStep as S, _dual_models_stamp)

    step = S.__new__(S)
    monkeypatch.setattr(
        settings, "floor_plan_vlm_readback_dual_enabled", True)
    before = step._config_hash()
    monkeypatch.setattr(settings, "gemini_text_model", "다른-모델")
    assert step._config_hash() != before
    assert "다른-모델" in _dual_models_stamp()


def test_the_stamp_is_empty_when_the_flag_is_off(monkeypatch):
    """OFF 에서는 스탬프가 비어 기존 지문과 어긋나지 않아야 한다."""
    from app.core.config import settings
    from app.core.steps.floor_plan_geometry_readback_step import (
        FloorPlanGeometryReadbackStep as S)
    import json as _j
    import inspect

    src = inspect.getsource(S._config_hash)
    assert '"floor_plan_vlm_readback_dual_models"' in src
    monkeypatch.setattr(
        settings, "floor_plan_vlm_readback_dual_enabled", False)
    step = S.__new__(S)
    # 값 자체를 못 꺼내니 해시가 안 터지는 것만 본다.
    assert isinstance(step._config_hash(), str)
    del _j


def test_grok_resolves_through_openrouter_or_stays_an_alias(monkeypatch):
    """★모델이 설정에 없으면 **지어내지 않는다.**

    엉뚱한 모델을 사는 것보다 실패가 낫고, 그 실패는 fail-open 이 잡는다.
    """
    from app.core.config import settings
    from app.modules.pipeline.floor_plan_vlm_provider import _physical_of

    monkeypatch.setattr(settings, "grok_judge_model", "x-ai/grok-4.6")
    assert _physical_of("grok") == "openrouter/x-ai/grok-4.6"

    monkeypatch.setattr(settings, "grok_judge_model", "")
    assert _physical_of("grok") == "grok"


def test_the_adjacent_radius_is_recorded_as_measured():
    """★문턱을 먼저 박고 재지 않았다는 것을 코드가 말하게 한다."""
    import inspect
    from app.modules.pipeline import floor_plan_vlm_provider as fp

    assert fp.CELL_ADJACENT_RADIUS == 1
    doc = inspect.getdoc(fp.compare_readbacks) or ""
    assert "37%" in doc, "실측 값이 계약 옆에 없다 — 다음 사람이 근거를 못 찾는다"


@pytest.mark.parametrize("bad", [None, {}, {"observed_markers": None}])
def test_a_broken_readback_does_not_raise(bad):
    from app.modules.pipeline.floor_plan_vlm_provider import compare_readbacks

    got = compare_readbacks(bad, _rb((1, 0, 0, "k")))
    assert got["markers"] == {}
    assert got["only_secondary"] == [1]