"""문 **아래 두 겹**을 정말 잠그나. ★유료 0 — 바깥으로 안 나간다.

Codex (2026-09-02): 「SDK 재시도를 0으로 실제 client에서 확인할 수 없으면
raw 70을 주장할 수 없으므로 live하지 말고 다시 올리십시오.」

    문 위(세어진다)   tier × 키 슬롯      ← tier 를 1 로 닫는다
    문 아래(안 세짐)  router × SDK 재시도  ← 둘 다 0 으로 박는다
"""
from __future__ import annotations

import pytest

from tools.grounding_audit import canary_request_lock as rl


class TestTheSdkLayerCanActuallyBePinned:
    """★★계약이 「못 바꾼다」고 적혀 있던 자리 — **실측으로 뒤집혔다**."""

    def test_the_default_client_retries_twice(self):
        """★음성 대조 — 아무것도 안 주면 **2** 로 지어진다."""
        import openai

        seen = []
        real = openai.OpenAI.__init__

        def _spy(self, *a, **k):
            seen.append(k.get("max_retries", "<없음>"))
            return real(self, *a, **k)

        openai.OpenAI.__init__ = _spy
        try:
            openai.OpenAI(api_key="sk-not-real")
        finally:
            openai.OpenAI.__init__ = real
        assert seen == ["<없음>"], seen

    def test_zero_reaches_the_client(self):
        """★양성 — `max_retries=0` 이 **그 객체까지** 간다."""
        import openai

        c = openai.OpenAI(api_key="sk-not-real", max_retries=0)
        assert c.max_retries == 0


class TestTheLockClosesBothLayers:

    def test_it_injects_zero_before_sending(self, monkeypatch):
        """★`_completion` 에 들어가는 kwargs 에 **0 이 박힌다**."""
        from app.modules.llm import llm_client as lc

        got = {}
        monkeypatch.setattr(rl, "router_already_built", lambda: False)
        monkeypatch.setattr(lc, "_completion",
                            lambda b, m, kw: got.update(kw) or "resp")
        with rl.canary_request_lock() as locked:
            lc._completion(None, "m", {"messages": []})
        assert got["max_retries"] == 0
        assert locked["sdk_max_retries"] == 0

    def test_it_closes_the_tier_fallback(self, monkeypatch):
        from app.modules.llm import llm_client as lc

        got = {}
        monkeypatch.setattr(rl, "router_already_built", lambda: False)
        monkeypatch.setattr(lc, "call_structured",
                            lambda *a, **k: got.update(k) or {})
        with rl.canary_request_lock():
            lc.call_structured(step="s")
        assert got["enable_fallback"] is False

    def test_everything_is_put_back(self, monkeypatch):
        """★★나올 때 **production 을 되돌린다** — 잠금이 새면 안 된다."""
        import openai

        from app.modules.llm import llm_client as lc

        monkeypatch.setattr(rl, "router_already_built", lambda: False)
        before = (lc._completion, lc.call_structured, openai.OpenAI.__init__)
        with rl.canary_request_lock():
            assert lc._completion is not before[0]
        assert (lc._completion, lc.call_structured,
                openai.OpenAI.__init__) == before


class TestTheObservationIsTheEvidence:
    """★「0 으로 보냈다」와 「0 으로 지어졌다」는 다르다."""

    def setup_method(self):
        rl.OBSERVED_CLIENT_RETRIES.clear()

    def test_no_observation_is_not_a_failure(self):
        """★Gemini 경로는 openai client 를 안 짓는다 — 빈 관측은 정상."""
        assert rl.assert_client_retries_observed()["locked"] is True

    def test_a_nonzero_client_stops(self):
        rl.OBSERVED_CLIENT_RETRIES.extend([0, 2])
        with pytest.raises(rl.RequestContractRefused):
            rl.assert_client_retries_observed()

    def test_all_zero_passes(self):
        rl.OBSERVED_CLIENT_RETRIES.extend([0, 0, 0])
        got = rl.assert_client_retries_observed()
        assert got["locked"] is True and got["bad"] == []

    def test_the_spy_records_what_the_client_got(self, monkeypatch):
        """★★끝점 — 잠금 **안에서** 지어진 client 가 0 으로 적힌다."""
        import openai

        monkeypatch.setattr(rl, "router_already_built", lambda: False)
        with rl.canary_request_lock():
            openai.OpenAI(api_key="sk-not-real", max_retries=0)
        assert rl.OBSERVED_CLIENT_RETRIES == [0]
        rl.assert_client_retries_observed()


class TestAnInnerScopeCannotHideCalls:
    """★★★스텝이 **제 예산 scope** 를 열면 canary 계수기가 가려진다.

    실측 (2026-09-02 유료 재개): `grounding_chunk` 이
    `research_run_scope(cap=24)` 를 열었고, 그것이 `install_budget` 으로 canary
    예산을 **갈아 끼웠다**. 그 스텝의 유료 호출 2건이 장부에 **counted 0** 으로
    적혔다 — 즉 「run 전체 정지선 93」이 그 스텝을 못 봤다(제 상한 24 는 따로).
    """

    def test_the_step_scope_really_replaces_the_budget(self):
        """★근거 — 계약이 아니라 **그 함수**가 갈아 끼운다."""
        import inspect

        from app.core import research_call_budget as rcb

        src = inspect.getsource(rcb.research_run_scope)
        assert "install_budget(budget)" in src
        assert "prev_budget = get_current_budget()" in src

    def test_the_transport_counter_still_sees_them(self, monkeypatch):
        """★★안쪽 scope 안에서 난 전송도 **운반층 계수기**에는 보인다."""
        from app.core import research_call_budget as rcb

        monkeypatch.setattr(rl, "router_already_built", lambda: False)
        with rl.canary_request_lock(transport_cap=5) as locked:
            t = locked["transport"]
            with rcb.research_run_scope(cap=99):     # ★스텝이 제 것을 연다
                rcb.reserve_current_research_call(source="안쪽")
                rcb.reserve_current_research_call(source="안쪽")
            rcb.reserve_current_research_call(source="바깥")
        got = t.snapshot()
        assert got["used"] == 3, got
        assert got["by_source"] == {"안쪽": 2, "바깥": 1}

    def test_it_stops_before_the_network(self, monkeypatch):
        from app.core import research_call_budget as rcb

        monkeypatch.setattr(rl, "router_already_built", lambda: False)
        with rl.canary_request_lock(transport_cap=2) as locked:
            t = locked["transport"]
            with rcb.research_run_scope(cap=99):
                rcb.reserve_current_research_call(source="x")
                rcb.reserve_current_research_call(source="x")
                with pytest.raises(rl.TransportBudgetExceeded):
                    rcb.reserve_current_research_call(source="x")
        assert t.snapshot()["denied"] == 1

    def test_it_is_put_back(self, monkeypatch):
        from app.core import research_call_budget as rcb

        monkeypatch.setattr(rl, "router_already_built", lambda: False)
        before = rcb.reserve_current_research_call
        with rl.canary_request_lock(transport_cap=1):
            assert rcb.reserve_current_research_call is not before
        assert rcb.reserve_current_research_call is before


class TestTheTransportRecordSurvivesDeath:
    """★★★메모리에만 세면 `os._exit` 뒤 **또 0 으로 보인다** (Codex 09-02)."""

    def test_it_writes_before_every_send(self, tmp_path, monkeypatch):
        from app.core import research_call_budget as rcb

        wrote = []
        monkeypatch.setattr(rl, "router_already_built", lambda: False)
        with rl.canary_request_lock(transport_cap=3,
                                    transport_sink=wrote.append):
            rcb.reserve_current_research_call(source="a")
            rcb.reserve_current_research_call(source="b")
        assert [w["used"] for w in wrote] == [1, 2], wrote

    def test_a_denial_is_written_too(self, tmp_path, monkeypatch):
        from app.core import research_call_budget as rcb

        wrote = []
        monkeypatch.setattr(rl, "router_already_built", lambda: False)
        with rl.canary_request_lock(transport_cap=1,
                                    transport_sink=wrote.append):
            rcb.reserve_current_research_call(source="a")
            with pytest.raises(rl.TransportBudgetExceeded):
                rcb.reserve_current_research_call(source="a")
        assert wrote[-1]["denied"] == 1

    def test_it_refuses_to_send_when_it_cannot_record(self, monkeypatch):
        """★★기록을 못 남기면 **보내지 않는다**."""
        from app.core import research_call_budget as rcb

        def _broken(_snap):
            raise OSError("디스크가 꽉 찼다")

        monkeypatch.setattr(rl, "router_already_built", lambda: False)
        with rl.canary_request_lock(transport_cap=5, transport_sink=_broken):
            with pytest.raises(rl.TransportBudgetExceeded) as e:
                rcb.reserve_current_research_call(source="a")
        assert "기록을 못 남겼다" in str(e.value)


class TestBootstrapAndPipelineAreSeparateScopes:
    """★한 계수기가 둘을 같이 세면 **뜻이 섞인다** (Codex 09-02)."""

    def test_switching_scope_starts_a_new_count(self, monkeypatch):
        from app.core import research_call_budget as rcb

        monkeypatch.setattr(rl, "router_already_built", lambda: False)
        with rl.canary_request_lock(transport_cap=4,
                                    transport_name="bootstrap") as locked:
            t = locked["transport"]
            rcb.reserve_current_research_call(source="boot")
            rcb.reserve_current_research_call(source="boot")
            t.set_scope("pipeline", 10)
            rcb.reserve_current_research_call(source="pipe")
        got = t.snapshot()
        assert got["scope"] == "pipeline" and got["used"] == 1
        assert got["cap"] == 10
        earlier = got["earlier_scopes"]
        assert len(earlier) == 1
        assert earlier[0]["scope"] == "bootstrap" and earlier[0]["used"] == 2

    def test_the_run_opens_bootstrap_first_then_pipeline(self):
        import inspect

        from tools.grounding_audit import canary_run as cr

        src = inspect.getsource(cr.run)
        i = src.index('transport_name="bootstrap"')
        assert i < src.index('set_scope("pipeline"')
        assert "cp.run_pipeline(" in src
        assert src.index('set_scope("pipeline"') < src.index("cp.run_pipeline(")

    def test_the_run_opens_only_what_is_left(self):
        """★★승인선을 판마다 **새로 안 연다**."""
        import inspect

        from tools.grounding_audit import canary_run as cr

        src = inspect.getsource(cr.run)
        assert "_left = cp.remaining_cap(" in src
        assert 'set_scope("pipeline", _left)' in src


class TestTheCounterIsThreadSafe:
    """★★★fan-out worker 가 **동시에** 부른다 (Codex 2026-09-02).

    검사→증가→기록을 한 임계구역으로 안 묶으면 cap 을 넘겨 나간다.
    """

    def _hammer(self, counter, n):
        import threading

        ok, denied = [], []

        def _one(i):
            try:
                counter.note(f"w{i % 4}")
                ok.append(i)
            except rl.TransportBudgetExceeded:
                denied.append(i)

        ts = [threading.Thread(target=_one, args=(i,)) for i in range(n)]
        for t in ts:
            t.start()
        for t in ts:
            t.join()
        return ok, denied

    def test_exactly_the_cap_gets_through(self):
        c = rl.TransportCounter(20)
        ok, denied = self._hammer(c, 20)
        assert len(ok) == 20 and denied == []
        assert c.used == 20

    def test_one_over_is_refused(self):
        c = rl.TransportCounter(20)
        ok, denied = self._hammer(c, 30)
        assert len(ok) == 20, f"★{len(ok)} 개가 지나갔다 — cap 을 넘겼다"
        assert len(denied) == 10
        assert c.used == 20 and c.denied == 10

    def test_the_durable_record_ends_at_the_cap(self):
        wrote = []
        c = rl.TransportCounter(15, sink=wrote.append)
        self._hammer(c, 40)
        assert max(w["used"] for w in wrote) == 15
        assert sum(c.by_source.values()) == 15


class TestItPicksUpWhereADeadRunLeftOff:
    """★★앞 판이 죽어 남긴 수를 **0 으로 지우지 않는다**."""

    def test_adopt_raises_the_count(self):
        c = rl.TransportCounter(10)
        c.adopt(6, {"a": 4, "b": 2})
        assert c.used == 6 and c.by_source == {"a": 4, "b": 2}
        c.note("a")
        assert c.used == 7

    def test_adopt_never_lowers_it(self):
        c = rl.TransportCounter(10)
        c.note("a")
        c.note("a")
        c.adopt(1, {"a": 1})
        assert c.used == 2, "★이어받기가 수를 **낮췄다**"

    def test_the_cap_still_holds_after_adopting(self):
        c = rl.TransportCounter(3)
        c.adopt(3, {})
        with pytest.raises(rl.TransportBudgetExceeded):
            c.note("a")

    def test_the_run_reads_the_record_back(self):
        import inspect

        from tools.grounding_audit import canary_run as cr

        from tools.grounding_audit import canary_pipeline as cp

        src = inspect.getsource(cr.run)
        # ★파일 이름은 **한 곳**(`canary_pipeline.TRANSPORT_LOG`)이 안다
        assert "cp.TRANSPORT_LOG" in src
        assert cp.TRANSPORT_LOG.endswith(".jsonl")
        assert '_transport_so_far("bootstrap")' in src
        assert '_transport_so_far("pipeline")' in src
        assert ".adopt(" in src

    def test_the_record_is_append_only(self):
        import inspect

        from tools.grounding_audit import canary_run as cr

        src = inspect.getsource(cr.run)
        assert 'open(_tpath, "a"' in src, "★덮어쓰고 있다"
        assert "os.fsync" in src, "★죽으면 사라진다"


class TestItSurvivesAnActualProcessDeath:
    """★★★진짜 `os._exit` — 예약 1회 뒤 죽어도 durable 이 **남는다**.

    Codex 조건 (2026-09-02): 「실제 subprocess os._exit 끝점: 전송 예약 1회 뒤
    죽음 → durable 1회 보존 → 다음 판은 화해 전 provider 0 또는 화해 후 정확한
    잔여만 허용.」
    """

    CHILD = """
import json, os, sys
sys.path.insert(0, {backend!r})
os.chdir({backend!r})
from tools.grounding_audit import canary_request_lock as rl
from app.core import research_call_budget as rcb

path = {path!r}

def sink(snap):
    with open(path, "a", encoding="utf-8") as fh:
        fh.write(json.dumps({{"scope": snap["scope"], "attempt_id": "",
                              "used": snap["used"],
                              "by_source": snap["by_source"]}}) + chr(10))
        fh.flush()
        os.fsync(fh.fileno())

rl.router_already_built = lambda: False
with rl.canary_request_lock(transport_cap=5, transport_sink=sink):
    rcb.reserve_current_research_call(source="child")
    os._exit(9)          # ★파이썬이 못 잡는다
"""

    def test_the_record_is_there_after_the_kill(self, tmp_path):
        import json
        import subprocess
        import sys
        from pathlib import Path

        backend = str(Path(__file__).resolve().parents[2])
        log = tmp_path / "transport_used.jsonl"
        code = self.CHILD.format(backend=backend, path=str(log))
        got = subprocess.run([sys.executable, "-c", code],
                             capture_output=True, text=True)
        assert got.returncode == 9, got.stderr[-400:]
        rows = [json.loads(x) for x in
                log.read_text(encoding="utf-8").splitlines() if x.strip()]
        assert [r["used"] for r in rows] == [1], rows
        assert rows[0]["by_source"] == {"child": 1}

    def test_the_next_run_reconciles_it(self, tmp_path):
        """★죽은 판의 1 을 **이어받아** 남은 것이 그만큼 준다."""
        import json
        import subprocess
        import sys
        from pathlib import Path

        from tools.grounding_audit import canary_pipeline as cp

        backend = str(Path(__file__).resolve().parents[2])
        log = tmp_path / cp.TRANSPORT_LOG
        code = self.CHILD.format(backend=backend, path=str(log))
        subprocess.run([sys.executable, "-c", code], capture_output=True)
        got = cp.unsettled_transport(tmp_path, "pipeline")
        assert got["used"] == 1, got
        c = rl.TransportCounter(3)
        c.adopt(got["used"], got["by_source"])
        assert c.used == 1
        c.note("a")
        c.note("a")
        with pytest.raises(rl.TransportBudgetExceeded):
            c.note("a")


class TestTheClientLockActuallyPins:
    """★★★앞 판은 **보기만** 했다 (2026-09-02 유료 canary ① 실측).

    중앙 조사가 쓰는 Responses client 가 `max_retries` 없이 지어져
    `'<없음>'` 으로 관측됐고, 파이프라인이 **다 돈 뒤에** 이 문이 서서
    판이 끝났다. openai SDK 기본값은 2 라 한 논리 호출이 최대 3회 나갈 수
    있었다 — 보기 전에 **박아야** raw 상한을 말할 수 있다.
    """

    def test_a_client_built_without_the_kwarg_gets_zero(self, monkeypatch):
        import openai

        from tools.grounding_audit import canary_request_lock as rl

        monkeypatch.setattr(rl, "OBSERVED_CLIENT_RETRIES", [])
        monkeypatch.setattr(rl, "REQUESTED_CLIENT_RETRIES", [])
        seen = {}

        def _fake_init(self, *a, **k):
            seen.update(k)

        monkeypatch.setattr(openai.OpenAI, "__init__", _fake_init,
                            raising=False)
        real = rl._install_client_spy()
        try:
            openai.OpenAI(api_key="x")
        finally:
            rl._remove_client_spy(real)

        assert seen.get("max_retries") == rl.CANARY_SDK_MAX_RETRIES, (
            "★안 준 것을 안 채웠다 — SDK 기본값(2)으로 열린 채 나간다")
        got = rl.assert_client_retries_observed()
        assert got["locked"] is True
        assert got["forced_to_zero"] == 1
        assert got["requested"] == ["<없음>"], "★원래 값이 안 남았다"

    def test_a_client_built_with_a_bad_value_still_stops(self, monkeypatch):
        """★사람이 **명시로** 2 를 주면 그건 계약 위반이라 선다 — 안 덮는다."""
        import openai
        import pytest

        from tools.grounding_audit import canary_request_lock as rl

        monkeypatch.setattr(rl, "OBSERVED_CLIENT_RETRIES", [])
        monkeypatch.setattr(rl, "REQUESTED_CLIENT_RETRIES", [])
        monkeypatch.setattr(openai.OpenAI, "__init__",
                            lambda self, *a, **k: None, raising=False)
        real = rl._install_client_spy()
        try:
            openai.OpenAI(api_key="x", max_retries=2)
        finally:
            rl._remove_client_spy(real)
        with pytest.raises(rl.RequestContractRefused):
            rl.assert_client_retries_observed()