"""두 VLM 실행부 — 계약을 **나가는 것**에서 잠근다 (2026-08-27, #92 1단계). 사용자 지시: 「무조건 gemini 3.1 pro 와 grok 최신 모델 **둘을 사용**해야해」. 이 실행부는 **합의를 안 한다.** 업무마다 실패의 뜻이 달라서다 — `ref_image_pipeline` 의 두 호출자는 같은 운반층을 쓰는데 하나는 severe 면 **돈 내고 재생성**하고 하나는 이미 산 두 장 중 승자를 고른다. ## 여기서 잠그는 계약 넷 ① 두 alias 를 **각각 한 번씩** — 재시도도 메우기도 없다 ② `enable_fallback=False` — 열리면 Sol 이 조용히 돌아온다 ③ 한쪽이 죽어도 나머지를 부른다 — 안 그러면 「못 물어봤다」와 「둘 다 나쁘다」가 구분이 안 된다 ④ 미보고 usage 를 **0 으로 안 채운다** — 0 은 「안 썼다」로 읽힌다 ★시험이 대역을 두는 자리는 **`call_structured` 경계**다. 실행부 안쪽을 흉내 내면 「무엇을 넘겼나」를 못 잰다. """ from __future__ import annotations import ast import pathlib import pytest from app.modules.llm.dual_vlm import ( DUAL_VLM_ALIASES, DualResult, OneCall, ask_both, ) class _Spy: """`call_structured` 대역 — 넘어온 인자를 통째로 기록한다.""" def __init__(self, behave=None): self.calls = [] self._behave = behave or (lambda alias: {"v": alias}) def __call__(self, step, sys_p, user_p, schema, **kw): self.calls.append({"step": step, "kw": kw}) alias = (kw.get("project_config") or {}).get(step, {}).get("model") out = self._behave(alias) if isinstance(out, Exception): raise out sink = kw.get("usage_sink") if sink is not None and isinstance(out, dict): sink.update(out.pop("_usage", {}) or {}) return out def _run(monkeypatch, spy, **kw): import app.modules.llm.llm_client as lc monkeypatch.setattr(lc, "call_structured", spy) base = dict(step="probe", system_prompt="s", user_prompt="u", response_schema={"type": "object"}) base.update(kw) return ask_both(**base) # ── ① 각각 한 번씩 ─────────────────────────────────────────────────── def test_each_alias_is_asked_exactly_once(monkeypatch): spy = _Spy() got = _run(monkeypatch, spy) asked = [(c["kw"]["project_config"] or {})[c["step"]]["model"] for c in spy.calls] assert asked == list(DUAL_VLM_ALIASES), f"부른 순서/횟수가 다르다: {asked}" assert len(spy.calls) == 2, "한 모델을 두 번 불렀다" assert got.complete is True def test_a_failure_is_not_retried(monkeypatch): """★실패한 모델을 다시 부르면 **독립 두 의견**이라는 전제가 깨진다.""" spy = _Spy(lambda alias: RuntimeError("boom") if alias == "grok" else {"v": alias}) got = _run(monkeypatch, spy) assert len(spy.calls) == 2, "재시도가 일어났다" assert got.complete is False def test_one_failure_does_not_stop_the_other(monkeypatch): """★한쪽이 죽었다고 나머지를 안 물으면 「못 물어봤다」와 「둘 다 나쁘다」가 구분이 안 된다.""" spy = _Spy(lambda alias: RuntimeError("boom") if alias == "gemini-pro" else {"v": alias}) got = _run(monkeypatch, spy) assert len(spy.calls) == 2 assert [c.ok for c in got.calls] == [False, True] assert got.payloads == [{"v": "grok"}], "성공한 쪽 답이 사라졌다" # ── ② fallback 봉인 ────────────────────────────────────────────────── def test_fallback_is_sealed_on_every_call(monkeypatch): """★★열리면 Gemini/Grok 실패 뒤 **GPT Sol 이 조용히 돌아온다.** 사용자 결정을 어기고 기록에도 Sol 이 안 남는다. """ spy = _Spy() _run(monkeypatch, spy) for c in spy.calls: assert c["kw"].get("enable_fallback") is False, ( f"{c['step']} 에서 fallback 이 열려 있다") def test_router_retries_are_sealed_too(monkeypatch): """★★**`enable_fallback=False` 는 Router 재시도를 안 막는다** (2026-08-27 Codex BLOCK). 그 인자는 `call_structured` 의 Tier 2/3 만 닫는다. Router 자체는 `num_retries=settings.llm_max_retries`(기본 **3**)로 지어지므로 한 alias 가 **최대 4번 전송**될 수 있고, 재시도는 다 과금되는데 `usage_sink` 는 **마지막 응답만** 본다 — 앞선 지출이 기록에서 사라진다. ★이 시험은 `call_structured` 인자가 아니라 **Router 가 받는 kwargs** 를 본다. 인자만 재면 「모델마다 한 번」이 실제로는 네 번인 것을 못 잡는다. """ import app.modules.llm.llm_client as lc seen = [] def _fake_completion(binding, model, kwargs): seen.append((model, kwargs.get("num_retries"))) return type("R", (), { "choices": [type("C", (), {"message": type("M", (), { "content": '{"v": 1}'})()})()], "usage": None, "model": f"phys/{model}"})() monkeypatch.setattr(lc, "_completion", _fake_completion) monkeypatch.setattr(lc, "_get_router_binding", lambda: object()) from app.core.config import settings monkeypatch.setattr(settings, "openrouter_api_key", "sk", raising=False) monkeypatch.setattr(settings, "grok_judge_model", "x", raising=False) ask_both("probe", "s", "u", {"type": "object"}) assert len(seen) == len(DUAL_VLM_ALIASES) for model, n in seen: assert n == 0, ( f"{model}: Router 재시도가 열려 있다(num_retries={n!r}) — " "한 판정이 최대 4번 과금될 수 있다") def test_the_source_says_it_too(): """★인자를 잰 시험은 **대역이 그 인자를 받는 한** 초록이다. 호출을 통째로 다른 함수로 바꾸면 대역이 안 불려 시험이 조용히 무의미해질 수 있다 — 소스에도 그 인자가 있는지 AST 로 본다. """ src = (pathlib.Path(__file__).resolve().parents[2] / "app" / "modules" / "llm" / "dual_vlm.py").read_text("utf-8") calls = [n for n in ast.walk(ast.parse(src)) if isinstance(n, ast.Call) and getattr(n.func, "id", "") == "call_structured"] assert calls, "실행부가 call_structured 를 안 부른다" for c in calls: kws = {k.arg: k.value for k in c.keywords} assert isinstance(kws.get("enable_fallback"), ast.Constant) assert kws["enable_fallback"].value is False # ★Router 재시도 봉인도 소스에 있어야 한다 — 대역이 그 인자를 # 안 받게 바뀌면 위 시험이 조용히 무의미해진다. assert isinstance(kws.get("num_retries"), ast.Constant) assert kws["num_retries"].value == 0 # ── ③ 판단은 안 한다 ───────────────────────────────────────────────── def test_the_executor_does_not_decide(): """★합의·기각·승자 선정이 여기 들어오면 업무마다 다른 실패의 뜻이 섞인다. 자료만 있는지 소스로 본다.""" src = (pathlib.Path(__file__).resolve().parents[2] / "app" / "modules" / "llm" / "dual_vlm.py").read_text("utf-8") body = "\n".join(ln for ln in src.splitlines() if not ln.strip().startswith("#")) for banned in ("winner", "reject", "consensus", "majority", "vote"): assert banned not in body.lower(), ( f"실행부가 판단을 한다: {banned}") def test_complete_is_data_not_a_verdict(): r = DualResult(calls=[OneCall("a", True, {"x": 1}), OneCall("b", False, None, "boom")]) assert r.complete is False assert r.payloads == [{"x": 1}], "성공한 답은 그대로 준다" assert r.by_alias("b").error == "boom" # ── ④ 기록 ─────────────────────────────────────────────────────────── def test_provenance_keeps_every_model_raw(monkeypatch): spy = _Spy(lambda alias: {"v": alias, "_usage": {"physical_model": f"phys/{alias}", "completion_tokens": 11}}) got = _run(monkeypatch, spy) prov = got.provenance() assert prov["contract"] == "dual_vlm_v1" assert prov["aliases"] == list(DUAL_VLM_ALIASES) assert [c["raw"] for c in prov["calls"]] == [ {"v": a} for a in DUAL_VLM_ALIASES], "모델별 raw 가 사라졌다" assert [c["physical_model"] for c in prov["calls"]] == [ f"phys/{a}" for a in DUAL_VLM_ALIASES] def test_unreported_usage_is_absent_not_zero(monkeypatch): """★★**0 은 「안 썼다」로 읽힌다.** 내가 그 함정을 이미 밟았다 — `cost` 가 딕셔너리인데 스칼라로 읽어 0 이 나왔고, 그 0 을 「기록이 없다」로 사용자에게 보고했다. """ spy = _Spy(lambda alias: {"v": alias}) # usage 를 아예 안 준다 got = _run(monkeypatch, spy) for c in got.provenance()["calls"]: assert "completion_tokens" not in c["usage"], ( "안 온 값을 0 으로 채웠다") assert c["physical_model"] == "", "alias 로 물리 모델을 지어냈다" def test_a_failed_call_still_records_its_shape(monkeypatch): spy = _Spy(lambda alias: RuntimeError("nope") if alias == "grok" else {"v": alias}) prov = _run(monkeypatch, spy).provenance() bad = [c for c in prov["calls"] if not c["ok"]][0] assert bad["alias"] == "grok" assert "nope" in bad["error"] assert bad["raw"] is None assert prov["complete"] is False # ── 짝 ─────────────────────────────────────────────────────────────── def test_the_pair_is_what_the_user_asked_for(): assert DUAL_VLM_ALIASES == ("gemini-pro", "grok") assert "gpt" not in DUAL_VLM_ALIASES def test_a_caller_can_override_the_pair(monkeypatch): """실측·비교를 위해 짝을 바꿀 수 있어야 한다 — 기본은 안 바뀐다.""" spy = _Spy() _run(monkeypatch, spy, aliases=("gemini-pro", "gemini-flash")) asked = [(c["kw"]["project_config"] or {})[c["step"]]["model"] for c in spy.calls] assert asked == ["gemini-pro", "gemini-flash"] assert DUAL_VLM_ALIASES == ("gemini-pro", "grok"), "기본이 바뀌었다" def test_the_step_tag_splits_by_alias(monkeypatch): """기록에서 두 호출이 갈려야 나중에 모델별로 읽는다.""" spy = _Spy() _run(monkeypatch, spy) tags = [c["step"] for c in spy.calls] assert tags == [f"probe__{a}" for a in DUAL_VLM_ALIASES] assert len(set(tags)) == 2, "두 호출이 같은 태그로 섞였다" def test_empty_alias_list_is_not_complete(): """★부른 것이 없는데 `complete=True` 면 호출자가 빈 합의로 결정한다.""" assert DualResult(calls=[]).complete is False if __name__ == "__main__": raise SystemExit(pytest.main([__file__]))