"""GROUNDING-V2 §8.5 — **시간 통제 세 겹**을 끝점에서 잰다.

★셋은 서로 다른 것을 막는다. 하나가 다른 하나를 대신 못 한다.
    ① 한 호출의 벽시계   `call_with_deadline`
    ② 주행 전체 전송 수  `ResearchCallBudget`
    ③ 주행 전체 벽시계   `install_stop_check` 에 마감을 보는 check

★판정은 셋 다 같다 — `time_capped` · `unresolved` · retryable.
 구분은 `limit_kind` 가 진다. `GAP_REASONS` 는 안 늘린다 (Codex).
"""
import json
from types import SimpleNamespace

import pytest

from app.core import research_call_budget as rb
from app.core.image_call_budget import (install_stop_check,
                                        uninstall_stop_check)
from app.modules.pipeline import grounding_claims_search as g
from app.modules.pipeline.grounding_claims import (
    GAP_TIME_CAPPED, LIMIT_ADMISSION, LIMIT_PER_CALL_DEADLINE,
    LIMIT_RUN_DEADLINE, LIMIT_TRANSMISSION_BUDGET, build_gap)


@pytest.fixture(autouse=True)
def _clean():
    """★스레드 지역이라 새어 나가면 **다음 시험이 남의 예산을 본다**."""
    rb.uninstall_budget()
    uninstall_stop_check()
    yield
    rb.uninstall_budget()
    uninstall_stop_check()


class TestTheStopCheckRunsBeforeTheBudget:
    """★★★Codex ① — 이미 마감된 주행은 **예산도 슬롯도 안 먹고** provider 0회.

    정지 확인을 예산 **뒤에** 두면, 멈춰야 할 주행이 예산을 한 칸 깎고 나서야
    선다. 그 칸은 영영 안 돌아온다.
    """

    def test_a_stopped_run_spends_no_budget(self):
        budget = rb.ResearchCallBudget(cap=5)
        rb.install_budget(budget)
        install_stop_check(_stop("주행 마감"))
        with rb.research_calls_armed():
            with pytest.raises(RuntimeError, match="주행 마감"):
                rb.reserve_current_research_call(source="x")
        assert budget.snapshot()["used"] == 0, "멈췄는데 예산을 먹었다"
        assert budget.snapshot()["denied"] == 0

    def test_a_stopped_run_calls_the_provider_zero_times(self, monkeypatch):
        """★끝점 — **진짜 `_invoke`** 를 태운다. 흉내 내면 이 순서를 못 잰다."""
        from app.core import openai_keys as ok

        hits = {"n": 0}

        class _Target:
            def __call__(self, **kw):
                hits["n"] += 1
                return "ok"

        budget = rb.ResearchCallBudget(cap=5)
        rb.install_budget(budget)
        install_stop_check(_stop("주행 마감"))
        monkeypatch.setattr(ok, "slot_count", lambda: 2)
        client = _FakeFailover(_Target())
        with rb.research_calls_armed():
            with pytest.raises(RuntimeError, match="주행 마감"):
                client.invoke()
        assert hits["n"] == 0, "멈췄는데 provider 를 불렀다"
        assert budget.snapshot()["used"] == 0, "멈췄는데 예산을 먹었다"


class TestTheBudgetOnlyCountsResearchCalls:
    """★★★Codex ② — 예산이 **조사 호출에만** 걸린다.

    `reserve` 를 `FailoverOpenAIClient._invoke` 라는 공용 경계에 뒀으므로,
    범위를 안 좁히면 같은 스레드의 **다른** OpenAI 호출이 조사 예산을 먹는다.
    """

    def test_an_unrelated_call_does_not_spend_it(self):
        budget = rb.ResearchCallBudget(cap=1)
        rb.install_budget(budget)
        # ★팔을 안 들었다 — 이미지·요약 등 남의 호출이 지나가는 판이다
        for _ in range(5):
            rb.reserve_current_research_call(source="image.generate")
        assert budget.snapshot()["used"] == 0

    def test_the_research_call_does_spend_it(self):
        """★positive control — 안 세면 상한이 아무것도 안 막는다."""
        budget = rb.ResearchCallBudget(cap=3)
        rb.install_budget(budget)
        with rb.research_calls_armed():
            rb.reserve_current_research_call(source="responses.create")
        assert budget.snapshot()["used"] == 1

    def test_the_arm_is_dropped_after_the_block(self):
        assert rb.is_armed() is False
        with rb.research_calls_armed():
            assert rb.is_armed() is True
        assert rb.is_armed() is False


class TestEachSlotAttemptReservesOne:
    """★★★Codex ③ — **물리 전송마다** 한 칸이다.

    슬롯 failover 는 `_invoke` 안에서 돌아 논리 하나가 물리 둘이 된다.
    호출 단위로 세면 그 두 번째가 **상한 밖에서** 나간다.
    """

    def test_two_slots_take_two_units(self, monkeypatch):
        from app.core import openai_keys as ok

        budget = rb.ResearchCallBudget(cap=5)
        rb.install_budget(budget)
        tries = {"n": 0}

        class _Target:
            def __call__(self, **kw):
                tries["n"] += 1
                if tries["n"] == 1:
                    raise RuntimeError("첫 슬롯이 튕겼다")
                return "ok"

        client = _FakeFailover(_Target())
        monkeypatch.setattr(ok, "slot_count", lambda: 2)
        monkeypatch.setattr(ok, "failover_on",
                            lambda exc, where, attempted_slot: True)
        with rb.research_calls_armed():
            assert client.invoke() == "ok"
        assert tries["n"] == 2
        assert budget.snapshot()["used"] == 2, "물리 두 번인데 한 칸만 먹었다"

    def test_the_budget_stops_the_second_slot(self, monkeypatch):
        from app.core import openai_keys as ok

        budget = rb.ResearchCallBudget(cap=1)
        rb.install_budget(budget)
        tries = {"n": 0}

        class _Target:
            def __call__(self, **kw):
                tries["n"] += 1
                raise RuntimeError("튕김")

        client = _FakeFailover(_Target())
        monkeypatch.setattr(ok, "slot_count", lambda: 2)
        monkeypatch.setattr(ok, "failover_on",
                            lambda exc, where, attempted_slot: True)
        with rb.research_calls_armed():
            with pytest.raises(rb.ResearchCallBudgetExceeded):
                client.invoke()
        assert tries["n"] == 1, "상한을 넘고도 두 번째 슬롯이 나갔다"


class TestTheLimitKindIsDerivedFromTheType:
    """★★타입으로 가른다 — 글자로 가르면 provider 문구가 바뀔 때 조용히 틀린다."""

    def test_a_per_call_deadline(self):
        from app.modules.pipeline.llm_deadline import HardDeadlineExceeded

        assert g._limit_kind_of(HardDeadlineExceeded("늦다")) == \
            LIMIT_PER_CALL_DEADLINE

    def test_a_transmission_budget(self):
        exc = rb.ResearchCallBudgetExceeded(cap=1, used=1, source="x")
        assert g._limit_kind_of(exc) == LIMIT_TRANSMISSION_BUDGET

    def test_a_run_deadline_is_marked(self):
        exc = RuntimeError("주행 마감")
        exc.is_run_deadline = True
        assert g._limit_kind_of(exc) == LIMIT_RUN_DEADLINE

    def test_a_plain_provider_error_is_not_a_limit(self):
        """★전송이 깨진 것은 **제한이 아니다** — 고칠 곳이 다르다."""
        assert g._limit_kind_of(RuntimeError("ConnectionReset")) == ""


class TestTheLimitKindSurvivesTheRoundTrip:
    """★★★저장 → 재구성 → 재개까지 **끝까지 실려 간다** (Codex).

    칸을 더하고 **그 칸을 복사하는 줄**을 안 고치면, 재구성 자리에서 계약
    위반이 되어 gap 이 통째로 버려진다 — 그러면 「무슨 제한에 걸렸나」가
    기록에서 사라진다. 실제로 그렇게 됐었다.
    """

    @pytest.mark.parametrize("kind", [
        LIMIT_ADMISSION, LIMIT_PER_CALL_DEADLINE,
        LIMIT_TRANSMISSION_BUDGET, LIMIT_RUN_DEADLINE])
    def test_it_round_trips_through_the_revision_row(self, kind):
        from app.modules.pipeline.grounding_claims import (STATUS_RETRYABLE,
                                                           build_revision_row)

        from tests.grounding.test_grounding_research_record import SID

        sid = SID
        gap = build_gap(sid, reason=GAP_TIME_CAPPED, note="상한",
                        required_discriminator=True, limit_kind=kind)
        row = build_revision_row(claims=[], gaps=[gap], **_base(sid))
        stored = json.loads(json.dumps(row))          # ★저장 왕복
        # ★저장 칸 이름은 `gaps_json` 이다 — 손으로 짐작하지 말고 실제 행에서
        #  가져온다(조립부 이름을 그대로 쓴다).
        kinds = [x.get("limit_kind")
                 for x in json.loads(stored["gaps_json"])]
        assert kind in kinds, f"저장을 지나며 {kind} 가 사라졌다"
        assert stored["status"] == STATUS_RETRYABLE, "다시 살 수 있어야 한다"

    def test_a_gap_without_a_kind_is_refused(self):
        with pytest.raises(ValueError, match="limit_kind"):
            build_gap("rs_x", reason=GAP_TIME_CAPPED)

    def test_a_kind_on_another_reason_is_refused(self):
        from app.modules.pipeline.grounding_claims import GAP_NOT_FOUND

        with pytest.raises(ValueError, match="time_capped 에만"):
            build_gap("rs_x", reason=GAP_NOT_FOUND,
                      limit_kind=LIMIT_RUN_DEADLINE)


# ── 대역 ─────────────────────────────────────────────────────────────────
def _stop(msg):
    def _check():
        exc = RuntimeError(msg)
        exc.is_run_deadline = True
        raise exc
    return _check


class _FakeFailover:
    """★`_invoke` 를 **진짜 그 코드로** 태운다 — 흉내 내지 않는다."""

    def __init__(self, target):
        self._target = target

    def _raw(self):
        return ("slot0", SimpleNamespace(responses=SimpleNamespace(
            create=self._target)))

    def invoke(self):
        from app.core.openai_keys import FailoverOpenAIClient

        return FailoverOpenAIClient._invoke(
            self, ("responses", "create"), (), {"model": "m"})


def _base(sid):
    """★행 대역은 **한 곳**에서 온다 — 두 벌로 적으면 한쪽만 고쳐진다."""
    from tests.grounding.test_grounding_research_record import BASE

    return {**BASE, "research_subject_id": sid}


class TestTheRunDeadlineStopsBeforeTheNetwork:
    """★★★③ 주행 전체 벽시계. **`reserve` 자리에서** 막힌다.

    「새 대상을 안 시작한다」는 loop 조건으로만 두면, 팬아웃 안에서 **이미
    제출된 것들**이 계속 나간다 — 그건 마감이 아니다.
    """

    def test_it_blocks_at_the_reserve_point(self):
        clock = {"t": 0.0}
        with rb.research_run_scope(cap=10, deadline_seconds=5.0,
                                   now=lambda: clock["t"]) as budget:
            with rb.research_calls_armed():
                rb.reserve_current_research_call(source="a")
                clock["t"] = 5.1                      # ★마감이 지났다
                with pytest.raises(rb.RunDeadlineExceeded):
                    rb.reserve_current_research_call(source="b")
        assert budget.snapshot()["used"] == 1, "마감 뒤에 한 칸 더 먹었다"

    def test_the_deadline_error_is_a_run_deadline_kind(self):
        exc = rb.RunDeadlineExceeded("지났다")
        assert g._limit_kind_of(exc) == LIMIT_RUN_DEADLINE

    def test_an_earlier_stop_check_still_wins(self):
        """★사용자 취소가 마감보다 **먼저** 들린다 — 덮어쓰지 않는다."""
        install_stop_check(_stop("사용자가 멈췄다"))
        with rb.research_run_scope(cap=10, deadline_seconds=999.0):
            with rb.research_calls_armed():
                with pytest.raises(RuntimeError, match="사용자가 멈췄다"):
                    rb.reserve_current_research_call(source="a")

    def test_the_scope_restores_what_was_there(self):
        """★주행이 끝나면 **남의 호출이 이 상한을 안 본다**."""
        assert rb.get_current_budget() is None
        with rb.research_run_scope(cap=2, deadline_seconds=1.0):
            assert rb.get_current_budget() is not None
        assert rb.get_current_budget() is None
        with rb.research_calls_armed():
            rb.reserve_current_research_call(source="after")   # 안 터진다

    def test_the_budget_still_caps_inside_the_scope(self):
        """★positive control — 마감만 막고 상한을 안 막으면 반쪽이다."""
        with rb.research_run_scope(cap=1, deadline_seconds=999.0):
            with rb.research_calls_armed():
                rb.reserve_current_research_call(source="a")
                with pytest.raises(rb.ResearchCallBudgetExceeded):
                    rb.reserve_current_research_call(source="b")


class TestTheDeadlineWrapperDoesNotConsumeALateReturn:
    """★마감에 버려진 worker 의 **반환값을 안 쓴다**.

    `call_with_deadline` 은 **취소가 아니라 포기**다. 그 스레드는 계속 돌다가
    나중에 값을 들고 온다 — 그 값이 행에 끼어들면 「마감에 걸렸다」와 「결과가
    있다」가 같은 행에 함께 남는다.

    ★그렇다고 그 시도를 **기록에서 지우지도 않는다** — 비용·trace 는 남는다.

    ★★★**여기까지가 이 시험이 증명하는 전부다** (Codex). 이건 wrapper 의
    **generic 성질**이고, 「버려진 worker 가 checkpoint·DB 를 안 건드린다」는
    **프로덕션 불변식이 아니다** — 그건 실제 step 호출부가 있어야 잰다.
    지금 `search_claims` 를 부르는 프로덕션 코드는 **0건**이다.
    이 시험을 §4b ④ 의 완료로 세지 않는다.
    """

    def _row(self, monkeypatch, *, late_holder):
        import time as _t

        from app.modules.llm import image_tracer

        calls = []
        monkeypatch.setattr(image_tracer, "record_provider_call",
                            lambda **kw: (calls.append(kw), "trace_1")[1])
        monkeypatch.setattr(image_tracer, "ambient_call_meta", lambda: {})
        monkeypatch.setattr(image_tracer, "resolve_step_name",
                            lambda name, meta: name)
        monkeypatch.setattr(g, "PER_CALL_DEADLINE_SECONDS", 0.15)

        class _Slow:
            @property
            def responses(self):
                class _R:
                    @staticmethod
                    def create(**kw):
                        _t.sleep(0.5)          # ★마감보다 늦다
                        late_holder.append("늦게 도착")
                        return SimpleNamespace(id="resp_late", output=[])
                return _R

        out = g.search_claims(
            _Slow(), [{"research_subject_id": "rs_a", "surface_form": "가",
                       "owner_type": "prop", "source_quote": "문장"}],
            model="m", era="1983년", region="대한민국", batch_size=1)
        return out["batches"][0], calls

    def test_the_row_says_capped_and_carries_no_result(self, monkeypatch):
        late = []
        row, _calls = self._row(monkeypatch, late_holder=late)
        assert row["limit_kind"] == LIMIT_PER_CALL_DEADLINE
        assert row["error"], "마감인데 사유가 안 남았다"
        assert not row.get("parsed"), "버린 worker 의 값이 행에 들어왔다"

    def test_the_late_arrival_does_not_rewrite_it(self, monkeypatch):
        import time as _t

        late = []
        row, _calls = self._row(monkeypatch, late_holder=late)
        before = dict(row)
        _t.sleep(0.6)                       # ★버린 worker 가 끝날 시간을 준다
        assert late, "대역이 안 돌았다 — 이 시험이 아무것도 안 재고 있다"
        assert row == before, "늦게 온 값이 행을 고쳤다"

    def test_the_attempt_is_still_recorded(self, monkeypatch):
        """★기록에서 지우지는 않는다 — 비용·trace 는 남는다."""
        late = []
        _row, calls = self._row(monkeypatch, late_holder=late)
        assert len(calls) == 1
        assert calls[0]["status"] == "error"
        assert LIMIT_PER_CALL_DEADLINE in calls[0]["error"]


class TestAUserCancelIsNotSwallowed:
    """★★★Codex 실측 — 취소가 **일반 전송 실패 행**으로 바뀌고 있었다.

    그러면 ①취소가 「provider 장애」로 잘못 기록되고 ②호출자에게 안 전해져
    **다음 batch 가 계속 돈다**. 사람이 멈춘 것은 우리 상한과 **다르다**.
    """

    def _run(self, monkeypatch):
        from app.core.errors import AppError
        from app.modules.llm import image_tracer

        calls = []
        monkeypatch.setattr(image_tracer, "record_provider_call",
                            lambda **kw: (calls.append(kw), "t")[1])
        monkeypatch.setattr(image_tracer, "ambient_call_meta", lambda: {})
        monkeypatch.setattr(image_tracer, "resolve_step_name",
                            lambda n, m: n)
        hits = {"n": 0}

        def _cancel():
            raise AppError(code="step.cancelled", message="사용자가 멈췄다")

        install_stop_check(_cancel)
        rows = [{"research_subject_id": f"rs_{i}", "surface_form": "가",
                 "owner_type": "prop", "source_quote": "문장"}
                for i in range(3)]
        return _real_client(hits), hits, calls, rows

    def test_it_propagates_and_buys_nothing(self, monkeypatch):
        from app.core.errors import AppError

        client, hits, _calls, rows = self._run(monkeypatch)
        with rb.research_run_scope(cap=10) as budget:
            with pytest.raises(AppError) as exc:
                g.search_claims(client, rows, model="m", era="1983년",
                                region="대한민국", batch_size=1)
        assert exc.value.code == "step.cancelled", "취소 코드가 사라졌다"
        assert hits["n"] == 0, "취소인데 provider 를 불렀다"
        assert budget.snapshot()["used"] == 0, "취소인데 예산을 먹었다"

    def test_the_later_subjects_do_not_keep_running(self, monkeypatch):
        """★전파가 안 되면 **다음 대상이 계속 돈다** — 그게 이 결함의 값이다."""
        from app.core.errors import AppError

        client, hits, calls, rows = self._run(monkeypatch)
        with rb.research_run_scope(cap=10):
            with pytest.raises(AppError):
                g.search_claims(client, rows, model="m", era="1983년",
                                region="대한민국", batch_size=1)
        assert hits["n"] == 0
        # ★취소를 「전송 실패」로 적지도 않는다
        assert not any(c.get("status") == "error" for c in calls), \
            "취소가 provider 장애로 기록됐다"

    def test_our_own_limits_are_still_folded(self, monkeypatch):
        """★positive control — 우리 상한까지 전파시키면 주행이 통째로 죽는다."""
        from app.modules.llm import image_tracer

        monkeypatch.setattr(image_tracer, "record_provider_call",
                            lambda **kw: "t")
        monkeypatch.setattr(image_tracer, "ambient_call_meta", lambda: {})
        monkeypatch.setattr(image_tracer, "resolve_step_name",
                            lambda n, m: n)

        rows = [{"research_subject_id": "rs_0", "surface_form": "가",
                 "owner_type": "prop", "source_quote": "문장"}]
        with rb.research_run_scope(cap=0):          # ★상한 0 — 바로 걸린다
            out = g.search_claims(_real_client({"n": 0}), rows, model="m",
                                  era="1983년", region="대한민국",
                                  batch_size=1)
        assert out["batches"][0]["limit_kind"] == LIMIT_TRANSMISSION_BUDGET


def _real_client(hits):
    """★**진짜 `FailoverOpenAIClient`** 를 쓴다 — 그래야 `_invoke` 의 reserve 를
    지난다. 평범한 대역을 쓰면 그 자리가 통째로 안 돌아 아무것도 안 재게 된다.
    """
    from app.core.openai_keys import FailoverOpenAIClient

    def _create(**kw):
        hits["n"] += 1
        return SimpleNamespace(id="r", output=[])

    client = FailoverOpenAIClient.__new__(FailoverOpenAIClient)
    client._raw = lambda: ("slot0", SimpleNamespace(
        responses=SimpleNamespace(create=_create)))
    return client