"""PR #82 리뷰(2026-09-03) 에서 병합 전에 고친 다섯 — 각각 결함을 재현하고 고쳐진 것을 잠근다."""
from __future__ import annotations

import ast
import inspect
import json
import threading

import pytest


# ── 1) legacy 지문은 옛 값 그대로 — 기존 에피소드의 재개가 막히지 않는다
class TestLegacyHashIsByteIdentical:
    @pytest.mark.parametrize("mode", ["legacy", "v2"])
    def test_legacy_and_v2_return_the_mixin_base_unchanged(self, monkeypatch, mode):
        from app.core.steps import entity_steps as es
        from app.core.steps.entity_steps import EntityDetailStep, EntityT2iStep
        for cls in (EntityDetailStep, EntityT2iStep):
            st = cls.__new__(cls); st.project_config = {"grounding_mode": mode}; st.project_id, st.episode_id = "p", "e"
            monkeypatch.setattr(es._EntityStepMixin, "_config_hash", lambda self: "md5base16", raising=False)
            monkeypatch.setattr(st, "_a0_candidates", lambda: None, raising=False)
            assert st._config_hash() == "md5base16", f"{cls.__name__}: legacy CP 의 지문이 움직이면 재개가 409 로 막힌다"
            monkeypatch.undo()

    def test_v2_chunk_still_binds_the_contract(self, monkeypatch):
        from app.core.steps import entity_steps as es
        from app.core.steps.entity_steps import EntityT2iStep
        st = EntityT2iStep.__new__(EntityT2iStep); st.project_config = {"grounding_mode": "v2_chunk"}; st.project_id, st.episode_id = "p", "e"
        monkeypatch.setattr(es._EntityStepMixin, "_config_hash", lambda self: "base", raising=False)
        monkeypatch.setattr(st, "_a0_candidates", lambda: [], raising=False)
        assert st._config_hash() != "base"


# ── 4) 우리 문이 세운 것은 uncertain 이 아니라 빈 자리다
class TestALocalStopIsNotUncertain:
    def _journal(self, tmp_path):
        from app.modules.pipeline.grounding_chunk_journal import ChunkJournal
        return ChunkJournal(tmp_path / "j.json", contract={"wiring": "t"})

    def test_budget_stop_before_any_transmission_leaves_the_slot_free(self, tmp_path):
        from app.core.research_call_budget import ResearchCallBudgetExceeded, research_calls_armed
        from app.modules.pipeline import grounding_chunk_journal as cj
        j = self._journal(tmp_path)
        def send():
            raise ResearchCallBudgetExceeded(cap=1, used=1, source="t")
        with research_calls_armed():
            with pytest.raises(ResearchCallBudgetExceeded):
                cj.buy_or_reuse(j, "id1", cap=5, send=send)
        assert j.entries["id1"]["status"] == cj.STATUS_NOT_SENT and j.entries["id1"]["outbound"] == 0   # ★put 은 meta 를 행에 펼친다
        assert j.reserve("id1", cap=5) is True, "★빈 자리 — 다음 판이 산다"

    def test_user_abort_before_any_transmission_leaves_the_slot_free(self, tmp_path):
        from app.core.errors import AppError
        from app.core.research_call_budget import research_calls_armed
        from app.modules.pipeline import grounding_chunk_journal as cj
        j = self._journal(tmp_path)
        def send():
            raise AppError(code="step.cancelled", message="stop", status_code=409)
        with research_calls_armed():
            with pytest.raises(AppError):
                cj.buy_or_reuse(j, "id2", cap=5, send=send)
        assert j.entries["id2"]["status"] == cj.STATUS_NOT_SENT

    def test_a_local_stop_after_one_transmission_is_uncertain(self, tmp_path):
        """★Codex BLOCK: send 하나가 전송 여럿 — 둘째 전송 앞에서 섰어도 첫 전송은 나갔다."""
        from app.core.errors import AppError
        from app.core.research_call_budget import research_calls_armed, research_run_scope, reserve_current_research_call
        from app.modules.pipeline import grounding_chunk_journal as cj
        j = self._journal(tmp_path)
        def send():
            reserve_current_research_call(source="search")          # 첫 전송이 나간다
            raise AppError(code="step.cancelled", message="stop", status_code=409)   # 둘째 앞에서 정지
        with research_run_scope(cap=10):
            with research_calls_armed():
                with pytest.raises(AppError):
                    cj.buy_or_reuse(j, "id3", cap=5, send=send)
        e = j.entries["id3"]
        assert e["status"] == cj.STATUS_UNCERTAIN and e["outbound"] == 1
        assert j.reserve("id3", cap=5) is False, "★샀을 수도 있는 자리는 안 내준다"

    def test_a_local_stop_without_counting_is_uncertain(self, tmp_path):
        """★팔을 안 들었으면(계수 없음) 「안 나갔다」를 증명할 수 없다 — uncertain."""
        from app.core.errors import AppError
        from app.modules.pipeline import grounding_chunk_journal as cj
        j = self._journal(tmp_path)
        def send():
            raise AppError(code="step.cancelled", message="stop", status_code=409)
        with pytest.raises(AppError):
            cj.buy_or_reuse(j, "id4", cap=5, send=send)
        assert j.entries["id4"]["status"] == cj.STATUS_UNCERTAIN and j.entries["id4"]["counted"] is False

    def test_workers_count_only_their_own_attempt(self):
        from app.core.research_call_budget import (OutboundAttempt, bind_current_research_budget, outbound_attempt,
                                                   research_calls_armed, research_run_scope, reserve_current_research_call)
        got = {}
        def worker(name, n):
            att = OutboundAttempt()
            with outbound_attempt(att):
                for _ in range(n):
                    reserve_current_research_call(source=name)
            got[name] = att.count
        with research_run_scope(cap=100):
            with research_calls_armed():
                a = bind_current_research_budget(lambda: worker("a", 3)); b = bind_current_research_budget(lambda: worker("b", 1))
                ta = threading.Thread(target=a); tb = threading.Thread(target=b); ta.start(); tb.start(); ta.join(); tb.join()
        assert got == {"a": 3, "b": 1}

    def test_an_unknown_transport_error_is_still_uncertain(self, tmp_path):
        from app.modules.pipeline import grounding_chunk_journal as cj
        j = self._journal(tmp_path)
        def send():
            raise RuntimeError("connection reset")
        with pytest.raises(RuntimeError):
            cj.buy_or_reuse(j, "id3", cap=5, send=send)
        assert j.entries["id3"]["status"] == cj.STATUS_UNCERTAIN, "★샀을 수도 있는 것은 그대로 uncertain"


# ── 3) 상한은 지금 대상의 자리만 센다 · 스스로 넓히지 않는다
class TestTheCapCountsOnlyCurrentTargets:
    def test_an_orphan_slot_does_not_eat_the_cap(self, tmp_path):
        from app.modules.pipeline.grounding_chunk_journal import ChunkJournal
        j = ChunkJournal(tmp_path / "j.json", contract={"wiring": "t"})
        assert j.reserve("old", cap=1, slot="OLD#detail:1") is True       # 옛 의무 ID 의 자리
        j.put("old", {"x": 1})
        assert j.bought_slots() == 1
        assert j.bought_slots(among={"NEW#detail"}) == 0
        assert j.reserve("new", cap=1, slot="NEW#detail:1", slot_universe={"NEW#detail"}) is True, "★고아 자리가 상한을 안 먹는다"

    def test_the_cap_is_still_a_cap_for_current_targets(self, tmp_path):
        from app.modules.pipeline.grounding_chunk_journal import BudgetExceeded, ChunkJournal
        j = ChunkJournal(tmp_path / "j.json", contract={"wiring": "t"})
        assert j.reserve("a", cap=1, slot="A#detail:1", slot_universe={"A#detail", "B#detail"}) is True
        with pytest.raises(BudgetExceeded):
            j.reserve("b", cap=1, slot="B#detail:1", slot_universe={"A#detail", "B#detail"})

    def test_run_no_longer_widens_its_own_cap(self):
        from app.modules.pipeline import grounding_central_acquisition as ca
        src = inspect.getsource(ca.run)
        tree = ast.parse(src)
        widened = [n.lineno for n in ast.walk(tree)
                   if isinstance(n, (ast.Assign, ast.AugAssign))
                   and "cap_n" in ast.unparse(n).split("=")[0]
                   and "orphan" in ast.unparse(n)]
        assert widened == [], f"상한을 장부 이력으로 넓히는 줄 {widened}"
        assert "slot_universe=known" in src


# ── 2) 사는 lane 에 물리 전송 문이 있다 · worker 도 팔을 든다
class TestTheBuyingLaneHasAPhysicalDoor:
    def test_central_opens_the_run_scope_and_arms(self):
        from app.core.steps.reference_acquisition_step import ReferenceAcquisitionStep as S
        src = inspect.getsource(S._central)
        assert "research_run_scope(" in src and "research_calls_armed()" in src
        assert S.TRANSMISSIONS_PER_LOGICAL_SLOT >= 5

    def test_outdoor_supplement_opens_the_same_door(self):
        import app.core.steps.outdoor_structure_form_reference_step as m
        src = inspect.getsource(m)
        i = src.index("got = ca.run(ledger, journal=journal")
        window = src[max(0, i - 900):i]
        assert "research_run_scope(" in window and "research_calls_armed()" in window

    def test_a_worker_inherits_the_armed_flag_and_the_budget(self):
        from app.core.research_call_budget import (
            ResearchCallBudgetExceeded, bind_current_research_budget, is_armed,
            research_calls_armed, research_run_scope, reserve_current_research_call)
        seen = {}
        def work():
            seen["armed"] = is_armed()
            reserve_current_research_call(source="t")
            reserve_current_research_call(source="t")      # ★두 번째가 cap 1 에 걸려야 한다
        with research_run_scope(cap=1):
            with research_calls_armed():
                bound = bind_current_research_budget(work)
                err = {}
                def _t():
                    try:
                        bound()
                    except BaseException as exc:      # noqa: BLE001
                        err["exc"] = exc
                th = threading.Thread(target=_t); th.start(); th.join()
        assert seen["armed"] is True, "★worker 가 팔을 안 들면 전송이 한 번도 안 세어진다"
        assert isinstance(err.get("exc"), ResearchCallBudgetExceeded)

    def test_without_arming_the_reserve_is_a_noop(self):
        from app.core.research_call_budget import bind_current_research_budget, is_armed, research_run_scope, reserve_current_research_call
        seen = {}
        def work():
            seen["armed"] = is_armed(); reserve_current_research_call(source="t"); reserve_current_research_call(source="t")
        with research_run_scope(cap=1):
            bind_current_research_budget(work)()
        assert seen["armed"] is False


# ── 4) 운영자 도구
class TestTheSettleTool:
    def test_list_then_settle_not_bought_frees_the_slot(self, tmp_path, capsys):
        from app.modules.pipeline import grounding_chunk_journal as cj
        from tools.grounding_audit import journal_settle as tool
        j = cj.ChunkJournal(tmp_path / "j.json", contract={"wiring": "t"})
        def send():
            raise RuntimeError("timeout")
        with pytest.raises(RuntimeError):
            cj.buy_or_reuse(j, "idx", cap=5, send=send)
        assert tool.main(["list", "--journal", str(tmp_path / "j.json")]) == 2
        out = json.loads(capsys.readouterr().out)
        assert [r["identity"] for r in out["pending"]] == ["idx"]
        assert tool.main(["settle", "--journal", str(tmp_path / "j.json"), "--identity", "idx", "--not-bought",
                          "--why", "Opik 에 trace 없음", "--by", "ops"]) == 0
        j2 = cj.ChunkJournal(tmp_path / "j.json", contract={"wiring": "t"})
        assert j2.entries["idx"]["status"] == cj.STATUS_NOT_SENT
        assert tool.main(["list", "--journal", str(tmp_path / "j.json")]) == 0

    def test_settling_as_bought_is_refused(self, tmp_path):
        from tools.grounding_audit import journal_settle as tool
        from app.modules.pipeline import grounding_chunk_journal as cj
        cj.ChunkJournal(tmp_path / "j.json", contract={"wiring": "t"}).put("i", None, status=cj.STATUS_UNCERTAIN)
        with pytest.raises(SystemExit):
            tool.main(["settle", "--journal", str(tmp_path / "j.json"), "--identity", "i", "--why", "x", "--by", "y"])


# ── Codex 재리뷰 A·B: settle 도구는 contract 를 지키고 release 는 정정 사건이다
class TestTheSettleToolKeepsTheLedgerAppendOnly:
    def _uncertain_journal(self, tmp_path, contract):
        from app.modules.pipeline import grounding_chunk_journal as cj
        j = cj.ChunkJournal(tmp_path / "j.json", contract=contract)
        def send():
            raise RuntimeError("timeout")
        with pytest.raises(RuntimeError):
            cj.buy_or_reuse(j, "idx", cap=5, send=send)
        return tmp_path / "j.json"

    def test_settle_keeps_the_stored_contract_and_the_calls_bytes(self, tmp_path):
        from tools.grounding_audit import journal_settle as tool
        path = self._uncertain_journal(tmp_path, {"wiring": "keep-me", "supplement": "v1"})
        before = json.loads(path.read_text(encoding="utf-8"))
        assert tool.main(["settle", "--journal", str(path), "--identity", "idx", "--not-bought", "--why", "Opik 없음", "--by", "ops"]) == 0
        after = json.loads(path.read_text(encoding="utf-8"))
        assert after["contract"] == {"wiring": "keep-me", "supplement": "v1"} == before["contract"], "★contract 가 {} 로 덮이면 production 이 drift 로 선다"
        assert after["calls"] == before["calls"], "★원행은 한 바이트도 안 바뀐다"
        assert len(after["settlements"]) == len(before["settlements"]) + 1
        # ★production 이 같은 contract 로 다시 열어도 drift 가 아니다
        from app.modules.pipeline.grounding_chunk_journal import ChunkJournal
        assert ChunkJournal(path, contract={"wiring": "keep-me", "supplement": "v1"}).contract_drifted() is False

    def test_release_is_a_settlement_and_only_for_reserved(self, tmp_path):
        from app.modules.pipeline import grounding_chunk_journal as cj
        from tools.grounding_audit import journal_settle as tool
        path = tmp_path / "j.json"
        j = cj.ChunkJournal(path, contract={"wiring": "t"})
        assert j.reserve("dead", cap=5) is True                       # worker 가 자리를 잡고 죽었다
        before = json.loads(path.read_text(encoding="utf-8"))
        assert tool.main(["release", "--journal", str(path), "--identity", "dead", "--why", "worker 죽음", "--by", "ops"]) == 0
        after = json.loads(path.read_text(encoding="utf-8"))
        assert after["calls"] == before["calls"] and len(after["settlements"]) == 1
        assert after["settlements"][0]["from"] == "reserved" and after["settlements"][0]["to"] == "released"
        j2 = cj.ChunkJournal(path, contract={"wiring": "t"})
        assert j2.entries["dead"]["status"] == cj.STATUS_RELEASED and j2.reserve("dead", cap=5) is True
        # ★uncertain 줄은 release 로 못 놓는다
        j2.put("unc", None, status=cj.STATUS_UNCERTAIN)
        with pytest.raises(ValueError):
            j2.settle_reserved("unc", why="x", decided_by="y")

    def test_a_journal_without_a_contract_is_refused(self, tmp_path):
        from tools.grounding_audit import journal_settle as tool
        path = tmp_path / "j.json"
        path.write_text(json.dumps({"contract": {}, "calls": [], "settlements": []}), encoding="utf-8")
        with pytest.raises(SystemExit):
            tool.main(["list", "--journal", str(path)])


# ── Codex 재리뷰 C: 사는 lane 은 SDK·Router 재시도 0 이라 예약 수 = 전송 수
class TestTheLaneLocksRetriesToZero:
    def test_search_and_writer_clients_have_sdk_retries_zero(self, monkeypatch):
        from app.core.steps import reference_acquisition_step as ras
        made = []
        class _Fake:
            def __init__(self, **kw):
                made.append(kw)
        import app.core.openai_keys as ok
        monkeypatch.setattr(ok, "FailoverOpenAIClient", _Fake)
        ras.make_search()
        ras.make_writer(world={}, source_text="", project_id="p", episode_id="e", step_id="s")
        assert made and all(kw.get("max_retries") == 0 for kw in made), made

    def test_the_judge_passes_router_retries_zero(self, monkeypatch):
        from app.core.steps import reference_acquisition_step as ras
        import app.modules.llm.llm_client as lc
        seen = {}
        def fake_call_structured(*a, **kw):
            seen.update(kw); return {"verdicts": []}
        monkeypatch.setattr(lc, "call_structured", fake_call_structured)
        from app.modules.pipeline import coarse_type_pick as ctp
        import app.modules.pipeline.multiroll_gemini as mg
        monkeypatch.setattr(ctp, "load_pack", lambda: {"stems": {ctp.SYSTEM_STEM: {"content": "s"}, ctp.SCHEMA_STEM: {"content": {}}}})
        monkeypatch.setattr(mg, "png_part", lambda p: {"type": "text", "text": "png"})
        judge = ras.make_judge(project_id="p", episode_id="e", step_id="s")
        judge([{"index": 1, "path": "/nope/x.png", "kind_name": "k"}])
        assert seen.get("num_retries") == 0

    def test_the_cap_derivation_matches_the_locked_retries(self):
        from app.core.steps import reference_acquisition_step as ras
        assert ras.LANE_CLIENT_KWARGS == {"max_retries": 0} and ras.LANE_LLM_NUM_RETRIES == 0
        # 조사 1 + 검색 2 + 판정 2×3 tier = 9 논리 · 키 슬롯 ×2 = 18
        assert ras.ReferenceAcquisitionStep.TRANSMISSIONS_PER_LOGICAL_SLOT == (1 + 2 + 2 * 3) * 2