"""②판정·④비평·⑧온전성의 두 모델을 **동시에** 부른다 (2026-08-29 지시).

> "2, 4, 8 병렬로"

★제약은 돈이 아니라 **생성 시간**이다. 한 콜이 평균 118초인데 두 모델을 한
 줄로 세우면 그대로 두 배다. 둘은 같은 입력을 각자 보고 각자 답할 뿐 서로를
 안 보므로 겹쳐도 된다.

여기서 재는 것 셋 — **말이 아니라 동작으로**:
  ① 진짜로 **겹치는가** (한쪽을 붙잡아 두고 다른 쪽이 먼저 끝나는지)
  ② 결과가 **완료 순서가 아니라 슬롯 순서**인가 (기록이 안 흔들려야 한다)
  ③ 한쪽이 실패해도 **나머지는 살아남는가** (기존 계약)
"""
from __future__ import annotations

import threading
import time
from typing import Any, Dict, List

import pytest


class _Gate:
    """두 호출이 실제로 겹쳤는지 재는 문.

    첫 호출은 두 번째가 들어올 때까지 **기다린다**. 순차면 영영 안 열려
    타임아웃 — 그것이 「안 겹쳤다」의 증거다.
    """

    def __init__(self, n: int = 2, timeout: float = 3.0):
        self.barrier = threading.Barrier(n, timeout=timeout)
        self.overlapped = False

    def hit(self):
        try:
            self.barrier.wait()
            self.overlapped = True
        except threading.BrokenBarrierError:
            pass


# ── ⑧ ask_both ─────────────────────────────────────────────────

def _fake_call_structured(gate: _Gate, order: List[str], delays: Dict[str, float]):
    def fn(tag, sys_p, user_p, schema, **kw):
        alias = (kw.get("project_config") or {}).get(tag, {}).get("model", tag)
        gate.hit()
        time.sleep(delays.get(alias, 0.0))
        order.append(alias)
        return {"who": alias}
    return fn


def test_ask_both_parallel_actually_overlaps(monkeypatch):
    """두 호출이 **동시에 살아 있어야** 문이 열린다."""
    import app.modules.llm.dual_vlm as dv

    gate, order = _Gate(), []
    monkeypatch.setattr(
        "app.modules.llm.llm_client.call_structured",
        _fake_call_structured(gate, order, {}))
    dv.ask_both("t", "sys", "u", {}, aliases=["gemini-pro", "grok"],
                parallel=True)
    assert gate.overlapped, "두 호출이 안 겹쳤다 — 병렬이 아니다"


def test_ask_both_sequential_by_default(monkeypatch):
    """★기본은 순차다 — 손대지 않은 호출부(참조 검증·비교)를 안 바꾼다."""
    import app.modules.llm.dual_vlm as dv

    gate, order = _Gate(timeout=0.4), []
    monkeypatch.setattr(
        "app.modules.llm.llm_client.call_structured",
        _fake_call_structured(gate, order, {}))
    dv.ask_both("t", "sys", "u", {}, aliases=["gemini-pro", "grok"])
    assert not gate.overlapped, "기본이 병렬이 됐다 — opt-in 이어야 한다"


def test_ask_both_result_order_is_slot_order_not_completion(monkeypatch):
    """★느린 쪽이 **먼저** 오는 슬롯이어도 결과 순서는 슬롯 순서다."""
    import app.modules.llm.dual_vlm as dv

    gate, order = _Gate(), []
    monkeypatch.setattr(
        "app.modules.llm.llm_client.call_structured",
        # 첫 슬롯을 일부러 느리게 → 완료 순서는 뒤집힌다
        _fake_call_structured(gate, order, {"gemini-pro": 0.25}))
    res = dv.ask_both("t", "sys", "u", {},
                      aliases=["gemini-pro", "grok"], parallel=True)
    assert order[0] == "grok", "완료 순서가 안 뒤집혔다 — 시험이 무의미하다"
    assert [c.alias for c in res.calls] == ["gemini-pro", "grok"], (
        "결과가 완료 순서로 담겼다 — 기록이 호출마다 흔들린다")


def test_ask_both_parallel_one_failure_keeps_the_other(monkeypatch):
    """한쪽이 죽어도 나머지는 산다 (기존 계약 보존)."""
    import app.modules.llm.dual_vlm as dv

    def fn(tag, sys_p, user_p, schema, **kw):
        alias = (kw.get("project_config") or {}).get(tag, {}).get("model", tag)
        if alias == "grok":
            raise RuntimeError("boom")
        return {"who": alias}

    monkeypatch.setattr(
        "app.modules.llm.llm_client.call_structured", fn)
    res = dv.ask_both("t", "sys", "u", {},
                      aliases=["gemini-pro", "grok"], parallel=True)
    by = {c.alias: c for c in res.calls}
    assert by["gemini-pro"].ok is True
    assert by["grok"].ok is False and "boom" in (by["grok"].error or "")


# ── ② _judge_gq · ④ 관찰 — 호출부가 병렬인가 (AST) ────────────────

def _has_pool_around(path: str, fn_name: str) -> bool:
    """그 함수 안에 ThreadPoolExecutor 가 있는가."""
    import ast
    import pathlib

    src = pathlib.Path(path).read_text(encoding="utf-8")
    tree = ast.parse(src)
    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) \
                and node.name == fn_name:
            return "ThreadPoolExecutor" in ast.dump(node)
    raise AssertionError(f"{fn_name} 을 못 찾았다 — 이름이 바뀌었나")


BACKEND = "app/modules/pipeline/multiroll_gemini.py"


def test_gg46_judge_calls_two_models_in_parallel():
    """② 선정 판정 — 두 심판이 겹친다."""
    assert _has_pool_around(BACKEND, "_judge_gq"), (
        "_judge_gq 가 아직 순차다 — 두 심판이 한 줄로 기다린다")


def test_gg46_critique_observers_run_in_parallel():
    """④ 비평 — 관찰 둘이 겹친다(취합은 그 뒤라 순차가 맞다)."""
    assert _has_pool_around(BACKEND, "make_gg46_critique_fn"), (
        "관찰 둘이 아직 순차다")


def test_context_vars_are_bound_in_every_parallel_site():
    """★셋 다 명시 전파해야 한다 — 하나라도 빠지면 그 축만 조용히 무너진다.

    예산·capture·trace 는 ContextVar 라 worker thread 가 **상속하지 않는다**.
    롤 생성 병렬이 이미 같은 처리를 하고 주석이 그 사고를 적고 있다.
    """
    import pathlib

    for path, fn in ((BACKEND, "_judge_gq"),
                     (BACKEND, "make_gg46_critique_fn"),
                     ("app/modules/llm/dual_vlm.py", "ask_both")):
        import ast
        tree = ast.parse(pathlib.Path(path).read_text(encoding="utf-8"))
        body = ""
        for node in ast.walk(tree):
            if isinstance(node, ast.FunctionDef) and node.name == fn:
                body = ast.dump(node)
                break
        for binder in ("bind_current_budget", "bind_current_trace",
                       "bind_current_generation_context"):
            assert binder in body, f"{fn}: {binder} 가 없다"


@pytest.mark.parametrize("needle", ["parallel=True"])
def test_cine_verify_opts_into_parallel(needle):
    """⑧ 은 호출부가 켠다 — `ask_both` 기본을 안 바꿨으므로."""
    import pathlib
    src = pathlib.Path(
        "app/modules/pipeline/cine_verify.py").read_text(encoding="utf-8")
    assert needle in src, "cine 온전성 검증이 병렬을 안 켰다"