"""GROUNDING-V2 §4b — 검색 배선. ★**바깥 호출 0**으로 batch 기구를 잰다.

★막는 자리는 **진짜 바깥 경계**(`client.responses.create`)다. 그 위의 우리
함수를 막으면 우리 코드가 안 돌고, 안 도는 코드는 시험이 안 잰다 — 이 판에서
이미 한 번 그렇게 놓쳤다.
"""
import json
import pathlib
from types import SimpleNamespace
from unittest.mock import patch

import pytest

from app.modules.pipeline import grounding_claims_search as g

#: ★probe 는 **주소와 응답 원형 둘 다**를 본다 — 산 경로는 늘 원형을
#:  넣으므로, 모양만 재는 시험에도 정상 원형을 같이 준다.
_DUMP = {"id": "resp_test", "output": []}



def _a0_manifest():
    """도구 모듈을 **파일 경로로** 읽는다.

    ★`from tools.prompt_measure import …` 은 쓸 수 없다 — 저장소 **루트에도
    `tools/` 패키지가 있어서** 수집 순서에 따라 그쪽이 `backend/tools` 를 가린다.
    혼자 돌 때는 통과하고 `tests/core` 와 같이 돌면 `ModuleNotFoundError` 가
    나는, **순서에 흔들리는 시험**이었다.
    """
    import importlib.util
    import pathlib as _pl

    path = (_pl.Path(__file__).resolve().parents[2] / "tools"
            / "prompt_measure" / "grounding_a0_manifest.py")
    spec = importlib.util.spec_from_file_location("_a0_manifest_probe", path)
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


def _batch_experiment():
    """실행기도 **파일 경로로** 읽는다 — 루트 `tools` 가 가린다."""
    import importlib.util
    import pathlib as _pl

    path = (_pl.Path(__file__).resolve().parents[2] / "tools"
            / "prompt_measure" / "grounding_batch_experiment.py")
    spec = importlib.util.spec_from_file_location("_batch_exp_probe", path)
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


class _Dumpable(SimpleNamespace):
    """★실제 SDK 응답은 `model_dump` 를 가진다 — 대역도 가져야 그 경로가 돈다."""

    def model_dump(self, **_kw):
        def _plain(x):
            if isinstance(x, SimpleNamespace):
                return {k: _plain(v) for k, v in vars(x).items()}
            if isinstance(x, list):
                return [_plain(v) for v in x]
            return x
        return _plain(self)


def _ph(subjects, *, model="m", era="e", region="r"):
    """그 호출의 실제 `payload_hash`. ★손으로 지어내면 신원이 안 맞는다."""
    out = g.search_claims(_Client(), subjects, model=model, era=era,
                          region=region, batch_size=len(subjects) or 1)
    return out["batches"][0]["provenance"]["payload_hash"]


def _subject(sid, surface="쇠사슬로 묶인 기계식 요금통", quote="원문 문장"):
    return {"research_subject_id": sid, "surface_form": surface,
            "owner_type": "prop", "source_quote": quote}


def _resp(subject_ids, *, resp_id="resp_1", sources=(("q", "https://a.org",
                                                     "제목", "본문 조각"),)):
    """provider 응답 대역.

    ★모양을 **실제 SDK 와 같게** 만들려 했는데 지어낸 필드가 있었다 —
    그 결함은 `TestTheRealSdkShapeIsWhatWeRead` 가 **진짜 SDK 객체**로 잡는다.
    여기서는 batch 기구를 재는 데 필요한 만큼만 흉내 낸다.
    """
    body = {"results": [{"research_subject_id": s, "claims": [], "gaps": []}
                        for s in subject_ids]}
    return _Dumpable(
        id=resp_id,
        output=[
            SimpleNamespace(
                type="web_search_call",
                action=SimpleNamespace(queries=[t[0] for t in sources]),
                results=[SimpleNamespace(url=t[1], title=t[2], snippet=t[3])
                         for t in sources]),
            SimpleNamespace(
                type="message",
                content=[SimpleNamespace(
                    text="```json\n" + json.dumps(body, ensure_ascii=False)
                         + "\n```")]),
        ])




def _no_pages(monkeypatch, m):
    """★시험은 **페이지를 안 받는다.** 진짜 `check_run` 을 쓰되 fetcher 만 막는다.

    돈 가드가 opener 층에서 막으면 `RuntimeError` 가 올라와 주행이 통째로
    터진다 — 여기서 재는 것은 인용 확인이 아니라 재사용·멈춤이다.
    """
    from app.modules.pipeline.grounding_claims_support import check_run

    # ★진짜 `check_run` 을 쓰되 **페이지만 가짜**다 — 인용 조각이 그 안에
    #  있으므로 「있다」가 나온다. `None` 을 주면 「못 열었다」가 되어
    #  기준선 gate 에 걸리는데, 여기서 재는 것은 그 gate 가 아니다.
    monkeypatch.setattr(m, "support_check_run",
                        lambda batches: check_run(
                            batches, fetch=lambda _u: "…본문 조각…"))


def _resp_with_claims(ids, *, resp_id="r"):
    """★확정된 claim 이 **있는** 응답. 기본 대역은 claim 0 이라 기준선 멈춤에
    걸린다 — 크기 비교까지 가는 시험은 이걸 쓴다."""
    body = {"results": [{
        "research_subject_id": s,
        "claims": [{
            "kind": "fact", "required": True,
            "statement_native": f"{s} 에 대한 문장",
            "delta_effect": "supports_difference",
            "target_binding": {"is_about_target": True,
                               "target_words_in_statement": s,
                               "why": "대상 자체"},
            "sources": ["https://a.org"],
            "evidence_span": "본문 조각"}],
        "gaps": []} for s in ids]}
    return _Dumpable(id=resp_id, output=[
        SimpleNamespace(type="web_search_call",
                        action=SimpleNamespace(queries=["q"]),
                        results=[SimpleNamespace(url="https://a.org",
                                                 title="제목",
                                                 snippet="본문 조각")]),
        SimpleNamespace(type="message", content=[SimpleNamespace(
            text=json.dumps(body, ensure_ascii=False))])])



def _resp_with_sourced_no(ids, *, resp_id="r"):
    """★출처로 **「차이 없음」을 확정**한 응답. 이것도 조사가 된 것이다 —
    빈손이 아니다. `yes` 만 보면 이걸 빈손으로 오인한다."""
    body = {"results": [{
        "research_subject_id": s,
        "claims": [{
            "kind": "fact", "required": True,
            "statement_native": f"{s} 는 지금과 다르지 않다",
            "delta_effect": "supports_no_difference",
            "target_binding": {"is_about_target": True,
                               "target_words_in_statement": s,
                               "why": "대상 자체"},
            "sources": ["https://a.org"],
            "evidence_span": "본문 조각"}],
        "gaps": []} for s in ids]}
    return _Dumpable(id=resp_id, output=[
        SimpleNamespace(type="web_search_call",
                        action=SimpleNamespace(queries=["q"]),
                        results=[SimpleNamespace(url="https://a.org",
                                                 title="제목",
                                                 snippet="본문 조각")]),
        SimpleNamespace(type="message", content=[SimpleNamespace(
            text=json.dumps(body, ensure_ascii=False))])])


class _Client:
    """★호출마다 **무엇을 받았는지** 남긴다 — batch 경계를 그것으로 잰다."""

    def __init__(self, maker=None, fail_on=()):
        self.calls = []
        self._maker = maker
        self._fail_on = set(fail_on)
        self.responses = SimpleNamespace(create=self._create)

    def _create(self, **kw):
        n = len(self.calls)
        self.calls.append(kw)
        if n in self._fail_on:
            raise RuntimeError("ConnectionResetError: peer closed")
        text = kw["input"][0]["content"][0]["text"]
        ids = [ln[4:].strip() for ln in text.splitlines()
               if ln.startswith("### ")]
        return (self._maker or _resp)(ids, resp_id=f"resp_{n}")


@pytest.fixture(autouse=True)
def _no_tracer():
    """호출 기록은 별도 축이다 — 여기서는 batch 기구만 잰다."""
    with patch("app.modules.llm.image_tracer.record_provider_call",
               return_value="trace_1"), \
         patch("app.modules.llm.image_tracer.ambient_call_meta",
               return_value={}), \
         patch("app.modules.llm.image_tracer.resolve_step_name",
               return_value=g.STEP_NAME):
        yield


class TestBatchOnlyCuts:
    """★batch 는 **자르기만 한다.** 크기를 바꿔도 대상이 늘거나 줄지 않는다."""

    @pytest.mark.parametrize("n,size,expect_calls", [
        (12, 1, 12), (12, 2, 6), (12, 4, 3), (12, 8, 2), (12, 12, 1),
        (5, 2, 3),   # 나머지가 남는 판
    ])
    def test_the_call_count_is_the_ceiling(self, n, size, expect_calls):
        subs = [_subject(f"rs_{i:02}") for i in range(n)]
        c = _Client()
        out = g.search_claims(c, subs, model="m", era="1983년", region="한국",
                              batch_size=size)
        assert len(c.calls) == expect_calls
        assert out["logical_calls"] == expect_calls

    @pytest.mark.parametrize("size", [1, 2, 4, 8, 12])
    def test_every_subject_is_asked_exactly_once(self, size):
        """★크기를 바꿔도 **하나도 빠지지 않고 겹치지도 않는다**."""
        subs = [_subject(f"rs_{i:02}") for i in range(12)]
        c = _Client()
        out = g.search_claims(c, subs, model="m", era="1983년", region="한국",
                              batch_size=size)
        asked = [s for b in out["batches"] for s in b["requested"]]
        assert sorted(asked) == sorted(s["research_subject_id"] for s in subs)

    def test_the_order_does_not_change_with_batch_size(self):
        """★순서가 흔들리면 같은 실험이 아니다."""
        subs = [_subject(f"rs_{i:02}") for i in range(12)]
        seqs = []
        for size in (1, 2, 4, 8, 12):
            out = g.search_claims(_Client(), subs, model="m", era="e",
                                  region="r", batch_size=size)
            seqs.append([s for b in out["batches"] for s in b["requested"]])
        assert all(s == seqs[0] for s in seqs)


class TestTheBatchSizeHasNoDefault:
    """★기본값을 두면 실험이 정하기 전에 그 값이 **계약처럼 굳는다** (Codex)."""

    def test_it_is_a_required_argument(self):
        with pytest.raises(TypeError):
            g.search_claims(_Client(), [_subject("rs_1")], model="m",
                            era="e", region="r")

    @pytest.mark.parametrize("bad", [0, -1, "4", 2.0, None, True, False])
    def test_a_bad_size_is_refused(self, bad):
        """★`isinstance(True, int)` 는 참이라 `batch_size=True` 가 **batch 1 로
        조용히 돌았다**. 이 판에서 `bool` 로 다섯 번째 데인 부류다."""
        with pytest.raises(ValueError):
            g.search_claims(_Client(), [_subject("rs_1")], model="m",
                            era="e", region="r", batch_size=bad)

    def test_the_module_declares_no_default(self):
        assert g.BATCH_SIZE_UNSET is None


class TestTheSubjectIdIsCarriedNotGuessed:
    """★이름이 비슷한 두 대상에서 **짐작은 반드시 틀린다**."""

    def test_the_prompt_names_every_subject_id(self):
        subs = [_subject("rs_a", "시외버스 1호차 내부"),
                _subject("rs_b", "시외버스 2호차 내부")]
        c = _Client()
        g.search_claims(c, subs, model="m", era="e", region="r", batch_size=2)
        text = c.calls[0]["input"][0]["content"][0]["text"]
        assert "### rs_a" in text and "### rs_b" in text

    def test_a_subject_without_an_id_is_refused(self):
        with pytest.raises(ValueError, match="research_subject_id"):
            g.search_claims(_Client(), [{"surface_form": "무엇"}],
                            model="m", era="e", region="r", batch_size=1)

    @pytest.mark.parametrize("era,region", [("", "한국"), ("1983", ""),
                                            ("  ", "한국")])
    def test_a_missing_context_is_refused(self, era, region):
        """★맥락 없이 조사하면 **무엇과 비교할지가 없다**."""
        with pytest.raises(ValueError, match="시대/지역"):
            g.search_claims(_Client(), [_subject("rs_1")], model="m",
                            era=era, region=region, batch_size=1)


class TestTheSourceTextIsNotCut:
    """★절대 규칙 — LLM 에 가는 원문을 **자르지 않는다**."""

    def test_a_long_quote_survives_whole(self):
        quote = "가" * 4000
        c = _Client()
        g.search_claims(c, [_subject("rs_1", quote=quote)], model="m",
                        era="e", region="r", batch_size=1)
        assert quote in c.calls[0]["input"][0]["content"][0]["text"]


class TestOnlyTextComesBack:
    """★사진은 참조 층이 따로 산다 — 여기서 같이 받으면 값을 매번 치른다."""

    def test_the_search_tool_is_text_only(self):
        c = _Client()
        g.search_claims(c, [_subject("rs_1")], model="m", era="e", region="r",
                        batch_size=1)
        tool = c.calls[0]["tools"][0]
        assert tool == {"type": "web_search"}
        assert "image_settings" not in tool


class TestWhatTheSearchActuallySawIsKept:
    """★★`evidence_span` 이 진짜 그 페이지에 있는지 확인하려면 **본 것**이
    남아야 한다. 확인 못 하는 것은 「문제없음」이 아니라 미확정이다 (Codex).
    """

    def test_the_snippets_are_collected(self):
        c = _Client()
        out = g.search_claims(c, [_subject("rs_1")], model="m", era="e",
                              region="r", batch_size=1)
        got = out["batches"][0]["sources"]
        assert got and got[0]["url"] == "https://a.org"
        assert got[0]["snippet"] == "본문 조각"

    def test_a_response_without_results_leaves_an_empty_list(self):
        """★없는 것을 **없다고** 남긴다 — 조용히 빼면 나중에 구분이 안 된다."""
        def _bare(ids, **kw):
            r = _resp(ids, **kw)
            r.output = [o for o in r.output if o.type == "message"]
            return r
        out = g.search_claims(_Client(maker=_bare), [_subject("rs_1")],
                              model="m", era="e", region="r", batch_size=1)
        assert out["batches"][0]["sources"] == []


class TestProvenanceIsAssembledHere:
    """★검색 배선이 **첫 실제 writer** 다 — 11칸을 여기서 채운다."""

    def _prov(self, **kw):
        out = g.search_claims(_Client(**kw), [_subject("rs_1")], model="m",
                              era="e", region="r", batch_size=1)
        return out["batches"][0]["provenance"]

    def test_a_successful_call_carries_the_request_id(self):
        p = self._prov()
        assert p["transmission_status"] == "ok"
        assert p["provider_request_id"] == "resp_0"
        assert p["local_trace_id"] == "trace_1"

    def test_the_pack_coordinates_are_real(self):
        p = self._prov()
        assert p["prompt_locator"] and p["schema_locator"]
        assert p["prompt_raw_hash"] != p["schema_raw_hash"]
        assert p["prompt_pack_version"] == g.PROMPT_PACK_VERSION

    def test_a_broken_transmission_is_recorded_not_swallowed(self):
        """★★전송이 깨져도 **행이 남는다** — 무엇이 왜 실패했는지가 몫이다."""
        p = self._prov(fail_on=(0,))
        assert p["transmission_status"] == "failed"
        assert "ConnectionResetError" in p["transmission_error"]
        assert "provider_request_id" not in p

    def test_one_broken_batch_does_not_kill_the_others(self):
        """★한 batch 가 깨졌다고 나머지를 안 산 것으로 만들지 않는다."""
        subs = [_subject(f"rs_{i}") for i in range(4)]
        out = g.search_claims(_Client(fail_on=(0,)), subs, model="m", era="e",
                              region="r", batch_size=2)
        assert len(out["batches"]) == 2
        assert out["batches"][0]["error"]
        assert not out["batches"][1]["error"]

    def test_the_payload_hash_moves_with_the_batch(self):
        """★같은 hash 가 두 batch 에 붙으면 무엇을 보냈는지 못 되짚는다."""
        subs = [_subject(f"rs_{i}") for i in range(4)]
        out = g.search_claims(_Client(), subs, model="m", era="e", region="r",
                              batch_size=2)
        a, b = (x["provenance"]["payload_hash"] for x in out["batches"])
        assert a != b


class TestThisFunctionDoesNotJudge:
    """★조사와 채점이 같은 함수에 있으면 **채점이 조사를 봐준다**."""

    def test_the_raw_text_is_kept_as_is(self):
        out = g.search_claims(_Client(), [_subject("rs_1")], model="m",
                              era="e", region="r", batch_size=1)
        assert out["batches"][0]["raw"].startswith("```json")

    def test_unparseable_output_is_not_turned_into_an_empty_result(self):
        """★파싱 실패를 `{}` 로 삼키면 **조사 0건이 「깨끗함」**으로 읽힌다."""
        def _junk(ids, **kw):
            return SimpleNamespace(id="r", output=[SimpleNamespace(
                type="message",
                content=[SimpleNamespace(text="검색에 실패했습니다")])])
        out = g.search_claims(_Client(maker=_junk), [_subject("rs_1")],
                              model="m", era="e", region="r", batch_size=1)
        assert out["batches"][0]["parsed"] is None
        assert out["batches"][0]["raw"] == "검색에 실패했습니다"


class TestTheA0GateNeverWidensItsOwnApprovedScope:
    """★★가드와 대상 목록이 **따로 놀면 둘 중 하나가 거짓말**이 된다.

    처음에 `MAX_A0_CALLS=1` 로 박아 두고 목록에 2개를 넣었더니, 선언한 fallback
    이 **닿을 수 없는 코드**였다. 고치면서 상한을 2로 올렸는데 그것도 틀렸다 —
    이번 주행의 승인 범위는 **한 에피소드**이고, 12개를 못 채우면 다른
    에피소드로 넘어가는 게 아니라 **그 자리에서 서서 보고**한다 (Codex).

    ★코드가 스스로 승인 범위를 넓히면 안 된다. 다음 후보는 적어 두되
    (결과를 본 뒤 고르지 않기 위해) **손으로 옮겨야** 돈다.
    """

    def test_the_cap_is_derived_from_the_approved_list(self):
        m = _a0_manifest()

        assert m.MAX_A0_CALLS == len(m.EPISODE_ORDER)

    def test_the_approved_run_is_one_episode(self):
        m = _a0_manifest()

        assert len(m.EPISODE_ORDER) == 1

    def test_the_next_candidate_is_written_down_but_not_approved(self):
        """★어디로 갈지를 **결과를 본 뒤** 고르지 않기 위해 적어 둔다."""
        m = _a0_manifest()

        assert m.NEXT_CANDIDATES
        approved = {(p, e) for p, e, _ in m.EPISODE_ORDER}
        assert not approved & {(p, e) for p, e, _ in m.NEXT_CANDIDATES}

    def test_a_short_result_stops_instead_of_moving_on(self, monkeypatch):
        """★★끝점 — 모자라면 **다음 후보를 안 부르고 선다**."""
        from pathlib import Path

        m = _a0_manifest()

        seen = []

        def _fake(fulltext, **kw):
            seen.append(kw["episode_id"])
            return {"candidates": [{"surface_form": "x"}], "hallucinated": []}

        monkeypatch.setattr(m, "_episode_dir",
                            lambda root, p, e: Path(f"/fake/{p}/{e}"))
        monkeypatch.setattr(m, "_cleaned_text", lambda ep: "원문")
        monkeypatch.setattr(m, "_rules", lambda ep: {"era": "e", "region": "r"})
        import app.modules.pipeline.grounding_a0 as a0
        monkeypatch.setattr(a0, "collect_candidates", _fake)

        with pytest.raises(SystemExit, match="여기서"):
            m.run_a0(Path("/fake"))
        assert len(seen) == 1, "승인 안 된 에피소드까지 샀다"

    def test_a_passing_result_returns_the_approved_episode(self, monkeypatch):
        """★positive control — 12개가 나오면 그대로 고른다."""
        from pathlib import Path

        m = _a0_manifest()

        monkeypatch.setattr(m, "_episode_dir",
                            lambda root, p, e: Path(f"/fake/{p}/{e}"))
        monkeypatch.setattr(m, "_cleaned_text", lambda ep: "원문")
        monkeypatch.setattr(m, "_rules", lambda ep: {"era": "e", "region": "r"})
        import app.modules.pipeline.grounding_a0 as a0
        monkeypatch.setattr(a0, "collect_candidates", lambda t, **kw: {
            "candidates": [{"surface_form": f"x{i}"} for i in range(12)],
            "hallucinated": []})
        assert m.run_a0(Path("/fake"))["coord"].endswith(
            m.EPISODE_ORDER[0][1])


class TestTheSizeCapSpillsInsteadOfCutting:
    """★★상한이 **둘**이다 — subject 수와 직렬화된 prompt 크기 (Codex).

    §4b 실험은 1.3KB 검증 원고로 한다. 거기서 나온 subject 수를 실제 에피소드에
    그대로 쓰면 prompt 가 몇 배가 된다 — 원고 길이가 대상마다 다르니 **개수만으로는
    크기를 못 정한다**.

    ★크기를 넘어도 **원문을 자르지 않는다.** 자르는 순간 그 대상은 다른 것을
    조사한 것이 되고, 결과가 `subject_payload_hash` 와도 안 맞는다.
    """

    def test_a_big_subject_spills_to_the_next_batch(self):
        subs = [_subject(f"rs_{i}", quote="가" * 3000) for i in range(4)]
        c = _Client()
        out = g.search_claims(c, subs, model="m", era="e", region="r",
                              batch_size=4, max_prompt_bytes=6000)
        assert len(out["batches"]) > 1, "크기 상한이 안 걸렸다"
        assert all(b["prompt_bytes"] <= 6000 or b["subject_count"] == 1
                   for b in out["batches"])

    def test_nothing_is_dropped_when_it_spills(self):
        """★넘기는 것과 **빼는 것**은 다르다."""
        subs = [_subject(f"rs_{i}", quote="가" * 3000) for i in range(7)]
        out = g.search_claims(_Client(), subs, model="m", era="e", region="r",
                              batch_size=7, max_prompt_bytes=6000)
        asked = [s for b in out["batches"] for s in b["requested"]]
        assert sorted(asked) == sorted(s["research_subject_id"] for s in subs)

    def test_a_single_oversized_subject_still_gets_its_own_batch(self):
        """★★혼자서 상한을 넘어도 **건너뛰지 않는다** — 건너뛰면 조용히 조사에서
        빠지고 「12개를 다 샀다」가 거짓이 된다."""
        subs = [_subject("rs_big", quote="가" * 50_000)]
        out = g.search_claims(_Client(), subs, model="m", era="e", region="r",
                              batch_size=4, max_prompt_bytes=100)
        assert [b["requested"] for b in out["batches"]] == [["rs_big"]]
        # ★원문이 통째로 나갔는지 — 자르지 않았다는 뜻이다
        assert out["batches"][0]["prompt_bytes"] > 50_000

    def test_the_actual_size_is_recorded(self):
        """★크기를 안 남기면 검증 원고의 수를 실제 에피소드에 써도 되는지
        판단할 근거가 없다."""
        out = g.search_claims(_Client(), [_subject("rs_1")], model="m",
                              era="e", region="r", batch_size=1)
        b = out["batches"][0]
        assert b["prompt_bytes"] > 0 and b["subject_count"] == 1
        assert out["max_prompt_bytes_seen"] == b["prompt_bytes"]

    def test_the_size_cap_does_not_reorder(self):
        """★넘기더라도 **순서는 그대로**다 — 흔들리면 같은 실험이 아니다."""
        subs = [_subject(f"rs_{i:02}", quote="가" * (500 * (i + 1)))
                for i in range(8)]
        out = g.search_claims(_Client(), subs, model="m", era="e", region="r",
                              batch_size=8, max_prompt_bytes=4000)
        asked = [s for b in out["batches"] for s in b["requested"]]
        assert asked == [s["research_subject_id"] for s in subs]

    @pytest.mark.parametrize("bad", [0, -1, "4", 2.0, None, True])
    def test_a_bad_size_cap_is_refused(self, bad):
        with pytest.raises(ValueError, match="max_prompt_bytes"):
            g.search_claims(_Client(), [_subject("rs_1")], model="m", era="e",
                            region="r", batch_size=1, max_prompt_bytes=bad)


class TestWhatWasPaidForIsSavedBeforeItIsJudged:
    """★★유료로 산 것은 **판정과 무관하게 먼저 남긴다**.

    실물 — A0 를 한 번 사서 후보 **23개**를 받았는데, 게이트가 먼저 서면서 그
    23개가 통째로 사라졌다. 무엇이 나왔는지 보려면 **다시 사야** 했다.
    「기록이 안 남으면 되짚을 수 없다」를 하루 종일 남의 코드에 적용해 놓고
    내 도구에는 안 했다.
    """

    def test_the_raw_output_is_written_even_when_the_gate_stops(self, tmp_path,
                                                                monkeypatch):
        from pathlib import Path

        m = _a0_manifest()

        monkeypatch.setattr(m, "_episode_dir",
                            lambda root, p, e: Path(f"/fake/{p}/{e}"))
        monkeypatch.setattr(m, "_cleaned_text", lambda ep: "원문")
        monkeypatch.setattr(m, "_rules", lambda ep: {"era": "e", "region": "r"})
        import app.modules.pipeline.grounding_a0 as a0
        monkeypatch.setattr(a0, "collect_candidates", lambda t, **kw: {
            "candidates": [{"surface_form": f"x{i}"} for i in range(23)],
            "hallucinated": []})
        # 게이트는 반드시 선다 — 결속을 0으로 만든다
        monkeypatch.setattr(
            "app.modules.pipeline.grounding_shadow"
            ".build_subjects_from_saved_episode",
            lambda *a, **k: {"subjects": [], "unbound": []})

        out = tmp_path / "manifest.json"
        monkeypatch.setattr("sys.argv", [
            "x", "--run", "--out", str(out), "--projects-root", "/fake"])
        with pytest.raises(SystemExit):
            m.main()
        raw = out.with_name(out.stem + "_a0_raw.json")
        assert raw.exists(), "게이트가 서면서 산 것이 사라졌다"
        assert len(json.loads(raw.read_text(encoding="utf-8"))
                   ["a0"]["candidates"]) == 23

    def test_the_stop_message_says_why_it_is_short(self, monkeypatch):
        """★「2개뿐이다」만으로는 **A0 가 못 건진 것인지 결속이 안 된 것인지**
        구분이 안 된다 — 다음에 무엇을 고칠지가 갈린다."""
        from pathlib import Path

        m = _a0_manifest()

        monkeypatch.setattr(
            "app.modules.pipeline.grounding_shadow"
            ".build_subjects_from_saved_episode",
            lambda *a, **k: {"subjects": [{"quote_source": "entity_description"}],
                             "unbound": [{}, {}],
                             "carry_reasons": {"ambiguous": 5, "none": 1}})
        monkeypatch.setattr(m, "_cleaned_text", lambda ep: "원문")
        with pytest.raises(SystemExit, match="ambiguous"):
            m.build_manifest({
                "episode_dir": "/fake", "project_id": "p", "episode_id": "e",
                "a0": {"candidates": [{}] * 23}})


class TestSoftAndHardSizeCapsAreDifferentThings:
    """★★soft 를 혼자 넘으면 **보낸다**(singleton). hard 를 넘으면 **안 보낸다**.

    보내 봐야 provider 가 거절하고, 그 거절을 「조사했는데 못 찾았다」로 읽으면
    거짓이 된다 (Codex).
    """

    def test_a_soft_oversized_subject_is_still_sent_and_counted(self):
        subs = [_subject("rs_big", quote="가" * 20_000)]
        c = _Client()
        out = g.search_claims(c, subs, model="m", era="e", region="r",
                              batch_size=4, max_prompt_bytes=1000,
                              hard_prompt_bytes=10_000_000)
        assert len(c.calls) == 1, "soft 를 넘었다고 안 보냈다"
        assert out["oversized_subject_count"] == 1
        assert out["not_sent_count"] == 0

    def test_a_hard_oversized_subject_is_not_sent(self):
        subs = [_subject("rs_big", quote="가" * 20_000)]
        c = _Client()
        out = g.search_claims(c, subs, model="m", era="e", region="r",
                              batch_size=4, max_prompt_bytes=1000,
                              hard_prompt_bytes=5000)
        assert c.calls == [], "hard 를 넘었는데 보냈다"
        assert out["not_sent_count"] == 1
        assert out["batches"][0]["not_sent"] is True

    def test_a_not_sent_batch_says_the_transmission_failed(self):
        """★★전송은 없었지만 **무엇을 보내려 했는지는 안다** (Codex).

        전에는 `provenance=None` 이었다. 그러면 저장 계약이 터지거나(스텝 예외)
        걸러 버려 **조용히 사라진다** — 둘 다 틀렸다. 전송 0인 bounded-run
        미결로 남아 재개할 수 있어야 한다.
        ★응답이 없었으므로 `provider_request_id` 는 **비어 있는 것이 맞다**.
        """
        out = g.search_claims(_Client(), [_subject("rs_big", quote="가" * 9000)],
                              model="m", era="e", region="r", batch_size=1,
                              max_prompt_bytes=100, hard_prompt_bytes=500)
        row = out["batches"][0]
        # ★provider 실패가 **아니다** — 물리 전송 0인 로컬 상한이다
        assert row["limit_kind"] == "admission_limit"
        prov = row["provenance"]
        assert prov["transmission_status"] == "not_sent"
        assert prov["provider_request_id"] == ""
        # ★팩 좌표와 그 호출의 신원은 **실재한다** — 지어낸 값이 아니다
        assert prov["payload_hash"] and prov["prompt_raw_hash"]
        assert "상한" in prov["transmission_error"]

    def test_a_not_sent_batch_still_names_its_subjects(self):
        """★안 보낸 것도 **누구였는지**는 남는다 — 조용히 사라지면 안 된다."""
        out = g.search_claims(_Client(), [_subject("rs_x", quote="가" * 9000)],
                              model="m", era="e", region="r", batch_size=1,
                              max_prompt_bytes=100, hard_prompt_bytes=500)
        assert out["batches"][0]["requested"] == ["rs_x"]

    def test_a_hard_smaller_than_soft_is_refused(self):
        """★hard 가 soft 보다 작으면 singleton 갈래가 통째로 죽는다."""
        with pytest.raises(ValueError, match="hard"):
            g.search_claims(_Client(), [_subject("rs_1")], model="m", era="e",
                            region="r", batch_size=1, max_prompt_bytes=1000,
                            hard_prompt_bytes=999)

    def test_a_normal_run_counts_neither(self):
        """★positive control — 평범한 판에서는 둘 다 0이다."""
        out = g.search_claims(_Client(), [_subject("rs_1")], model="m",
                              era="e", region="r", batch_size=1)
        assert out["oversized_subject_count"] == 0
        assert out["not_sent_count"] == 0


class TestTheSampleIsNineNotTwelve:
    """★★「12를 9로 바꿔 같은 실험」이 **아니다** (Codex).

    9개를 batch=12 로 보내는 것은 한 묶음으로 보내는 것일 뿐, 12-subject
    결속·leakage 를 검증하지 못한다. 후보에서 빼야 나중에 「12로 쟀다」가
    안 나온다.
    """

    def test_the_need_is_nine(self):
        assert g.EXPERIMENT_SAMPLE_SIZE == 9

    def test_twelve_is_not_a_batch_candidate(self):
        assert 12 not in g.EXPERIMENT_BATCH_SIZES
        assert max(g.EXPERIMENT_BATCH_SIZES) <= g.EXPERIMENT_SAMPLE_SIZE

    def test_no_candidate_exceeds_the_sample(self):
        """★표본보다 큰 batch 는 **잴 수 없는 크기**다."""
        assert all(1 <= b <= g.EXPERIMENT_SAMPLE_SIZE
                   for b in g.EXPERIMENT_BATCH_SIZES)

    def test_the_logical_call_count_is_what_we_will_buy(self):
        """★살 것을 미리 센다 — 9+5+3+2 = 19."""
        import math

        assert sum(math.ceil(g.EXPERIMENT_SAMPLE_SIZE / b)
                   for b in g.EXPERIMENT_BATCH_SIZES) == 19

    def test_the_tool_does_not_restate_the_contract(self):
        """★★도구가 값을 다시 적으면 **두 곳이 갈린다**. 앱 모듈이 정본이다."""
        import importlib.util
        import pathlib as _pl

        src = (_pl.Path(__file__).resolve().parents[2] / "tools"
               / "prompt_measure" / "grounding_a0_manifest.py"
               ).read_text(encoding="utf-8")
        assert "EXPERIMENT_SAMPLE_SIZE as NEED" in src
        assert "NEED = 9" not in src and "BATCH_CANDIDATES = (" not in src


class TestReuseIsJudgedByInputNotByExpectedOutput:
    """★★재사용은 「같은 결과가 나올 것」이 아니라 **입력이 그대로인가**로
    판단한다 (Codex).

    글자 수만 보면 **같은 길이의 다른 원고**를 통과시킨다 — 그러면 그 산출은
    다른 판의 것인데 이 판의 것으로 쓰인다.
    """

    def _fx(self):
        import pathlib as _pl

        return json.loads((
            _pl.Path(__file__).resolve().parents[1] / "fixtures" / "grounding"
            / "a0_recovered_97375a4b.json").read_text(encoding="utf-8"))

    def test_the_fixture_pins_both_inputs_by_hash(self):
        fx = self._fx()
        for k in ("manuscript_sha256", "entity_merge_sha256"):
            assert len(fx.get(k) or "") == 64, f"{k} 가 없거나 hash 가 아니다"

    @pytest.mark.parametrize("what", ["manuscript", "entity_merge"])
    def test_a_changed_input_actually_stops_the_reuse(self, what, monkeypatch,
                                                     tmp_path):
        """★★**동작으로 잰다.** 소스에 낱말이 있는지 보는 시험은 「길이 비교로
        돌아갔다」를 못 잡는다 (Codex).

        ★원고는 **같은 길이의 다른 글**로 바꾼다 — 길이만 보면 통과한다.
        """
        import json as _json

        m = _a0_manifest()
        fx = self._fx()
        real_text = "가" * fx["manuscript_chars"]   # 길이는 같고 내용은 다르다
        ep = tmp_path / "p" / "checkpoints" / "episodes" / "e"
        (ep / "entity_merge").mkdir(parents=True)
        (ep / "entity_merge" / "manifest.json").write_text(_json.dumps(
            {"status": "completed", "data": fx["entity_merge"]}),
            encoding="utf-8")
        monkeypatch.setattr(m, "_episode_dir", lambda root, p_, e_: ep)
        monkeypatch.setattr(m, "_rules", lambda e_: {"era": "e", "region": "r"})
        if what == "manuscript":
            # ★원고만 바꾼다 — entity_merge 는 그대로
            monkeypatch.setattr(m, "_cleaned_text", lambda e_: real_text)
        else:
            # ★entity_merge 만 바꾼다 — 원고는 그대로
            monkeypatch.setattr(m, "_cleaned_text", lambda e_: self._real_text())
            (ep / "entity_merge" / "manifest.json").write_text(_json.dumps(
                {"status": "completed",
                 "data": {"characters": [], "locations": [], "props": []}}),
                encoding="utf-8")
        with pytest.raises(SystemExit, match="입력이 바뀌었다"):
            m.from_recovered(tmp_path)

    def _real_text(self):
        """fixture 를 만든 그 원고. ★없으면 entity_merge 축을 못 잰다."""
        import glob
        import json as _json
        import pathlib as _pl

        d = glob.glob(str(_pl.Path(__file__).resolve().parents[3]
                          / "projects" / "da049582*" / "checkpoints"
                          / "episodes" / "97375a4b*"))
        if not d:
            pytest.skip("원고 체크포인트가 없다")
        return _json.loads((_pl.Path(d[0]) / "text_cleanup" / "manifest.json")
                           .read_text(encoding="utf-8"))["data"]["cleaned_text"]

    def test_the_recovered_candidates_are_never_silently_dropped(self,
                                                                 monkeypatch,
                                                                 tmp_path):
        """★★인용이 원문에 없으면 **버리지 말고 선다**.

        「23개를 그대로 재사용」하는 경로라, 하나라도 검증에 실패하면 표본이
        줄어든 채 「9개로 쟀다」가 된다.
        """
        import json as _json

        m = _a0_manifest()
        fx = self._fx()
        text = self._real_text()
        # ★후보 하나의 인용만 원문에 없게 만든다
        broken = _json.loads(_json.dumps(fx))
        broken["a0_candidates"][0]["source_quote"] = "원문에 없는 문장"
        ep = tmp_path / "p" / "checkpoints" / "episodes" / "e"
        (ep / "entity_merge").mkdir(parents=True)
        (ep / "entity_merge" / "manifest.json").write_text(_json.dumps(
            {"status": "completed", "data": fx["entity_merge"]}),
            encoding="utf-8")
        monkeypatch.setattr(m, "_episode_dir", lambda root, p_, e_: ep)
        monkeypatch.setattr(m, "_rules", lambda e_: {"era": "e", "region": "r"})
        monkeypatch.setattr(m, "_cleaned_text", lambda e_: text)
        f = tmp_path / "fx.json"
        f.write_text(_json.dumps(broken, ensure_ascii=False), encoding="utf-8")
        monkeypatch.setattr(m, "RECOVERED_FIXTURE", str(f))
        with pytest.raises(SystemExit, match="인용이 원문에 없다"):
            m.from_recovered(tmp_path)

    def test_the_reused_pack_coordinate_comes_from_the_fixture(self):
        """★★재사용인데 **지금 코드의 팩 버전**을 적으면, 팩이 바뀐 날
        「이 후보는 새 팩이 낸 것」이라는 거짓말이 된다 (Codex).

        오늘은 우연히 둘이 같아서 안 보인다 — 그래서 **바꿔서** 잰다.
        """
        m = _a0_manifest()
        import app.modules.pipeline.grounding_a0 as a0

        picked = {"coord": "x/y", "why": "w", "project_id": "p",
                  "episode_id": "e", "era": "1983", "region": "한국",
                  "a0": {"candidates": []},
                  "episode_dir": "/none",
                  "a0_reused_from": {"pack_version": "9.999",
                                     "model": "옛모델"}}
        # ★현재 팩을 다른 값으로 바꿔도 재사용 좌표는 **fixture 것**이어야 한다
        real = a0.PROMPT_PACK_VERSION
        try:
            a0.PROMPT_PACK_VERSION = "0.000"
            got = {"a0_pack_version": (picked.get("a0_reused_from") or {}).get(
                "pack_version") or a0.PROMPT_PACK_VERSION,
                "builder_pack_version": a0.PROMPT_PACK_VERSION}
        finally:
            a0.PROMPT_PACK_VERSION = real
        assert got["a0_pack_version"] == "9.999"
        assert got["builder_pack_version"] == "0.000"
        import pathlib as _pl
        src = (_pl.Path(__file__).resolve().parents[2] / "tools"
               / "prompt_measure" / "grounding_a0_manifest.py"
               ).read_text(encoding="utf-8")
        assert "builder_pack_version" in src

    def test_the_fixture_is_the_approved_episode(self):
        """★★fixture 가 **승인된 대상의 것**이어야 한다. 다른 에피소드의
        산출로 이 판의 manifest 를 만들면 표본이 통째로 딴것이 된다."""
        m = _a0_manifest()
        proj, epi, _why = m.EPISODE_ORDER[0]
        assert self._fx()["episode"] == f"{proj}/{epi}"

    def test_the_fixture_says_the_reuse_rule(self):
        """★규칙이 파일 안에 없으면 다음 사람이 왜 hash 인지 모른다."""
        assert "입력이 그대로인가" in (self._fx().get("reuse_guard") or "")


class TestTheLockedManifestSaysWhatWeWillBuy:
    """★★유료 호출 **전에** 잠근 표본. 결과를 보고 표본을 고른 것과 구분되려면
    무엇을 언제 정했는지가 남아야 한다 (Codex).
    """

    _M = (pathlib.Path(__file__).resolve().parents[1] / "fixtures"
          / "grounding" / "claims_search_manifest.json")

    @pytest.fixture
    def man(self):
        return json.loads(self._M.read_text(encoding="utf-8"))

    def test_the_lock_covers_more_than_subjects(self, man):
        """★★subjects 만 잠그면 era·검색 모델·팩·상한이 바뀌어도 「그때 그
        실험」으로 읽힌다 (Codex)."""
        lock = man["locked"]
        for k in ("era", "region", "search_model", "search_pack_version",
                  "search_pack_manifest_hash", "max_prompt_bytes",
                  "hard_prompt_bytes", "batch_candidates", "text_only_search"):
            assert k in lock, f"{k} 를 안 잠갔다"

    def test_the_search_model_is_its_own_axis(self):
        """★★A0 모델에서 **빌리지 않는다** — 빌리면 A0 를 바꾼 날 검색 모델이
        따라 움직인다 (Codex)."""
        m = _a0_manifest()
        src = pathlib.Path(m.__file__ or "").read_text(encoding="utf-8") \
            if getattr(m, "__file__", None) else ""
        assert m.SEARCH_MODEL
        assert 'doc.get("a0_model")' not in (
            pathlib.Path(__file__).resolve().parents[2] / "tools"
            / "prompt_measure" / "grounding_batch_experiment.py"
        ).read_text(encoding="utf-8"), "검색 모델을 A0 것에서 빌린다"

    def test_it_matches_the_module_contract(self, man):
        """★도구가 낸 수와 모듈 계약이 갈리면 둘 중 하나가 거짓이다."""
        assert man["count"] == g.EXPERIMENT_SAMPLE_SIZE == 9
        assert tuple(man["batch_candidates"]) == g.EXPERIMENT_BATCH_SIZES
        assert man["logical_calls"] == 19

    def test_every_subject_is_manuscript_backed(self, man):
        """★★이 관문의 전부다 — 상상 묘사로 조사하지 않는다."""
        assert len(man["locked"]["subjects"]) == 9
        for s in man["locked"]["subjects"]:
            assert s["quote_source"] == "manuscript"
            assert s["source_quote"].strip()
            assert s["research_subject_id"] and s["subject_payload_hash"]

    def test_the_subject_ids_are_unique(self, man):
        ids = [s["research_subject_id"] for s in man["locked"]["subjects"]]
        assert len(ids) == len(set(ids)) == 9

    def test_it_says_the_a0_was_reused_not_bought(self, man):
        """★「그때 A0 를 샀나」를 못 되짚으면 표본 출처가 흐려진다."""
        r = man["a0_reused_from"]
        assert r["opik_trace"] and r["ran_at"] == "2026-08-30"
        assert man["a0_pack_version"] == r["pack_version"]

    def test_it_keeps_the_caveats(self, man):
        """★한계를 안 적으면 나중에 「실제 에피소드로 쟀다」가 된다."""
        assert "검증 원고" in man["sample_caveat"]
        assert "12" in man["why_not_twelve"]

    def test_the_content_hash_matches_everything_locked(self, man):
        """★★hash 가 **잠근 것 전부**를 덮어야 한다 — subjects 만 덮으면
        era 나 검색 모델이 바뀌어도 같은 hash 다."""
        import hashlib as _h

        body = json.dumps(man["locked"], ensure_ascii=False, sort_keys=True)
        assert _h.sha256(body.encode()).hexdigest()[:16] == man["content_hash"]

    def test_changing_any_locked_value_changes_the_hash(self, man):
        """★positive control — 하나만 바꿔도 hash 가 움직인다."""
        import hashlib as _h

        for k in ("era", "search_model", "max_prompt_bytes"):
            other = {**man["locked"], k: "딴값"}
            body = json.dumps(other, ensure_ascii=False, sort_keys=True)
            assert _h.sha256(body.encode()).hexdigest()[:16] != \
                man["content_hash"], f"{k} 를 바꿔도 hash 가 그대로다"


class TestTheCallCountIsCountedBeforeItIsSpent:
    """★★`ceil(len/batch)` 로는 모자란다 — 크기 상한에 걸리면 batch 가 더
    쪼개져 **계획보다 많이 나간다**.

    가드가 계획된 수만 보면 그 초과분을 못 막고, **이미 쓴 뒤에야** 다음
    회차에서 걸린다. 유료 직전 관문이라 여기서 막는다.
    """

    def test_a_plain_run_matches_the_ceiling(self):
        import math

        subs = [_subject(f"rs_{i}") for i in range(9)]
        for b in (1, 2, 4, 8):
            assert g.plan_calls(subs, b, era="e", region="r") == \
                math.ceil(9 / b)

    def test_a_size_split_shows_up_before_sending(self):
        """★★쪼개지는 판에서 **미리** 더 큰 수가 나와야 한다."""
        import math

        subs = [_subject(f"rs_{i}", quote="가" * 3000) for i in range(8)]
        got = g.plan_calls(subs, 8, era="e", region="r",
                           max_prompt_bytes=6000)
        assert got > math.ceil(8 / 8), "쪼개졌는데 계획 수가 그대로다"

    def test_the_counted_number_is_what_actually_goes_out(self):
        """★★끝점 — 센 것과 **실제로 나간 것**이 같아야 한다.

        다르면 둘 중 하나가 거짓이고, 가드는 뜻을 잃는다.
        """
        subs = [_subject(f"rs_{i}", quote="가" * 3000) for i in range(8)]
        for b, cap in ((8, 6000), (4, 6000), (2, 100_000), (1, 100_000)):
            counted = g.plan_calls(subs, b, era="e", region="r",
                                   max_prompt_bytes=cap)
            c = _Client()
            out = g.search_claims(c, subs, model="m", era="e", region="r",
                                  batch_size=b, max_prompt_bytes=cap)
            assert counted == out["logical_calls"] == len(c.calls), \
                f"batch {b}: 센 것 {counted} vs 나간 것 {out['logical_calls']}"

    def test_counting_is_free(self):
        """★세는 것 자체가 provider 를 부르면 「보내기 전에」가 거짓이 된다."""
        c = _Client()
        g.plan_calls([_subject("rs_1")], 1, era="e", region="r")
        assert c.calls == []


class TestLogicalCallsAreNotPhysicalTransmissions:
    """★★상한 19는 **논리**다. SDK 재시도와 키 슬롯 failover 는 그 아래에서
    더 보낸다 (Codex) — 「19회 샀다」로 쓰면 거짓이 된다.

    ★slot failover 는 client 안에서 돌아 도구가 **못 센다.** 그러니 「샀다」가
    아니라 **상한**으로 적는다.
    """

    def _src(self):
        import pathlib as _pl

        return (_pl.Path(__file__).resolve().parents[2] / "tools"
                / "prompt_measure" / "grounding_batch_experiment.py"
                ).read_text(encoding="utf-8")

    def test_the_sdk_retries_are_turned_off(self):
        """★재시도를 켜 두면 실패 하나가 조용히 두 번 산다.

        ★수는 이제 **정본 상수**에서 온다 — 값과 그 값을 쓰는 자리를 따로 본다.
        """
        assert g.SDK_RETRIES == 0
        assert "openai_client(max_retries=gcs.SDK_RETRIES)" in self._src()

    def test_the_physical_bound_is_reported_not_the_logical_count(self):
        src = self._src()
        assert "approved_physical_cap" in src
        # ★상한은 **신규 수 × 슬롯**이다 — 논리 19 로 적으면 안 살 것까지 센다
        assert "신규 {pre['new']} × 슬롯 {slots}" in src
        # ★승인 상한과 이번 주행 상한을 **갈라 쓴다**
        assert '"approved_physical_cap"' in src
        assert '"this_run_physical_upper_bound"' in src

    def test_the_result_says_it_could_not_count_physical(self):
        """★못 센 것을 「샀다」로 적으면 그 수가 거짓이 된다."""
        assert "못 셌다" in self._src()


class TestTheSchemaActuallyGoesOut:
    """★★팩에서 읽어 `payload_hash` 에만 넣고 **요청에는 안 실었다** (Codex).

    모델은 산문 지시만 받고, 돌아온 모양이 계약과 달라도 아무도 안 막는다.
    「schema 를 썼다」가 거짓이었다.
    """

    def test_the_request_carries_the_json_schema(self):
        c = _Client()
        g.search_claims(c, [_subject("rs_1")], model="m", era="e", region="r",
                        batch_size=1)
        fmt = c.calls[0]["text"]["format"]
        assert fmt["type"] == "json_schema"
        assert "results" in fmt["schema"]["properties"]

    def test_it_is_the_packs_schema_not_a_handwritten_one(self):
        """★손으로 다시 적으면 팩을 고쳐도 안 따라온다."""
        c = _Client()
        g.search_claims(c, [_subject("rs_1")], model="m", era="e", region="r",
                        batch_size=1)
        assert c.calls[0]["text"]["format"]["schema"] == \
            g.load_pack()["stems"][g.SCHEMA_STEM]["content"]

    def test_a_broken_schema_stops_before_sending(self):
        """★모양이 아니면 **선다** — 산문만 보내고 「schema 를 썼다」로 쓰면
        그 말이 거짓이 된다."""
        c = _Client()
        with patch.object(g, "load_pack", return_value={
                "version": "x", "pack_manifest_hash": "h",
                "stems": {g.SYSTEM_STEM: {"content": "sys", "locator": "l",
                                          "raw_content_hash": "a"},
                          g.SCHEMA_STEM: {"content": "schema 가 아니다",
                                          "locator": "l2",
                                          "raw_content_hash": "b"}}}):
            with pytest.raises(ValueError, match="JSON schema"):
                g.search_claims(c, [_subject("rs_1")], model="m", era="e",
                                region="r", batch_size=1)
        assert c.calls == [], "선다고 해 놓고 보냈다"


class TestEachCallIsHandedOverAsSoonAsItFinishes:
    """★★9회 중 7회째에 끊기면 앞의 7회를 **통째로 잃고 다시 사야** 했다."""

    def test_the_callback_fires_per_call_not_at_the_end(self):
        seen = []
        subs = [_subject(f"rs_{i}") for i in range(4)]
        g.search_claims(_Client(), subs, model="m", era="e", region="r",
                        batch_size=1, on_batch=lambda r: seen.append(
                            list(r["requested"])))
        assert seen == [["rs_0"], ["rs_1"], ["rs_2"], ["rs_3"]]

    def test_a_broken_call_is_handed_over_too(self):
        """★깨진 것도 넘긴다 — 안 넘기면 무엇이 실패했는지 밖에서 모른다."""
        seen = []
        subs = [_subject(f"rs_{i}") for i in range(3)]
        g.search_claims(_Client(fail_on=(1,)), subs, model="m", era="e",
                        region="r", batch_size=1, on_batch=seen.append)
        assert len(seen) == 3 and seen[1]["error"]

    def test_the_handed_rows_are_the_same_objects_as_the_result(self):
        """★따로 만들면 넘긴 것과 돌려준 것이 갈린다."""
        seen = []
        out = g.search_claims(_Client(), [_subject("rs_1")], model="m",
                              era="e", region="r", batch_size=1,
                              on_batch=seen.append)
        assert seen == out["batches"]


class TestARestartDoesNotReBuy:
    """★★**보존과 재개는 다르다** (Codex).

    끊긴 주행의 줄이 남아 있어도, 다시 돌릴 때 그걸 안 보면 **처음부터 다시
    산다**. 유료 실험에서 그건 그대로 돈이다.
    """

    def _rows(self, n=4):
        return [_subject(f"rs_{i}") for i in range(n)]

    def test_a_finished_call_is_not_bought_again(self):
        subs = self._rows()
        first = g.search_claims(_Client(), subs, model="m", era="e",
                                region="r", batch_size=2)
        c = _Client()
        again = g.search_claims(c, subs, model="m", era="e", region="r",
                                batch_size=2, already=first["batches"])
        assert c.calls == [], "이미 산 것을 또 샀다"
        assert again["bought_calls"] == 0 and again["reused_calls"] == 2

    def test_a_half_finished_run_buys_only_the_rest(self):
        """★★7회째에 끊긴 판 — 앞의 것은 재사용하고 **나머지만** 산다."""
        subs = self._rows(4)
        first = g.search_claims(_Client(), subs, model="m", era="e",
                                region="r", batch_size=1)
        c = _Client()
        again = g.search_claims(c, subs, model="m", era="e", region="r",
                                batch_size=1, already=first["batches"][:2])
        assert len(c.calls) == 2
        assert again["bought_calls"] == 2 and again["reused_calls"] == 2
        # ★재사용한 것도 결과에 **그대로 들어간다** — 빠지면 표본이 준다
        asked = [s for b in again["batches"] for s in b["requested"]]
        assert sorted(asked) == sorted(s["research_subject_id"] for s in subs)

    def test_a_broken_row_is_not_reused(self):
        """★깨진 것은 **안 산 것과 같다** — 재사용하면 실패가 결과로 굳는다."""
        subs = self._rows(2)
        broken = g.search_claims(_Client(fail_on=(0, 1)), subs, model="m",
                                 era="e", region="r", batch_size=1)
        c = _Client()
        again = g.search_claims(c, subs, model="m", era="e", region="r",
                                batch_size=1, already=broken["batches"])
        assert len(c.calls) == 2 and again["reused_calls"] == 0

    def test_a_different_batch_size_does_not_reuse_across(self):
        """★batch 를 바꾸면 **호출에 넣은 목록이 다르다** — 남의 답을 못 쓴다."""
        subs = self._rows(4)
        two = g.search_claims(_Client(), subs, model="m", era="e", region="r",
                              batch_size=2)
        c = _Client()
        four = g.search_claims(c, subs, model="m", era="e", region="r",
                               batch_size=4, already=two["batches"])
        assert four["reused_calls"] == 0 and len(c.calls) == 1

    @pytest.mark.parametrize("changed", [
        {"model": "다른모델"}, {"era": "2026년"}, {"region": "미국"}])
    def test_a_different_research_is_not_reused(self, changed):
        """★★★키가 subject id 뿐이면 **다른 조사를 옛 답으로** 쓴다.

        실측 반례(Codex): `model-A/1983/KR` 뒤 같은 id 로
        `model-B/2026/US` 를 돌리면 새 호출 0 · 재사용 1 이었고, 결과
        provenance 의 모델은 **model-A** 였다.
        """
        subs = self._rows(2)
        base = dict(model="m", era="1983년", region="한국")
        first = g.search_claims(_Client(), subs, batch_size=2, **base)
        c = _Client()
        again = g.search_claims(c, subs, batch_size=2,
                                **{**base, **changed}, already=first["batches"])
        assert again["reused_calls"] == 0, f"{changed} 인데 옛 답을 썼다"
        assert len(c.calls) == 1

    def test_a_reused_row_is_not_pushed_to_the_journal_again(self):
        """★★재사용 행을 `on_batch` 에 보내면 **매 재시작마다 같은 줄이 또
        쌓인다** (Codex)."""
        subs = self._rows(2)
        first = g.search_claims(_Client(), subs, model="m", era="e",
                                region="r", batch_size=2)
        seen = []
        again = g.search_claims(_Client(), subs, model="m", era="e",
                                region="r", batch_size=2,
                                already=first["batches"], on_batch=seen.append)
        assert again["reused_calls"] == 1
        assert seen == [], "재사용 행을 journal 에 또 썼다"

    def test_the_reused_row_is_marked(self):
        """★표를 안 달면 나중에 「이건 언제 산 것인가」를 못 되짚는다."""
        subs = self._rows(2)
        first = g.search_claims(_Client(), subs, model="m", era="e",
                                region="r", batch_size=2)
        again = g.search_claims(_Client(), subs, model="m", era="e",
                                region="r", batch_size=2,
                                already=first["batches"])
        assert again["batches"][0]["reused_from_earlier_run"] is True

    def test_bought_and_reused_are_reported_separately(self):
        """★합쳐 두면 「19회 샀다」가 거짓이 된다."""
        subs = self._rows(2)
        out = g.search_claims(_Client(), subs, model="m", era="e", region="r",
                              batch_size=1)
        assert out["bought_calls"] == 2 and out["reused_calls"] == 0
        assert out["logical_calls"] == out["bought_calls"] + out["reused_calls"]


class TestTheRealSdkShapeIsWhatWeRead:
    """★★내 fake 가 **지어낸 모양**이었다 (Codex).

    설치된 stable SDK(`openai 1.109`)의 `ResponseFunctionWebSearch` 는 필드가
    `id·action·status·type` 뿐이고 **`results` 가 없다**. `action.sources` 는
    있는데 원소가 `{type, url}` — **snippet 이 없다**. 그런데 내 시험 fixture 는
    임의로 `results` 를 붙인 것이라 **프로덕션에서 [] 가 나올 것**을 못 봤다.
    """

    def _real(self, sources=("https://a.org",)):
        from openai.types.responses.response_function_web_search import (
            ActionSearch, ActionSearchSource, ResponseFunctionWebSearch)

        return ResponseFunctionWebSearch(
            id="w1", status="completed", type="web_search_call",
            action=ActionSearch(
                type="search", query="요금통",
                sources=[ActionSearchSource(type="url", url=u)
                         for u in sources]))

    def test_the_stable_shape_is_read_at_all(self):
        """★전에는 `results` 만 읽어 **실제 SDK 객체에서 0건**이었다."""
        got = g._collect_sources(SimpleNamespace(output=[self._real()]))
        assert [s["url"] for s in got] == ["https://a.org"]
        assert got[0]["shape"] == "action.sources"

    def test_the_stable_shape_has_no_snippet_but_still_passes(self):
        """★★본문은 안 온다. 그래도 **주소가 오면 된다** — 인용 확인은
        `grounding_claims_support` 가 그 URL 을 직접 열어서 한다.

        ★전에는 snippet 을 요구해 첫 호출에서 늘 섰고, 그러면 HTTP 확인까지
        **갈 수가 없었다** (Codex).
        """
        got = g._collect_sources(SimpleNamespace(output=[self._real()]))
        assert got[0]["snippet"] == ""
        shape = g.capture_shape({"sources": got, "response_dump": _DUMP})
        assert shape["ok"] is True and shape["urls"] == 1
        assert shape["with_snippet"] == 0

    def test_the_beta_shape_still_works(self):
        """★API 가 `results` 를 extra 로 얹어 주면 그대로 읽는다."""
        got = g._collect_sources(_resp(["rs_1"]))
        assert got[0]["snippet"] == "본문 조각"
        assert g.capture_shape({"sources": got, "response_dump": _DUMP})["ok"] is True

    def test_both_shapes_do_not_double_count_one_url(self):
        """★같은 주소가 두 모양으로 오면 **한 번만** 센다."""
        beta = _resp(["rs_1"], sources=(("q", "https://a.org", "t", "조각"),))
        beta.output.insert(0, self._real(("https://a.org",)))
        got = g._collect_sources(beta)
        assert [s["url"] for s in got].count("https://a.org") == 1

    def test_the_probe_says_why_it_is_not_ok(self):
        """★「안 된다」만으로는 무엇을 고칠지 모른다."""
        shape = g.capture_shape({"sources": [{"url": "", "snippet": ""}]})
        assert shape["ok"] is False and "주소" in shape["why"]

    def test_a_citation_alone_is_enough(self):
        """★`action.sources` 가 비어도 `url_citation` 이 있으면 확인할 수 있다."""
        shape = g.capture_shape({"response_dump": _DUMP, "sources": [],
                                 "citations": ["https://a.org"]})
        assert shape["ok"] is True


class TestTheRunnerBuildsTheOwnershipBaselineFirst:
    """★★`batch=1` 이 **먼저** 돌아야 기준선이 생긴다 — 그 뒤에야 2/4/8 을
    소유권까지 재서 채점할 수 있다 (Codex).
    """

    def _src(self):
        return (pathlib.Path(__file__).resolve().parents[2] / "tools"
                / "prompt_measure" / "grounding_batch_experiment.py"
                ).read_text(encoding="utf-8")

    def test_the_sizes_run_in_ascending_order(self):
        src = self._src()
        assert "order = sorted(gcs.EXPERIMENT_BATCH_SIZES)" in src
        assert 'raise SystemExit("batch 1 이 없다' in src

    def test_the_baseline_is_passed_to_scoring(self):
        assert "baseline=baseline" in self._src()

    def test_one_is_in_the_contract(self):
        """★1이 빠지면 기준선을 만들 수 없다 — 계약이 그것을 담아야 한다."""
        assert 1 in g.EXPERIMENT_BATCH_SIZES
        assert min(g.EXPERIMENT_BATCH_SIZES) == 1


class TestAPriorCaptureFailureStopsTheRun:
    """★★★snippet 이 없으면 `evidence_span` 을 확인할 수 없어 그 호출은
    쓸모가 없다. 그런 줄이 앞 주행에 있으면 **새 호출을 한 번도 사지 않는다.**

    ★재사용도 안 하고 재구매도 안 한다 — 둘 다 틀렸다. 재사용하면 못 쓰는 답이
    결과로 굳고, 재구매하면 매 재실행마다 한 호출씩 더 산다.
    """

    @pytest.mark.parametrize("srcs", [[]])
    def test_a_prior_capture_failure_stops_with_zero_new_calls(self, srcs):
        """★★★「안 산 것」으로 치고 넘어가면 **재실행마다 한 호출씩 더 산다**
        — 이미 확정된 호환 불가를 매번 다시 사는 것이다 (Codex).

        URL 만 있고 snippet 이 없는 것도 못 쓴다 — 그게 실제 stable SDK 모양이다.
        """
        subs = [_subject("rs_1"), _subject("rs_2")]
        prior = {"requested": ["rs_1"], "error": "", "not_sent": False,
                 "sources": srcs, "parsed": {"results": []},
                 "provenance": {"payload_hash": _ph(subs[:1]), "model": "m"}}
        c = _Client()
        with pytest.raises(ValueError, match="다시 사지 않는다"):
            g.search_claims(c, subs, model="m", era="e", region="r",
                            batch_size=1, already=[prior])
        assert c.calls == [], "선다고 해 놓고 샀다"

    def test_a_good_row_is_still_reused(self):
        """★positive control — 포착이 된 것은 그대로 재사용한다."""
        subs = [_subject("rs_1")]
        first = g.search_claims(_Client(), subs, model="m", era="e",
                                region="r", batch_size=1)
        c = _Client()
        out = g.search_claims(c, subs, model="m", era="e", region="r",
                              batch_size=1, already=first["batches"])
        assert out["reused_calls"] == 1 and c.calls == []


class TestTheRunnerReadsOnlyTheLockedValues:
    """★★잠근 사본과 쓰는 사본이 **두 벌**이면 content hash 가 뜻을 잃는다."""

    def _src(self):
        return (pathlib.Path(__file__).resolve().parents[2] / "tools"
                / "prompt_measure" / "grounding_batch_experiment.py"
                ).read_text(encoding="utf-8")

    @pytest.mark.parametrize("key", ["era", "region", "search_model"])
    def test_the_run_inputs_come_from_lock(self, key):
        src = self._src()
        assert f'doc["{key}"]' not in src, f"{key} 를 top-level 에서 읽는다"
        assert f'lock["{key}"]' in src


class TestTheManifestHasOneEnvelopeOnly:
    """★★★같은 값을 top-level 에도 두면 잠근 사본과 쓰는 사본이 **두 벌**이다.

    실측(Codex): `locked`/`content_hash` 를 그대로 두고 top-level 의
    `era=2099`·`search_model`·`subjects[0].surface_form` 만 바꿔도 검증을
    통과했고, 그 바뀐 값으로 **다른 입력을 사고도 원래 hash 실험으로 기록**됐다.
    """

    _M = (pathlib.Path(__file__).resolve().parents[1] / "fixtures"
          / "grounding" / "claims_search_manifest.json")

    @pytest.mark.parametrize("key", ["era", "region", "search_model",
                                     "subjects"])
    def test_no_copy_survives_at_top_level(self, key):
        man = json.loads(self._M.read_text(encoding="utf-8"))
        assert key not in man, f"top-level 에 {key} 사본이 있다"
        assert key in man["locked"]

    def test_the_request_meaning_is_locked_too(self):
        """★같은 표본이라도 요청이 다르면 **다른 실험**이다."""
        lock = json.loads(self._M.read_text(encoding="utf-8"))["locked"]
        for k in ("text_only_search", "schema_strict", "source_capture",
                  "include", "store", "support_policy", "probe_requirement",
                  "sdk_retries"):
            assert k in lock, f"{k} 를 안 잠갔다"
        # ★strict 를 **켰다** — 안 켠 판에서 claim 16/16 이 `required` 를
        #  빠뜨려 계약에서 전부 거부됐다(2026-08-30 실측).
        assert lock["schema_strict"] is True
        assert lock["sdk_retries"] == 0

    def test_the_locked_request_is_what_the_code_actually_sends(self):
        """★★★**거짓 잠금**을 막는다 (Codex).

        잠근 것은 `results`(beta)·`store=null`·snippet 요구였고 코드는
        `action.sources`·`store=False`·HTTP 확인이었다. 그런데도 dry-run 은
        **초록**이었다 — 그 칸들이 대조 대상이 아니었기 때문이다.
        """
        lock = json.loads(self._M.read_text(encoding="utf-8"))["locked"]
        for k, v in g.request_contract().items():
            assert lock[k] == v, f"{k} 가 지금 코드와 다르다: {lock[k]!r} != {v!r}"

    def test_the_contract_is_not_written_twice(self):
        """★같은 값을 요청문에 또 적으면 **한쪽만 고쳐진다**."""
        src = pathlib.Path(g.__file__).read_text(encoding="utf-8")
        assert 'include=list(RESPONSE_INCLUDE)' in src
        assert 'store=PROVIDER_STORE' in src
        assert 'include=["web_search_call' not in src
        assert "store=False," not in src

    @pytest.mark.parametrize("key", ["include", "store", "source_capture",
                                     "support_policy", "probe_requirement",
                                     "text_only_search", "schema_strict",
                                     "sdk_retries"])
    def test_flipping_one_locked_request_field_stops_the_runner(
            self, tmp_path, key):
        """★끝점 — 칸 **하나씩** 바꿔 실제로 서는지 본다."""
        import hashlib

        m = _batch_experiment()
        man = json.loads(self._M.read_text(encoding="utf-8"))
        cur = man["locked"][key]
        man["locked"][key] = (["다른 것"] if isinstance(cur, list)
                              else (not cur) if isinstance(cur, bool)
                              else (cur + 1) if isinstance(cur, int)
                              else f"{cur} (바뀐 것)")
        # ★hash 는 **다시 맞춰 준다** — 안 그러면 hash 검사에서 먼저 서서
        #  「요청 계약 drift 를 잡았다」가 아니라 다른 것을 재게 된다.
        body = json.dumps(man["locked"], ensure_ascii=False, sort_keys=True)
        man["content_hash"] = hashlib.sha256(body.encode()).hexdigest()[:16]
        (tmp_path / "tests" / "fixtures" / "grounding").mkdir(parents=True)
        (tmp_path / m.MANIFEST).write_text(json.dumps(man, ensure_ascii=False),
                                           encoding="utf-8")
        with pytest.raises(SystemExit, match="코드가 바뀌었다"):
            m.load_manifest(tmp_path)

    def test_the_runner_refuses_a_manifest_with_copies(self, tmp_path):
        """★끝점 — 사본이 있는 manifest 는 **거부**한다."""
        m = _batch_experiment()
        man = json.loads(self._M.read_text(encoding="utf-8"))
        man["era"] = "2099년"
        root = tmp_path
        (root / "tests" / "fixtures" / "grounding").mkdir(parents=True)
        (root / m.MANIFEST).write_text(json.dumps(man, ensure_ascii=False),
                                       encoding="utf-8")
        with pytest.raises(SystemExit, match="사본이 있다"):
            m.load_manifest(root)


class TestTheResponseIsKeptLocallyNotAtTheProvider:
    """★★`store=True` 는 **데이터 정책 변경**이다 — 원고 인용이 외부 보존
    대상이 된다. 그리고 필요도 없다: 받자마자 **로컬에** 통째로 남기면 된다
    (Codex).

    ★값을 안 넘기면 서버 기본에 맡겨진다 — 실측으로 보존이 안 돼
    `responses.retrieve` 가 404 였다. 그래서 **명시로 끈다**.
    """

    def test_the_request_pins_store_off(self):
        c = _Client()
        g.search_claims(c, [_subject("rs_1")], model="m", era="e", region="r",
                        batch_size=1)
        assert c.calls[0]["store"] is False

    def test_the_official_include_is_used(self):
        """★`results` 는 stable include 목록에 **없다**."""
        c = _Client()
        g.search_claims(c, [_subject("rs_1")], model="m", era="e", region="r",
                        batch_size=1)
        assert c.calls[0]["include"] == ["web_search_call.action.sources"]

    def test_the_whole_response_is_dumped_locally(self):
        """★★provider 보존에 기대지 않고 되짚을 수 있어야 한다."""
        out = g.search_claims(_Client(), [_subject("rs_1")], model="m",
                              era="e", region="r", batch_size=1)
        d = out["batches"][0]["response_dump"]
        assert isinstance(d, dict) and d.get("output")

    def test_a_dump_failure_does_not_kill_the_research(self):
        """★남기다 터져도 산 것은 살린다."""
        class _NoDump(SimpleNamespace):
            def model_dump(self, **kw):
                raise RuntimeError("못 남긴다")

        def _mk(ids, **kw):
            r = _resp(ids, **kw)
            return _NoDump(id=r.id, output=r.output)

        out = g.search_claims(_Client(maker=_mk), [_subject("rs_1")],
                              model="m", era="e", region="r", batch_size=1)
        assert out["batches"][0]["response_dump"]["_dump_failed"] is True
        assert out["batches"][0]["parsed"] is not None


class TestOfficialCitationsAreCollectedForOwnershipOnly:
    """★공식 경로로 「모델이 무엇을 인용했다고 표시했나」를 받는다.
    본문은 안 오므로 `evidence_span` 확인에는 **못 쓴다** — 별도 축이다."""

    def _with_citation(self, ids, **kw):
        r = _resp(ids, **kw)
        msg = [o for o in r.output if o.type == "message"][0]
        msg.content[0].annotations = [
            SimpleNamespace(type="url_citation", url="https://cited.org"),
            SimpleNamespace(type="file_citation", url="https://무시.org")]
        return r

    def test_url_citations_are_kept(self):
        out = g.search_claims(_Client(maker=self._with_citation),
                              [_subject("rs_1")], model="m", era="e",
                              region="r", batch_size=1)
        assert out["batches"][0]["citations"] == ["https://cited.org"]

    def test_a_response_without_annotations_gives_an_empty_list(self):
        out = g.search_claims(_Client(), [_subject("rs_1")], model="m",
                              era="e", region="r", batch_size=1)
        assert out["batches"][0]["citations"] == []


class TestTheReaderDoesNotGuessKeyNames:
    """★★실제 키 이름을 모르는 채 여러 이름을 순서대로 보는 것은 **추측**이다.

    그걸 넣고 사면 「안 왔다」와 「이름이 달랐다」가 **안 갈린다** (Codex 금지).
    모양이 확정될 때까지 `snippet` 하나만 보고, 못 읽으면 세운다.
    """

    def test_only_the_declared_key_is_read(self):
        src = (pathlib.Path(__file__).resolve().parents[2] / "app" / "modules"
               / "pipeline" / "grounding_claims_search.py"
               ).read_text(encoding="utf-8")
        assert '_get(r, "text")' not in src
        assert '_get(r, "content")' not in src

    def test_an_unknown_key_shape_leaves_the_snippet_empty(self):
        """★이름이 다르면 **못 읽었다고 말한다** — 조용히 채우지 않는다.

        ★다만 주소는 읽혔으므로 실험은 계속된다. 본문 확인은 HTTP 가 한다.
        """
        resp = SimpleNamespace(output=[SimpleNamespace(
            type="web_search_call",
            action=SimpleNamespace(query="q", sources=[]),
            results=[{"url": "https://a.org", "딴이름": "본문"}])])
        got = g._collect_sources(resp)
        assert got[0]["snippet"] == ""
        shape = g.capture_shape({"sources": got, "response_dump": _DUMP})
        assert shape["ok"] is True and shape["with_snippet"] == 0

    def test_the_shape_probe_records_the_unknown_keys(self):
        """★★그리고 **그 이름이 무엇이었는지**를 남긴다 — 다음 판이 안다."""
        resp = SimpleNamespace(output=[SimpleNamespace(
            type="web_search_call",
            action=SimpleNamespace(query="q", sources=[]),
            results=[{"url": "https://a.org", "딴이름": "본문"}])])
        probe = g.raw_source_shape(resp)
        assert probe and "딴이름" in probe[0]["keys"]
        assert probe[0]["sample"]["딴이름"] == "본문"


class TestAnOldFailureDoesNotBlockANewPack:
    """★★★옛 팩·다른 모델의 실패 기록이 **새 판을 막으면 안 된다**.

    실측(Codex): 팩 1 의 포착 실패 한 줄이 팩 3 주행을 **0회에서** 막았다.
    기록은 보존하되, 판이 다르면 **남의 일**이다.
    """

    def _prior(self, *, payload="옛payload", model="m", sources=()):
        return {"requested": ["rs_1"], "error": "", "not_sent": False,
                "sources": list(sources), "parsed": {"results": []},
                "provenance": {"payload_hash": payload, "model": model}}

    def test_a_different_payload_failure_is_ignored(self):
        """★새 팩은 payload 가 다르다 — 그 실패는 이 판과 무관하다."""
        c = _Client()
        out = g.search_claims(c, [_subject("rs_1")], model="m", era="e",
                              region="r", batch_size=1,
                              already=[self._prior(payload="딴판")])
        assert len(c.calls) == 1 and out["reused_calls"] == 0

    def test_a_different_model_failure_is_ignored(self):
        c = _Client()
        g.search_claims(c, [_subject("rs_1")], model="m", era="e", region="r",
                        batch_size=1,
                        already=[self._prior(model="다른모델")])
        assert len(c.calls) == 1

    def test_the_same_call_failure_still_stops(self):
        """★★같은 호출이면 **다시 안 산다** — 확정된 호환 불가다."""
        subs = [_subject("rs_1")]
        # 먼저 실제 payload_hash 를 얻는다
        first = g.search_claims(_Client(), subs, model="m", era="e",
                                region="r", batch_size=1)
        ph = first["batches"][0]["provenance"]["payload_hash"]
        c = _Client()
        with pytest.raises(ValueError, match="다시 사지 않는다"):
            g.search_claims(c, subs, model="m", era="e", region="r",
                            batch_size=1,
                            already=[self._prior(payload=ph, model="m")])
        assert c.calls == []

    def test_the_stop_names_the_payload(self):
        """★어느 호출이 막았는지 안 적으면 못 되짚는다."""
        subs = [_subject("rs_1")]
        ph = g.search_claims(_Client(), subs, model="m", era="e", region="r",
                             batch_size=1)["batches"][0]["provenance"][
                                 "payload_hash"]
        with pytest.raises(ValueError, match=ph):
            g.search_claims(_Client(), subs, model="m", era="e", region="r",
                            batch_size=1,
                            already=[self._prior(payload=ph, model="m")])


class TestTheProbeAlsoNeedsTheRawResponseKept:
    """★★★주소가 왔어도 **산 것의 원형이 없으면** 진단을 이어가지 않는다 (Codex).

    앞 판에서 유료 A0 산출을 실제로 **잃었다**. 그때는 판정이 저장보다 먼저였고,
    지금은 순서를 고쳤지만 probe 가 dump 를 안 보면 dump 가 비어도 초록이라
    **또 원형 없이** 진단이 굴러간다. dump 실패는 그 호출을 기록한 채 **선다** —
    다시 사지 않는다.
    """

    _SRC = [{"url": "https://example.org/a", "snippet": "",
             "shape": "action.sources"}]

    def test_a_url_alone_is_not_enough(self):
        shape = g.capture_shape({"sources": self._SRC})
        assert shape["ok"] is False
        assert shape["response_dump_ok"] is False
        assert "원형" in shape["why"]

    def test_a_failed_dump_stops_it(self):
        shape = g.capture_shape({"sources": self._SRC,
                                 "response_dump": {"_dump_failed": True}})
        assert shape["ok"] is False and shape["response_dump_ok"] is False

    def test_an_empty_dump_stops_it(self):
        shape = g.capture_shape({"sources": self._SRC, "response_dump": {}})
        assert shape["ok"] is False

    def test_both_together_pass(self):
        """★positive control — 둘 다 있으면 지나간다."""
        shape = g.capture_shape({"sources": self._SRC,
                                 "response_dump": {"id": "resp_1"}})
        assert shape["ok"] is True and shape["response_dump_ok"] is True

    def test_the_missing_url_reason_is_not_hidden_by_the_dump_reason(self):
        """★두 사유가 **섞이지 않는다** — 고칠 곳이 다르다."""
        shape = g.capture_shape({"sources": [], "response_dump": {"id": "x"}})
        assert shape["ok"] is False and "주소" in shape["why"]


class TestThePaidModesAreExclusive:
    """★★★`--dry-run --run` 을 같이 주면 **전에는 run 이 이겨 실제로 샀다** (Codex).

    사용자가 서로 반대되는 두 플래그를 준 사실을 숨긴 채 사는 것이 문제다.
    이제 parser 가 서고, provider 호출은 **0 회**다.
    """

    def _parser_run(self, argv):
        import subprocess
        import sys
        return subprocess.run(
            [sys.executable, "-m",
             "tools.prompt_measure.grounding_batch_experiment", *argv],
            capture_output=True, text=True,
            cwd=str(pathlib.Path(__file__).resolve().parents[2]),
            env={**__import__("os").environ,
                 "THEROAD_TEST_BLOCK_OUTBOUND": "1"})

    @pytest.mark.parametrize("argv", [
        ["--dry-run", "--run"],
        ["--run", "--diagnostic-one"],
        ["--dry-run", "--diagnostic-one"],
    ])
    def test_conflicting_modes_stop_the_parser(self, argv):
        r = self._parser_run(argv)
        assert r.returncode == 2, f"parser 가 안 섰다: {r.returncode}"
        assert "not allowed with" in r.stderr or "허용" in r.stderr
        # ★provider 를 부르기 전에 서야 한다 — 표본을 읽은 흔적도 없어야 한다
        assert "산 것" not in r.stdout

    def test_no_argument_is_free(self):
        """★기본은 **무료**다."""
        r = self._parser_run([])
        assert r.returncode == 0
        assert "--run 이다" in r.stdout


class TestTheDiagnosticBuysExactlyOneAndDoesNotFeedChoice:
    """★★★승인 범위는 **known regression 1회**뿐이다 (Codex).

    - 잠근 **첫** subject 하나만, 같은 프로덕션 `search_claims` 경로로 산다
    - 산 것은 **판정보다 먼저** 디스크에 남는다
    - 그 결과는 batch 후보·choice 에 **안 들어간다** — 크기 고르기를 부르지도
      않고, 산출에 `final_candidate`/`mechanical_candidate` 가 없다
    """

    def _args(self, tmp_path):
        return SimpleNamespace(out=tmp_path / "out",
                              root=pathlib.Path("."), diagnostic_one=True,
                              run=False, dry_run=False)

    def _load(self):
        m = _batch_experiment()
        doc = m.load_manifest(pathlib.Path(__file__).resolve().parents[2])
        lock = doc["locked"]
        return m, doc, lock, lock["subjects"], m.unique_forms(lock["subjects"])

    def test_it_buys_one_call_for_the_locked_first_subject(
            self, tmp_path, monkeypatch):
        m, doc, lock, subs, uf = self._load()
        client = _Client()
        monkeypatch.setattr("app.core.openai_keys.openai_client",
                            lambda **kw: client)
        rc = m._diagnostic_one(self._args(tmp_path), doc, lock, subs, uf)

        assert rc == 0
        assert len(client.calls) == 1, f"{len(client.calls)}회 샀다"
        sent = client.calls[0]["input"][0]["content"][0]["text"]
        assert subs[0]["research_subject_id"] in sent
        # ★다른 대상이 같이 실려 나가지 않는다
        for other in subs[1:]:
            assert other["research_subject_id"] not in sent

    def test_what_was_bought_is_on_disk_before_it_is_judged(
            self, tmp_path, monkeypatch):
        """★유료 산출을 실제로 잃은 적이 있다 — 판정이 저장보다 먼저였다."""
        m, doc, lock, subs, uf = self._load()
        monkeypatch.setattr("app.core.openai_keys.openai_client",
                            lambda **kw: _Client())
        monkeypatch.setattr(m, "score_run",
                            lambda *a, **k: (_ for _ in ()).throw(
                                RuntimeError("판정이 터졌다")))
        with pytest.raises(RuntimeError, match="판정이 터졌다"):
            m._diagnostic_one(self._args(tmp_path), doc, lock, subs, uf)

        live = tmp_path / "out" / "diagnostic_one.jsonl"
        rows = [json.loads(x) for x in
                live.read_text(encoding="utf-8").splitlines() if x.strip()]
        assert len(rows) == 1 and rows[0]["response_dump"]

    def test_the_result_carries_no_candidate_for_choice(
            self, tmp_path, monkeypatch):
        m, doc, lock, subs, uf = self._load()
        monkeypatch.setattr("app.core.openai_keys.openai_client",
                            lambda **kw: _Client())
        called = []
        monkeypatch.setattr(m, "choose_batch_size",
                            lambda *a, **k: called.append(1))
        m._diagnostic_one(self._args(tmp_path), doc, lock, subs, uf)

        got = json.loads((tmp_path / "out" / "diagnostic_one_result.json")
                         .read_text(encoding="utf-8"))
        assert called == [], "진단이 크기 고르기를 불렀다"
        assert "final_candidate" not in got and "mechanical_candidate" not in got
        assert got["mode"] == "diagnostic_one"
        assert "안 넣는다" in got["not_a_candidate"]
        assert got["bought_calls"] == 1

    def test_running_it_twice_does_not_buy_twice(self, tmp_path, monkeypatch):
        """★끊겨서 다시 돌려도 **같은 호출은 다시 안 산다**."""
        m, doc, lock, subs, uf = self._load()
        client = _Client()
        monkeypatch.setattr("app.core.openai_keys.openai_client",
                            lambda **kw: client)
        m._diagnostic_one(self._args(tmp_path), doc, lock, subs, uf)
        m._diagnostic_one(self._args(tmp_path), doc, lock, subs, uf)
        assert len(client.calls) == 1, "두 번째 주행이 또 샀다"


class TestTheRequestActuallyConsumesTheContract:
    """★★★잠근 세 칸(글만·strict·재시도)을 **호출부가 실제로 쓴다**.

    manifest 에는 있는데 코드가 안 쓰면, `want_images`·`strict`·`max_retries`
    를 바꿔도 dry-run 이 **초록**이다 — 거짓 잠금이다 (Codex).
    """

    def test_the_call_is_text_only_and_not_strict(self):
        client = _Client()
        g.search_claims(client, [_subject("rs_a", "가", "문장")],
                        model="m", era="1983년", region="대한민국",
                        batch_size=1)
        kw = client.calls[0]
        tool = kw["tools"][0]
        assert "search_content_types" not in tool, "글만인데 사진을 켰다"
        assert kw["text"]["format"]["strict"] is g.SCHEMA_STRICT
        assert kw["include"] == list(g.RESPONSE_INCLUDE)
        assert kw["store"] is g.PROVIDER_STORE

    def test_the_runner_builds_its_client_from_the_constant(self):
        src = pathlib.Path(_batch_experiment().__file__).read_text(
            encoding="utf-8")
        assert "max_retries=0" not in src, "재시도 수를 두 곳에 적었다"
        assert src.count("max_retries=gcs.SDK_RETRIES") == 2


class TestPriorReuseIsScopedToTheSameRequest:
    """★★★프롬프트·subject 가 같아도 **요청이 다르면 다른 답**이다 (Codex).

    앞 판의 payload 신원은 system+user+schema 뿐이라, `include` 하나만 바꿔도
    옛 응답을 그대로 재사용했다. manifest 를 새로 굳혀도 journal 재사용은
    manifest 를 안 본다 — 그래서 **신원 자체**에 획득 좌표를 접는다.
    """

    def _first_run(self):
        client = _Client()
        out = g.search_claims(client, [_subject("rs_a", "가", "문장")],
                              model="m", era="1983년", region="대한민국",
                              batch_size=1)
        return out["batches"]

    def test_the_same_request_reuses(self):
        """★positive control — 안 바꾸면 재사용된다."""
        prior = self._first_run()
        client = _Client()
        out = g.search_claims(client, [_subject("rs_a", "가", "문장")],
                              model="m", era="1983년", region="대한민국",
                              batch_size=1, already=prior)
        assert out["reused_calls"] == 1 and out["bought_calls"] == 0
        assert client.calls == []

    @pytest.mark.parametrize("attr,val", [
        ("RESPONSE_INCLUDE", ("web_search_call.results",)),
        ("PROVIDER_STORE", True),
        ("TEXT_ONLY_SEARCH", False),
        ("SCHEMA_STRICT", False),
    ])
    def test_changing_one_acquisition_field_buys_again(
            self, monkeypatch, attr, val):
        prior = self._first_run()
        monkeypatch.setattr(g, attr, val)
        client = _Client()
        out = g.search_claims(client, [_subject("rs_a", "가", "문장")],
                              model="m", era="1983년", region="대한민국",
                              batch_size=1, already=prior)
        assert out["reused_calls"] == 0, f"{attr} 가 바뀌었는데 옛 답을 썼다"
        assert out["bought_calls"] == 1 and len(client.calls) == 1

    def test_a_free_rescoring_policy_does_not_buy_again(self, monkeypatch):
        """★후처리 정책은 **산 것을 안 바꾼다** — 다시 사면 그게 낭비다."""
        prior = self._first_run()
        monkeypatch.setattr(g, "SUPPORT_POLICY", "다른 후처리 정책")
        client = _Client()
        out = g.search_claims(client, [_subject("rs_a", "가", "문장")],
                              model="m", era="1983년", region="대한민국",
                              batch_size=1, already=prior)
        assert out["reused_calls"] == 1 and client.calls == []


class TestTheDiagnosticFailsClosedOnAWeakProbe:
    """★★★probe 요구를 어겼는데 `0` 으로 끝내면 **「진단 완료」로 읽힌다** (Codex).

    주소가 없거나 응답 원형을 잃은 것은 **실패**다. 유료 행과 실패 결과는
    **먼저 저장**하고, 그 다음 비영으로 선다. 같은 payload 재구매는 앞의
    prior gate 가 막는다.
    """

    def _setup(self, tmp_path, monkeypatch, maker):
        m = _batch_experiment()
        doc = m.load_manifest(pathlib.Path(__file__).resolve().parents[2])
        lock = doc["locked"]
        client = _Client(maker=maker)
        _no_pages(monkeypatch, m)
        monkeypatch.setattr("app.core.openai_keys.openai_client",
                            lambda **kw: client)
        args = SimpleNamespace(out=tmp_path / "out", root=pathlib.Path("."),
                               diagnostic_one=True, run=False, dry_run=False)
        rc = m._diagnostic_one(args, doc, lock, lock["subjects"],
                               m.unique_forms(lock["subjects"]))
        got = json.loads((tmp_path / "out" / "diagnostic_one_result.json")
                         .read_text(encoding="utf-8"))
        rows = [json.loads(x) for x in
                (tmp_path / "out" / "diagnostic_one.jsonl")
                .read_text(encoding="utf-8").splitlines() if x.strip()]
        return rc, got, rows, client

    def test_no_source_url_exits_nonzero(self, tmp_path, monkeypatch):
        def _no_src(ids, *, resp_id="r"):
            return _resp(ids, resp_id=resp_id, sources=())
        rc, got, rows, _c = self._setup(tmp_path, monkeypatch, _no_src)
        assert rc != 0, "포착 실패인데 0 으로 끝났다"
        assert got["probe_ok"] is False
        assert "주소" in got["capture_shape"]["why"]
        # ★산 것은 **그래도 남는다**
        assert len(rows) == 1

    def test_a_lost_response_dump_exits_nonzero(self, tmp_path, monkeypatch):
        class _NoDump:
            def __init__(self, inner):
                self._i = inner
                self.id = inner.id
                self.output = inner.output

            def model_dump(self, **kw):
                raise RuntimeError("원형을 못 뜬다")

        rc, got, rows, _c = self._setup(
            tmp_path, monkeypatch,
            lambda ids, *, resp_id="r": _NoDump(_resp(ids, resp_id=resp_id)))
        assert rc != 0
        assert got["probe_ok"] is False
        assert got["capture_shape"]["response_dump_ok"] is False
        assert len(rows) == 1, "원형을 잃었어도 산 행은 남아야 한다"

    def test_a_good_probe_exits_zero(self, tmp_path, monkeypatch):
        """★positive control — 막기만 하면 진단이 성립하지 않는다."""
        rc, got, rows, _c = self._setup(tmp_path, monkeypatch, None)
        assert rc == 0 and got["probe_ok"] is True and len(rows) == 1

    def test_a_failed_probe_is_not_bought_again(self, tmp_path, monkeypatch):
        """★실패한 조합을 **다시 사지 않는다** — 두 번째 주행은 선다."""
        def _no_src(ids, *, resp_id="r"):
            return _resp(ids, resp_id=resp_id, sources=())
        rc, _g, _r, client = self._setup(tmp_path, monkeypatch, _no_src)
        assert rc != 0 and len(client.calls) == 1

        m = _batch_experiment()
        doc = m.load_manifest(pathlib.Path(__file__).resolve().parents[2])
        lock = doc["locked"]
        again = _Client(maker=_no_src)
        monkeypatch.setattr("app.core.openai_keys.openai_client",
                            lambda **kw: again)
        args = SimpleNamespace(out=tmp_path / "out", root=pathlib.Path("."),
                               diagnostic_one=True, run=False, dry_run=False)
        with pytest.raises(ValueError, match="다시 사지 않는다"):
            m._diagnostic_one(args, doc, lock, lock["subjects"],
                              m.unique_forms(lock["subjects"]))
        assert again.calls == [], "실패한 조합을 또 샀다"


class TestTheDiagnosticCallIsNotBoughtTwice:
    """★★★진단 1회와 full 주행이 **서로 다른 파일만** 읽으면, 재개할 때 이미 산
    팩3 첫 subject 를 **다시 산다** (Codex 중복결제 BLOCK).

    끝점: 같은 fake client 로 진단 → full 을 이어서 돌리면 결과는 명목 19개를
    다 보되 **새로 사는 것은 15**(고유 payload 16 − 이미 산 1), 재사용 4
    (진단 1 + 꼬리 singleton 의 b2·b4·b8 3), 첫 subject 의 journal·결과 중복 0.
    """

    def _run(self, tmp_path, monkeypatch):
        m = _batch_experiment()
        doc = m.load_manifest(pathlib.Path(__file__).resolve().parents[2])
        lock = doc["locked"]
        # ★기준선 멈춤에 안 걸리게 **확정된 claim 이 있는** 대역을 쓴다 —
        #  여기서 재는 것은 재사용이지 멈춤이 아니다.
        client = _Client(maker=_resp_with_claims)
        _no_pages(monkeypatch, m)
        monkeypatch.setattr("app.core.openai_keys.openai_client",
                            lambda **kw: client)
        args = SimpleNamespace(out=tmp_path / "out", root=pathlib.Path("."),
                               diagnostic_one=True, run=False, dry_run=False)
        m._diagnostic_one(args, doc, lock, lock["subjects"],
                          m.unique_forms(lock["subjects"]))
        return m, doc, lock, client, args

    def test_the_full_run_reuses_the_diagnostic_call(self, tmp_path,
                                                     monkeypatch):
        m, doc, lock, client, args = self._run(tmp_path, monkeypatch)
        assert len(client.calls) == 1
        args.diagnostic_one, args.run = False, True
        m.run_experiment(args, doc, lock, lock["subjects"],
                         m.unique_forms(lock["subjects"]))

        # ★★명목 19 논리 호출인데 **산 것은 15** 다. 진단 1회가 batch 1 의 첫
        #  호출로 재사용되고(1), 표본이 9개라 2·4·8 의 **꼬리 묶음**이 전부
        #  「9번째 대상 혼자」여서 batch 1 의 그 호출과 payload 가 같다(3).
        #  ★이 수를 안 적으면 「19회 샀다」가 거짓이 된다.
        got = json.loads((tmp_path / "out" / "result.json")
                         .read_text(encoding="utf-8"))
        assert got["logical_calls_planned"] == 19
        assert got["reused_calls_total"] == 4
        assert got["logical_calls_spent"] == 15
        assert len(client.calls) == 1 + 15, f"{len(client.calls)}회 샀다"
        assert "꼬리" in got["why_reused"]

    def test_the_first_batch_one_call_is_the_diagnostic_one(self, tmp_path,
                                                            monkeypatch):
        """★진단이 산 그 호출은 batch 1 에서 **재사용**된다 — 다시 안 산다."""
        m, doc, lock, client, args = self._run(tmp_path, monkeypatch)
        args.diagnostic_one, args.run = False, True
        m.run_experiment(args, doc, lock, lock["subjects"],
                         m.unique_forms(lock["subjects"]))
        got = json.loads((tmp_path / "out" / "result.json")
                         .read_text(encoding="utf-8"))
        assert got["by_batch"]["1"]["reused_calls"] == 1
        assert got["by_batch"]["1"]["bought_calls"] == 8

    def test_isolating_the_journals_again_would_buy_more(self, tmp_path,
                                                         monkeypatch):
        """★★positive control — 15 가 **재사용 덕**이지 우연이 아님을 잡는다.

        크기별로 journal 을 격리하던 때로 되돌리면(=앞 주행을 못 보면) 명목
        19 를 **전부 새로 산다**. 이 시험이 없으면 나중에 격리로 되돌아가도
        아무도 모른다.
        """
        m, doc, lock, client, args = self._run(tmp_path, monkeypatch)
        monkeypatch.setattr(m, "read_prior", lambda _out: [])
        # ★격리로 되돌리면 신규가 19라 승인 문에 걸린다 — 여기서 재는 것은
        #  그 문이 아니라 **재사용이 없어지면 수가 는다**는 사실이다.
        monkeypatch.setattr(m, "MAX_NEW_LOGICAL_CALLS", 19)
        monkeypatch.setattr(m, "REQUIRE_DIAGNOSTIC_REUSE", False)
        args.diagnostic_one, args.run = False, True
        m.run_experiment(args, doc, lock, lock["subjects"],
                         m.unique_forms(lock["subjects"]))
        got = json.loads((tmp_path / "out" / "result.json")
                         .read_text(encoding="utf-8"))
        assert got["logical_calls_spent"] == 19, "격리했는데 15 가 나왔다"
        assert got["reused_calls_total"] == 0

    def test_the_scoring_population_is_unchanged_by_reuse(self, tmp_path,
                                                          monkeypatch):
        """★★재사용 행은 **그 크기의 결과에 들어간다** — 모집단이 줄면 안 된다.

        빠지면 batch 8 은 8개만 채점되고, 그 수로 크기를 고르면 **덜 잰 것을
        통과로 삼는다**.
        """
        m, doc, lock, client, args = self._run(tmp_path, monkeypatch)
        args.diagnostic_one, args.run = False, True
        m.run_experiment(args, doc, lock, lock["subjects"],
                         m.unique_forms(lock["subjects"]))
        got = json.loads((tmp_path / "out" / "result.json")
                         .read_text(encoding="utf-8"))
        for b, calls in (("1", 9), ("2", 5), ("4", 3), ("8", 2)):
            r = got["by_batch"][b]
            assert sum(r["counts"].values()) == 9, \
                f"batch {b} 의 채점 모집단이 {sum(r['counts'].values())}개다"
            assert r["bought_calls"] + r["reused_calls"] == calls

    def test_the_first_subject_is_not_written_twice(self, tmp_path,
                                                    monkeypatch):
        m, doc, lock, client, args = self._run(tmp_path, monkeypatch)
        args.diagnostic_one, args.run = False, True
        m.run_experiment(args, doc, lock, lock["subjects"],
                         m.unique_forms(lock["subjects"]))

        first = lock["subjects"][0]["research_subject_id"]
        rows = []
        for f in sorted((tmp_path / "out").glob("*.jsonl")):
            rows += [(f.name, json.loads(x)) for x in
                     f.read_text(encoding="utf-8").splitlines() if x.strip()]
        mine = [(n, r) for n, r in rows
                if r.get("requested") == [first]]
        assert len(mine) == 1, f"첫 subject 줄이 {len(mine)}개다: " \
                               f"{[n for n, _ in mine]}"
        assert mine[0][0] == "diagnostic_one.jsonl"


class TestTheBaselineGateIsClaimsAndSupport:
    """★★★기준선 gate 는 **「계약 통과 claim ≥1 그리고 확인된 인용 ≥1」**이다.

    전에는 semantic 확정(yes+no)이 0이면 세웠는데, strict 진단이 **valid claim
    4 · 인용 있다 3** 인데도 semantic 은 `unresolved` 였다 — 꼭 알아야 할
    구별점을 못 찾으면 계약대로 그렇게 된다. 그걸 「빈손」으로 읽으면 멀쩡한
    판을 세운다. §4b 가 재는 것은 **batch 의 결속·누락·출처**이지 A route 가
    완결됐나가 아니다 (Codex).

    ★이 gate 는 **결과를 본 뒤에 고친 계약**이라, 앞선 판들은
    development/regression 기록일 뿐이다.
    """

    def _run(self, tmp_path, monkeypatch, *, maker, page):
        m = _batch_experiment()
        doc = m.load_manifest(pathlib.Path(__file__).resolve().parents[2])
        lock = doc["locked"]
        client = _Client(maker=maker)
        from app.modules.pipeline.grounding_claims_support import check_run
        monkeypatch.setattr(m, "support_check_run",
                            lambda batches: check_run(batches,
                                                      fetch=lambda _u: page))
        monkeypatch.setattr(m, "REQUIRE_DIAGNOSTIC_REUSE", False)
        monkeypatch.setattr(m, "MAX_NEW_LOGICAL_CALLS", 19)
        monkeypatch.setattr("app.core.openai_keys.openai_client",
                            lambda **kw: client)
        args = SimpleNamespace(out=tmp_path / "out", root=pathlib.Path("."),
                               diagnostic_one=False, run=True, dry_run=False)
        m.run_experiment(args, doc, lock, lock["subjects"],
                         m.unique_forms(lock["subjects"]))
        return client, json.loads((tmp_path / "out" / "result.json")
                                  .read_text(encoding="utf-8"))

    def test_valid_claims_and_a_found_citation_let_it_continue(
            self, tmp_path, monkeypatch):
        """★semantic 이 unresolved 여도 **간다** — 그게 이 수정의 요지다."""
        client, got = self._run(tmp_path, monkeypatch,
                                maker=_resp_with_claims, page="…본문 조각…")
        assert sorted(got["by_batch"]) == ["1", "2", "4", "8"]
        b1 = got["by_batch"]["1"]
        assert b1["contract"]["passed"] >= 1
        assert b1["support"]["counts"]["있다"] >= 1

    def test_no_citation_found_stops_it(self, tmp_path, monkeypatch):
        """★claim 은 있는데 **인용을 못 확인하면** 크기 비교가 성립 안 한다."""
        client, got = self._run(tmp_path, monkeypatch,
                                maker=_resp_with_claims, page=None)
        assert list(got["by_batch"]) == ["1"], "인용 0인데 더 샀다"
        assert got["by_batch"]["1"]["support"]["counts"]["있다"] == 0

    def test_no_valid_claim_stops_it(self, tmp_path, monkeypatch):
        """★인용이 확인돼도 **계약 통과 claim 이 0이면** 선다."""
        client, got = self._run(tmp_path, monkeypatch, maker=None,
                                page="…본문 조각…")
        assert list(got["by_batch"]) == ["1"]
        assert got["by_batch"]["1"]["contract"]["passed"] == 0

    def test_the_result_says_the_gate_was_changed_after_seeing_results(
            self, tmp_path, monkeypatch):
        """★결과를 본 뒤 고친 계약이라는 사실을 **산출에 붙인다**."""
        _c, got = self._run(tmp_path, monkeypatch,
                            maker=_resp_with_claims, page="…본문 조각…")
        assert "결과를 본 뒤에" in got["post_hoc_gate_note"]
        assert "development/regression" in got["post_hoc_gate_note"]
        # ★semantic 은 숨기지 않는다
        assert "semantic" in got["by_batch"]["1"]


class TestAnEmptyBaselineStopsTheRun:
    """★★★기준선(batch 1)에서 확정된 것이 **하나도 없으면** 나머지를 안 산다.

    결속을 잴 claim 이 0 인데 2·4·8 을 사면 **빈손끼리 비교**가 되고, 그 수로
    크기를 고르면 아무것도 아닌 것을 통과로 삼는다. 실측(진단 1회)에서 첫
    대상이 claim 0 이었다 — 이 일이 실제로 일어날 수 있다.
    """

    def _run(self, tmp_path, monkeypatch, maker):
        m = _batch_experiment()
        doc = m.load_manifest(pathlib.Path(__file__).resolve().parents[2])
        lock = doc["locked"]
        client = _Client(maker=maker)
        _no_pages(monkeypatch, m)
        # ★여기서 재는 것은 **승인 범위 문이 아니다** — 그 문은 따로 잰다.
        #  진단을 안 거치는 판이라 신규가 16이므로 명시로 열어 둔다.
        monkeypatch.setattr(m, "MAX_NEW_LOGICAL_CALLS", 19)
        monkeypatch.setattr(m, "REQUIRE_DIAGNOSTIC_REUSE", False)
        monkeypatch.setattr("app.core.openai_keys.openai_client",
                            lambda **kw: client)
        args = SimpleNamespace(out=tmp_path / "out", root=pathlib.Path("."),
                               diagnostic_one=False, run=True, dry_run=False)
        m.run_experiment(args, doc, lock, lock["subjects"],
                         m.unique_forms(lock["subjects"]))
        return client, json.loads((tmp_path / "out" / "result.json")
                                  .read_text(encoding="utf-8"))

    def test_it_stops_after_the_baseline(self, tmp_path, monkeypatch):
        client, got = self._run(tmp_path, monkeypatch, None)  # 기본 fake = claim 0
        assert len(client.calls) == 9, f"{len(client.calls)}회 샀다 — 더 샀다"
        assert got["logical_calls_spent"] == 9
        assert list(got["by_batch"]) == ["1"], "2·4·8 을 재 버렸다"

    def test_it_does_not_stop_when_something_was_established(self, tmp_path,
                                                             monkeypatch):
        """★positive control — 확정된 것이 있으면 끝까지 간다."""
        client, got = self._run(tmp_path, monkeypatch, _resp_with_claims)
        assert sorted(got["by_batch"]) == ["1", "2", "4", "8"]
        # ★진단을 안 거쳤으니 재사용은 **꼬리 3** 뿐이다 — 19 - 3 = 16
        assert got["logical_calls_spent"] == 16
        assert got["reused_calls_total"] == 3


class TestASourcedNoIsNotEmpty:
    """★★★`yes` 만 보면 **출처로 확정한 `no`** 를 빈손으로 오인한다 (Codex).

    `supports_no_difference` 출처 claim 은 「이 시대와 다르지 않다」는 **확정된
    결론**이다. 그걸 빈손으로 읽고 멈추면, 조사가 잘 된 판을 못 재고 끝낸다.
    """

    def test_nine_sourced_no_goes_all_the_way(self, tmp_path, monkeypatch):
        m = _batch_experiment()
        doc = m.load_manifest(pathlib.Path(__file__).resolve().parents[2])
        lock = doc["locked"]
        client = _Client(maker=_resp_with_sourced_no)
        _no_pages(monkeypatch, m)
        monkeypatch.setattr("app.core.openai_keys.openai_client",
                            lambda **kw: client)
        monkeypatch.setattr(m, "MAX_NEW_LOGICAL_CALLS", 19)
        monkeypatch.setattr(m, "REQUIRE_DIAGNOSTIC_REUSE", False)
        args = SimpleNamespace(out=tmp_path / "out", root=pathlib.Path("."),
                               diagnostic_one=False, run=True, dry_run=False)
        m.run_experiment(args, doc, lock, lock["subjects"],
                         m.unique_forms(lock["subjects"]))
        got = json.loads((tmp_path / "out" / "result.json")
                         .read_text(encoding="utf-8"))
        assert sorted(got["by_batch"]) == ["1", "2", "4", "8"], "멈춰 버렸다"
        assert got["by_batch"]["1"]["semantic"]["counts"]["no"] == 9


class TestThePreflightStopsBeforeBuying:
    """★★★승인 상한은 **설명이 아니라 문**이다 (Codex).

    진단 journal 이 없거나 못 읽히면 지금 코드도 고유 16개를 새로 사서 승인
    범위(15)를 넘는다. 그때 「15 이하일 것이다」라는 기대값은 아무것도 안 막는다.
    **보내기 전에** 세고, 넘으면 한 번도 안 산다.
    """

    def _prep(self, tmp_path, monkeypatch):
        m = _batch_experiment()
        doc = m.load_manifest(pathlib.Path(__file__).resolve().parents[2])
        lock = doc["locked"]
        client = _Client(maker=_resp_with_claims)
        _no_pages(monkeypatch, m)
        monkeypatch.setattr("app.core.openai_keys.openai_client",
                            lambda **kw: client)
        args = SimpleNamespace(out=tmp_path / "out", root=pathlib.Path("."),
                               diagnostic_one=True, run=False, dry_run=False)
        m._diagnostic_one(args, doc, lock, lock["subjects"],
                          m.unique_forms(lock["subjects"]))
        args.diagnostic_one, args.run = False, True
        return m, doc, lock, client, args

    def test_it_counts_fifteen_new_and_proceeds(self, tmp_path, monkeypatch):
        m, doc, lock, client, args = self._prep(tmp_path, monkeypatch)
        pre = m.preflight_new_calls(args.out, lock, lock["subjects"])
        assert (pre["logical"], pre["unique"], pre["new"]) == (19, 16, 15)
        assert pre["diagnostic_reused"] is True
        m.run_experiment(args, doc, lock, lock["subjects"],
                         m.unique_forms(lock["subjects"]))
        assert len(client.calls) == 1 + 15

    def test_a_missing_diagnostic_journal_buys_nothing(self, tmp_path,
                                                       monkeypatch):
        m, doc, lock, client, args = self._prep(tmp_path, monkeypatch)
        bought = len(client.calls)
        (args.out / "diagnostic_one.jsonl").unlink()

        with pytest.raises(SystemExit) as exc:
            m.run_experiment(args, doc, lock, lock["subjects"],
                             m.unique_forms(lock["subjects"]))
        assert "16" in str(exc.value) and "15" in str(exc.value)
        assert len(client.calls) == bought, "서고도 샀다"

    def test_a_broken_journal_line_also_stops(self, tmp_path, monkeypatch):
        """★못 읽는 줄은 「없는 것」과 같다 — 조용히 다시 사면 안 된다."""
        m, doc, lock, client, args = self._prep(tmp_path, monkeypatch)
        bought = len(client.calls)
        (args.out / "diagnostic_one.jsonl").write_text("{깨진 줄\n",
                                                       encoding="utf-8")
        with pytest.raises(SystemExit):
            m.run_experiment(args, doc, lock, lock["subjects"],
                             m.unique_forms(lock["subjects"]))
        assert len(client.calls) == bought

    def test_the_physical_cap_is_new_times_slots(self, tmp_path, monkeypatch):
        m, doc, lock, client, args = self._prep(tmp_path, monkeypatch)
        monkeypatch.setattr("app.core.openai_keys.slot_count", lambda: 2)
        m.run_experiment(args, doc, lock, lock["subjects"],
                         m.unique_forms(lock["subjects"]))
        got = json.loads((tmp_path / "out" / "result.json")
                         .read_text(encoding="utf-8"))
        # ★논리 19×2=38 이 아니라 **신규 15×2=30** 이다 — 안 살 것까지 세지 않는다
        assert got["approved_physical_cap"] == 30
        # ★실제로 산 것 × 슬롯 — 못 센 것을 「actual」이라고 쓰지 않는다
        assert got["this_run_physical_upper_bound"] == got[
            "logical_calls_spent"] * 2



class TestABrokenBaselineStopsBeforeBuyingMore:
    """★★★Codex 재현 — preflight 15 인데 실제 **16**을 샀다.

    `by_batch (bought,reused)` = 1:(8,1) · 2:(5,0) · 4:(2,1) · 8:(1,1).
    batch 1 의 호출 하나가 **실패하면** 그 행은 재사용 대상이 아니라서, 크기마다
    겹치는 **꼬리 singleton** 을 batch 2 가 다시 산다 — 그래서 하나가 더 나간다.
    ★그 상태는 **소유권 기준선도 불완전**하므로 batch>1 은 어차피 후보가 못
    된다. 돈과 실험 의미가 같은 방향이라 **거기서 끝낸다**.
    """

    def test_a_failed_baseline_call_stops_the_run(self, tmp_path, monkeypatch):
        m = _batch_experiment()
        doc = m.load_manifest(pathlib.Path(__file__).resolve().parents[2])
        lock = doc["locked"]
        client = _Client(maker=_resp_with_claims)
        _no_pages(monkeypatch, m)
        monkeypatch.setattr("app.core.openai_keys.openai_client",
                            lambda **kw: client)
        args = SimpleNamespace(out=tmp_path / "out", root=pathlib.Path("."),
                               diagnostic_one=True, run=False, dry_run=False)
        m._diagnostic_one(args, doc, lock, lock["subjects"],
                          m.unique_forms(lock["subjects"]))
        assert len(client.calls) == 1

        # ★batch 1 의 마지막(꼬리 singleton) 호출을 실패시킨다 — 재사용 1 을
        #  빼면 8회를 사고, 그 마지막이 index 8 이다.
        client._fail_on = {8}
        args.diagnostic_one, args.run = False, True
        m.run_experiment(args, doc, lock, lock["subjects"],
                         m.unique_forms(lock["subjects"]))

        # ★진단 1 + batch 1 의 8 에서 **멈춘다** — 16번째를 사지 않는다
        assert len(client.calls) == 1 + 8, f"{len(client.calls)}회 샀다"
        got = json.loads((tmp_path / "out" / "result.json")
                         .read_text(encoding="utf-8"))
        assert list(got["by_batch"]) == ["1"]
        assert got["logical_calls_spent"] == 8


class TestTheEndReportSaysThisRunsBound:
    """★★승인 상한(30)을 끝 화면에 찍으면 8회를 사고도 「30까지 나갔다」로
    읽힌다 (Codex). **이번 주행 상한**은 `산 것 × 슬롯` 이다."""

    def _src(self):
        return pathlib.Path(_batch_experiment().__file__).read_text(
            encoding="utf-8")

    def test_the_final_line_reports_the_run_bound_not_the_cap(self):
        src = self._src()
        assert "이번 주행 물리 전송 ≤ " in src
        assert "{spent * max(1, slots)}" in src
        # ★승인 상한도 **같이** 말한다 — 둘을 갈라 쓴다
        assert "승인 상한은 {phys_cap}" in src

    def test_the_diagnostic_does_not_say_first_subject(self):
        """★대상은 `--subject` 로 고른다 — 「첫 대상」이라 쓰면 거짓이 된다."""
        src = self._src()
        for line in src.splitlines():
            if line.strip().startswith("#") or '"""' in line:
                continue
            assert "첫 대상의 재사용" not in line, line
        assert "research_subject_id" in src.split("★진단 1회:")[1][:200]


class TestEarlyStopReportsSixteenAndThirty:
    """★★★조기 종료 때 화면과 JSON 이 **서로 다른 두 수**를 말해야 한다 (Codex).

    승인 상한은 30(신규 15 × 슬롯 2)이고, 8회만 사고 섰으면 **이번 주행 상한은
    16**(8 × 2)이다. 화면에 30을 찍으면 「30까지 나갔다」로 읽힌다.
    """

    def test_screen_says_sixteen_and_json_keeps_both(self, tmp_path,
                                                     monkeypatch, capsys):
        m = _batch_experiment()
        doc = m.load_manifest(pathlib.Path(__file__).resolve().parents[2])
        lock = doc["locked"]
        # ★batch 1 만 사고 서게 만든다 — 인용을 못 확인하면 거기서 끝난다
        client = _Client(maker=_resp_with_claims)
        from app.modules.pipeline.grounding_claims_support import check_run
        monkeypatch.setattr(m, "support_check_run",
                            lambda b: check_run(b, fetch=lambda _u: None))
        monkeypatch.setattr("app.core.openai_keys.slot_count", lambda: 2)
        monkeypatch.setattr("app.core.openai_keys.openai_client",
                            lambda **kw: client)
        # ★진단을 먼저 사서 앞 기록을 만든다 — 그래야 신규가 15가 되고
        #  승인 상한이 30 이다. 실제 판과 같은 모양이다.
        args = SimpleNamespace(out=tmp_path / "out", root=pathlib.Path("."),
                               diagnostic_one=True, run=False, dry_run=False)
        m._diagnostic_one(args, doc, lock, lock["subjects"],
                          m.unique_forms(lock["subjects"]))
        args.diagnostic_one, args.run = False, True
        m.run_experiment(args, doc, lock, lock["subjects"],
                         m.unique_forms(lock["subjects"]))

        got = json.loads((tmp_path / "out" / "result.json")
                         .read_text(encoding="utf-8"))
        spent = got["logical_calls_spent"]
        assert spent == 8, f"조기 종료인데 {spent}회 샀다"
        assert got["approved_physical_cap"] == 30
        assert got["this_run_physical_upper_bound"] == spent * 2
        out = capsys.readouterr().out
        assert f"이번 주행 물리 전송 ≤ {spent * 2}" in out
        assert "승인 상한은 30" in out
        assert f"물리 전송 ≤ 30" not in out.replace(
            f"이번 주행 물리 전송 ≤ {spent * 2}", "")

    def test_no_string_calls_it_the_first_subjects_reuse(self):
        """★「첫 대상 재사용」은 이제 거짓이다 — 대상은 골라서 산다."""
        src = pathlib.Path(_batch_experiment().__file__).read_text(
            encoding="utf-8")
        assert "첫 대상의 재사용" not in src
        assert "첫 호출 재사용" not in src
        assert "앞서 산 exact payload" in src


class TestTheCommentsMatchTheCode:
    """★★계약을 코드와 **반대로** 적어 두면, 다음 사람이 그 주석을 믿고 고친다.

    실측(Codex 리뷰): `SCHEMA_STRICT=True` 인데 세 곳이 「strict 는 안 켠다」·
    「strict=false 라 우리가 막는다」로 적혀 있었다.
    """

    def _srcs(self):
        from app.modules.pipeline import grounding_claims_acceptance as acc
        return [pathlib.Path(g.__file__).read_text(encoding="utf-8"),
                pathlib.Path(acc.__file__).read_text(encoding="utf-8")]

    def test_no_source_says_strict_is_off(self):
        for src in self._srcs():
            assert "strict 는 안 켠다" not in src
            assert "`strict=false` 로 보내므로" not in src
            assert "★`strict=false` 라 우리가 막는다" not in src

    def test_the_local_check_is_described_as_independent(self):
        """★우리 검증은 strict 를 켠 **뒤에도** 남는다 — 그 이유를 적어 둔다."""
        for src in self._srcs():
            if "schema_violations" in src or "SCHEMA_STRICT" in src:
                assert "독립 fail-closed" in src