"""GROUNDING-V2 §2-4b — **프로덕션 조사 스텝**.

★이 스텝이 생기기 전에는 `search_claims` 를 부르는 프로덕션 코드가 **0건**
이었다. 그래서 「상한 안에서만 산다」·「마감 안 결과만 남는다」가 실험 도구
에서만 참이었다.
"""
from types import SimpleNamespace
from unittest.mock import patch

import pytest

from app.core.step_manifest import STEP_MANIFEST
from app.core.steps import STEP_CLASSES
from app.core.steps.grounding_steps import GroundingResearchStep


class TestItIsRegisteredInTheChain:
    """★등록이 안 되면 스텝이 **아예 안 돈다** — 코드만 있고 뜻이 없다."""

    def test_the_runner_is_dispatchable(self):
        assert STEP_CLASSES["grounding_research"] is GroundingResearchStep

    def test_it_sits_between_plan_and_filter(self):
        m = STEP_MANIFEST["grounding_research"]
        assert (STEP_MANIFEST["grounding_plan"]["order"] < m["order"]
                < STEP_MANIFEST["entity_filter"]["order"])

    def test_the_plan_invalidates_it(self):
        """★계획이 다시 돌면 조사도 다시 돈다 — 순서만이 아니라 무효화 사슬."""
        assert "grounding_plan" in STEP_MANIFEST[
            "grounding_research"]["depends_on"]

    def test_legacy_does_not_run_it(self):
        """★유료 호출이 있는 스텝이다. legacy 에서 돌면 안 된다."""
        assert STEP_MANIFEST["grounding_research"]["applicability"] == \
            "if_grounding_v2"

    def test_a_capped_run_blocks_downstream(self):
        """★상한에 걸린 것을 「조사할 게 없었다」로 읽지 않는다."""
        m = STEP_MANIFEST["grounding_research"]
        assert m["allow_partial_downstream"] is False
        assert m["partial_override_config_key"] == "allow_missing_grounding"


class TestTheBatchSizeIsWhatTheExperimentSaid:
    """★§4b 실험이 네 크기를 다 돌고 「후보 없음 · 개별 호출 유지」로 끝났다.

    어긋남은 어느 크기에서도 0이었지만 **크기를 키울수록 조사량이 말랐다**
    (계약 통과 claim 19 → 16 → 10 → 9).
    """

    def test_it_is_one(self):
        assert GroundingResearchStep.BATCH_SIZE == 1


class TestTheConfigHashFoldsTheRequestContract:
    """★★`strict` 나 `include` 를 고쳐도 resume 이 **옛 체크포인트를 건너뛰면**
    고친 것이 아무 데도 안 닿는다."""

    def _hash(self, step):
        return step._config_hash()

    def test_changing_the_request_changes_the_hash(self):
        from app.modules.pipeline import grounding_claims_search as gcs

        step = _step()
        before = self._hash(step)
        with patch.object(gcs, "SCHEMA_STRICT", not gcs.SCHEMA_STRICT):
            after = self._hash(step)
        assert before != after, "요청 계약이 바뀌었는데 지문이 같다"

    def test_the_same_config_gives_the_same_hash(self):
        """★positive control — 늘 다르면 resume 이 통째로 죽는다."""
        step = _step()
        assert self._hash(step) == self._hash(step)


class TestItOnlyBuysWhatThePlanMarked:
    """★`route="research"` 인 것만 산다 — `skip` 을 사면 계획이 뜻을 잃는다."""

    def test_skip_rows_are_not_bought(self):
        step = _step(decided=[
            {"research_subject_id": "rs_a", "route": "research"},
            {"research_subject_id": "rs_b", "route": "skip"},
        ], subjects=[_subj("rs_a"), _subj("rs_b")])
        picked, era, region, _ns, _ms = step._research_targets()
        assert [s["research_subject_id"] for s in picked] == ["rs_a"]
        assert era and region

    def test_a_subject_without_a_quote_is_not_bought(self):
        """★상상 묘사로 조사하면 그 결과가 무엇의 것인지 알 수 없다."""
        step = _step(decided=[{"research_subject_id": "rs_a",
                               "route": "research"}],
                     subjects=[_subj("rs_a", quote="")])
        picked, _e, _r, no_source, _ms = step._research_targets()
        assert picked == []
        # ★조용히 버리지 않는다 — 어디로 갔는지 남는다
        assert no_source == ["rs_a"]

    def test_no_world_context_stops_it(self):
        from app.core.errors import AppError

        step = _step(decided=[{"research_subject_id": "rs_a",
                               "route": "research"}],
                     subjects=[_subj("rs_a")], era="")
        with pytest.raises(AppError, match="시대/지역"):
            step._research_targets()


class TestTheRunHappensInsideTheScope:
    """★★★상한과 마감을 **열고 나서** 산다. 열기 전에 부르면 그 호출은 아무
    문도 안 지난다 — 이 스텝이 있는 이유의 절반이 그것이다."""

    def test_the_budget_is_installed_when_search_runs(self):
        from app.core import research_call_budget as rb

        seen = {}

        def _fake_search(client, subjects, **kw):
            seen["budget"] = rb.get_current_budget()
            seen["batch_size"] = kw.get("batch_size")
            return {"batches": [], "logical_calls": 0, "bought_calls": 0}

        step = _step(decided=[{"research_subject_id": "rs_a",
                               "route": "research"}],
                     subjects=[_subj("rs_a")])
        with patch("app.modules.pipeline.grounding_claims_search.search_claims",
                   _fake_search), \
             patch("app.core.openai_keys.openai_client", lambda **k: object()):
            step._execute()
        assert seen["budget"] is not None, "상한을 안 열고 샀다"
        assert seen["batch_size"] == 1
        # ★나가면 되돌아온다 — 남의 호출이 이 상한을 보면 안 된다
        assert rb.get_current_budget() is None

    def test_a_capped_row_counts_as_failed(self):
        """★「시간이 없어 못 산 것」이 `completed` 로 지나가면 하류가 오독한다."""
        rows = [{"limit_kind": "run_deadline", "error": "마감"},
                {"limit_kind": "", "error": "", "parsed": {"results": []}}]
        got = GroundingResearchStep._wrap_research("v2", rows=rows)
        assert got["failed_count"] == 1
        assert got["completed_count"] == 1
        assert got["data"]["limit_kinds"] == ["run_deadline"]

    def test_legacy_does_nothing(self):
        step = _step(mode="legacy")
        got = step._execute()
        assert got["data"]["skipped"] is True
        assert got["applicable_count"] == 0


# ── 대역 ─────────────────────────────────────────────────────────────────
def _subj(sid, quote="1983년 늦가을 저녁의 문장"):
    return {"research_subject_id": sid, "surface_form": "가",
            "owner_type": "prop", "source_quote": quote,
            "provenance": {"short_id": ""}}


def _step(*, decided=None, subjects=None, era="1983년 늦가을",
          mode="v2"):
    """★스텝을 **진짜 클래스로** 만든다 — 흉내 내면 배선을 안 재게 된다."""
    step = GroundingResearchStep.__new__(GroundingResearchStep)
    step.project_id = "p1"
    step.episode_id = "e1"
    step.project_config = {}
    step.db = None
    step.build_opik_metadata = lambda: {}
    cps = {
        "grounding_plan": {"data": {"decided": decided or []}},
        "visual_world_rules": {"data": {"era": era, "region": "대한민국"}},
        "entity_merge": {"data": {}},
        "grounding_a0": {"data": {"candidates": []}},
    }
    step._load_prev_checkpoint = lambda name: cps.get(name)
    patcher = patch(
        "app.modules.pipeline.grounding_carry.build_subjects",
        lambda *a, **k: {"subjects": subjects or [], "unbound": [],
                         "carry_reasons": {}, "carried": 0})
    patcher.start()
    patch("app.core.grounding_mode.resolve_grounding_mode",
          lambda cfg: mode).start()
    patch("app.core.steps.grounding_steps.resolve_grounding_mode",
          lambda cfg: mode).start()
    return step


@pytest.fixture(autouse=True)
def _stop_patches():
    yield
    patch.stopall()


class TestItSavesDurableRevisions:
    """★★산 것을 **정본 테이블**에 남긴다. 안 남기면 다음 주행이 다시 산다.

    ★판정은 `build_revision_row` 가 한 번에 한다 — 호출부가 따로 `decide_delta`
    를 부르고 결과만 넘기면 저장된 판정과 저장된 claims 가 갈린다.
    """

    def _saved(self, rows, subjects):
        added = []

        class _DB:
            @staticmethod
            def add(obj):
                added.append(obj)

            @staticmethod
            def flush():
                pass

        step = _step()
        step.db = _DB()
        n = step._save_revisions(rows, subjects, era="1983년", region="대한민국")
        return n, added

    def _row(self, sid, *, claims=(), gaps=(), limit_kind=""):
        return {
            "requested": [sid], "limit_kind": limit_kind,
            "parsed": {"results": [{"research_subject_id": sid,
                                    "claims": list(claims),
                                    "gaps": list(gaps)}]},
            "provenance": {"provider": "openai", "model": "m",
                           "prompt_pack_version": "3.1",
                           "prompt_raw_hash": "a" * 64,
                           "schema_raw_hash": "b" * 64,
                           "payload_hash": "c" * 16,
                           "prompt_locator": "x", "schema_locator": "y",
                           "local_trace_id": "t", "provider_request_id": "r",
                           "transmission_status": "ok"},
        }

    def test_a_row_is_saved_per_subject(self):
        n, added = self._saved([self._row("rs_a")], [_subj("rs_a")])
        assert n == 1 and len(added) == 1
        assert added[0].research_subject_id == "rs_a"

    def test_a_capped_row_is_saved_as_retryable(self):
        """★「시간에 잘린 것」과 「다 보고 못 정한 것」을 갈라야 다시 산다."""
        from app.modules.pipeline.grounding_claims import STATUS_RETRYABLE

        _n, added = self._saved(
            [self._row("rs_a", limit_kind="run_deadline")], [_subj("rs_a")])
        assert added[0].status == STATUS_RETRYABLE

    def test_the_payload_hash_comes_from_production(self):
        """★도구가 따로 계산하면 잠근 신원과 저장된 신원이 갈린다."""
        import json as _json

        from app.modules.pipeline.grounding_claims import subject_payload_hash

        subj = _subj("rs_a")
        _n, added = self._saved([self._row("rs_a")], [subj])
        ri = _json.loads(added[0].research_input_json)
        assert ri["subject_payload_hash"] == subject_payload_hash(subj)
        assert ri["research_subject_id"] == "rs_a"

    def test_a_subject_that_was_not_requested_is_not_saved(self):
        """★남의 줄이 섞여 와도 저장하지 않는다."""
        n, added = self._saved([self._row("rs_a")], [_subj("rs_b")])
        assert n == 0 and added == []


class TestNothingIsSilentlyDropped:
    """★★★계획이 `research` 로 표시했는데 **인용이 없으면**, 그건 「조사할 대상이
    없다」가 아니라 `no_source`·`unresolved` 다 (Codex).

    조용히 버리면 **유료 호출 0회로 `completed`** 가 되어 하류가 「조사할 게
    없었다」로 읽는다.
    """

    def test_all_without_quotes_buys_nothing_and_is_partial(self):
        called = {"n": 0}

        def _never(*a, **k):
            called["n"] += 1
            return {"batches": [], "logical_calls": 0, "bought_calls": 0}

        step = _step(decided=[{"research_subject_id": "rs_a",
                               "route": "research"}],
                     subjects=[_subj("rs_a", quote="")])
        with patch("app.modules.pipeline.grounding_claims_search.search_claims",
                   _never):
            got = step._execute()
        assert called["n"] == 0, "인용이 없는데 샀다"
        assert got["failed_count"] == 1, "실패로 안 셌다 — completed 로 지나간다"
        assert got["applicable_count"] == 1, "모집단에서 빠졌다"
        assert got["data"]["no_source"] == ["rs_a"]

    def test_a_subject_the_plan_wanted_but_we_could_not_build(self):
        """★subject 를 아예 못 세운 것도 **전수**에 남는다."""
        step = _step(decided=[{"research_subject_id": "rs_gone",
                               "route": "research"}], subjects=[])
        got = step._execute()
        assert got["data"]["unbuilt"] == ["rs_gone"]
        assert got["failed_count"] == 1

    def test_a_quoted_subject_is_still_bought(self):
        """★positive control — 다 막으면 아무것도 안 산다."""
        seen = {"n": 0}

        def _fake(*a, **k):
            seen["n"] += 1
            return {"batches": [], "logical_calls": 1, "bought_calls": 1}

        step = _step(decided=[{"research_subject_id": "rs_a",
                               "route": "research"}],
                     subjects=[_subj("rs_a")])
        with patch("app.modules.pipeline.grounding_claims_search.search_claims",
                   _fake), \
             patch("app.core.openai_keys.openai_client", lambda **k: object()):
            step._execute()
        assert seen["n"] == 1


class TestTheChainInvalidatesDownstream:
    """★★★조사가 다시 돌면 **참조 정책과 scene_detail 이 stale** 이어야 한다.

    무효화가 안 이어지면 고친 조사가 아무 데도 안 닿는다 — 하류가 옛 참조
    정책을 그대로 쓴다.
    """

    def test_the_policy_depends_on_research(self):
        assert "grounding_research" in STEP_MANIFEST[
            "episode_reference_policy"]["depends_on"]

    def test_scene_detail_depends_on_the_policy(self):
        """★사슬의 마지막 고리 — 여기가 끊기면 위가 다 헛돈다."""
        assert "episode_reference_policy" in STEP_MANIFEST[
            "scene_detail"]["depends_on"]

    def test_the_filter_does_not_depend_on_research(self):
        """★`entity_filter` 는 조사 결과를 **안 쓴다** — 거짓 gate 를 안 만든다."""
        assert "grounding_research" not in STEP_MANIFEST[
            "entity_filter"]["depends_on"]


class TestLegacyAddsNothing:
    """★★`legacy` 에서는 이 스텝이 **not_applicable** 이라 아무것도 안 더한다."""

    def test_it_produces_no_rows_and_no_calls(self):
        called = {"n": 0}

        def _never(*a, **k):
            called["n"] += 1
            return {"batches": []}

        step = _step(decided=[{"research_subject_id": "rs_a",
                               "route": "research"}],
                     subjects=[_subj("rs_a")], mode="legacy")
        with patch("app.modules.pipeline.grounding_claims_search.search_claims",
                   _never):
            got = step._execute()
        assert called["n"] == 0, "legacy 인데 유료 호출을 했다"
        assert got["data"]["rows"] == []
        assert got["failed_count"] == 0

    def test_the_policy_gets_an_empty_forced_set_without_a_db(self):
        """★조사 기록이 없으면 **강제도 차단도 없다** — legacy 가 그 판이다."""
        from app.core.steps.episode_reference_policy_step import (
            EpisodeReferencePolicyStep)

        step = EpisodeReferencePolicyStep.__new__(EpisodeReferencePolicyStep)
        step.project_id = "p1"
        step.episode_id = "e1"
        step.project_config = {}
        step.db = None
        forced, blocked = step._research_forced_short_ids()
        assert forced == set() and blocked == set()


class TestItDoesNotBuyWhatIsAlreadyDone:
    """★★★재개할 때 **같은 입력을 다시 사지 않는다** (§2-4a ②).

    이 배선이 없으면 재개마다 전부 다시 산다 — 이 프로젝트에서 제일 비싼
    결함 부류다.
    """

    def _run(self, done, subjects, *, cap=None):
        from types import SimpleNamespace

        seen = {"subjects": None}

        def _fake(client, subs, **kw):
            seen["subjects"] = [s["research_subject_id"] for s in subs]
            return {"batches": [], "logical_calls": 0, "bought_calls": 0}

        step = _step(decided=[{"research_subject_id": s["research_subject_id"],
                               "route": "research"} for s in subjects],
                     subjects=subjects)
        if cap:
            step.project_config = {"research_subject_cap": cap}

        class _Q:
            def filter(self, *a):
                return self

            def all(self):
                return done

        step.db = SimpleNamespace(query=lambda m: _Q(), add=lambda o: None,
                                  flush=lambda: None)
        with patch("app.modules.pipeline.grounding_claims_search.search_claims",
                   _fake), \
             patch("app.core.openai_keys.openai_client", lambda **k: object()):
            got = step._execute()
        return seen["subjects"], got

    def _done(self, sid, subj, *, status="completed"):
        from types import SimpleNamespace

        from app.modules.pipeline.grounding_claims import (
            research_input_hash, subject_payload_hash)

        step = _step()
        ri = step._research_inputs(era="1983년 늦가을", region="대한민국")
        return SimpleNamespace(
            research_subject_id=sid, status=status,
            research_input_hash=research_input_hash(
                research_subject_id=sid,
                subject_payload_hash=subject_payload_hash(subj), **ri))

    def test_a_completed_subject_is_not_bought_again(self):
        a, b = _subj("rs_a"), _subj("rs_b")
        bought, got = self._run([self._done("rs_a", a)], [a, b])
        assert bought == ["rs_b"], f"이미 끝난 것을 또 샀다: {bought}"
        assert got["data"]["already_done"] == ["rs_a"]

    def test_a_retryable_subject_is_bought_again(self):
        """★positive control — 다시 사야 할 것까지 막으면 영영 못 끝낸다."""
        a = _subj("rs_a")
        bought, _got = self._run([self._done("rs_a", a, status="retryable")],
                                 [a])
        assert bought == ["rs_a"]

    def test_the_cap_defers_instead_of_dropping(self):
        """★상한을 넘은 것은 **버리지 않는다** — 다음 주행이 이어간다."""
        a, b = _subj("rs_a"), _subj("rs_b")
        bought, got = self._run([], [a, b], cap=1)
        assert len(bought) == 1
        assert got["data"]["deferred"] == ["rs_b"]
        # ★밀린 것도 실패로 센다 — 「조사할 게 없었다」가 아니다
        assert got["failed_count"] >= 1

    def test_all_done_buys_nothing(self):
        a = _subj("rs_a")
        bought, got = self._run([self._done("rs_a", a)], [a])
        assert bought is None, "살 것이 없는데 호출했다"
        assert got["data"]["already_done"] == ["rs_a"]


class TestARowThatWasNeverSentIsStillRecorded:
    """★★★크기 hard 상한을 넘어 **안 보낸 행**은 provenance 가 없다 (Codex).

    그대로 계약에 넣으면 터지고, 걸러 버리면 **조용히 사라진다** — 둘 다 틀렸다.
    전송 0인 **bounded-run 미결**로 남아 재개할 수 있어야 한다.
    """

    def _not_sent_row(self):
        """★`search_claims` 가 **실제로 내는** 모양으로 만든다 — 손으로
        지어내면 이 시험이 다른 것을 잰다."""
        from app.modules.pipeline import grounding_claims_search as gcs

        out = gcs.search_claims(
            None, [{"research_subject_id": "rs_a", "surface_form": "가",
                    "owner_type": "prop",
                    "source_quote": "문장" * 200_000}],
            model="m", era="1983년", region="대한민국", batch_size=1,
            max_prompt_bytes=80, hard_prompt_bytes=100)
        row = out["batches"][0]
        assert row["not_sent"] is True
        return row

    def _save(self, row):
        added = []

        class _DB:
            @staticmethod
            def add(o):
                added.append(o)

            @staticmethod
            def flush():
                pass

        step = _step()
        step.db = _DB()
        n = step._save_revisions([row], [_subj("rs_a")], era="1983년",
                                 region="대한민국")
        return n, added

    def test_it_does_not_raise_and_leaves_a_row(self):
        n, added = self._save(self._not_sent_row())
        assert n == 1 and len(added) == 1

    def test_it_is_retryable(self):
        """★다시 살 수 있어야 한다 — 「다 보고 못 정했다」가 아니다."""
        from app.modules.pipeline.grounding_claims import STATUS_RETRYABLE

        _n, added = self._save(self._not_sent_row())
        assert added[0].status == STATUS_RETRYABLE

    def test_the_reason_survives(self):
        """★왜 못 보냈는지 안 남기면 고칠 곳을 못 찾는다."""
        import json as _json

        _n, added = self._save(self._not_sent_row())
        prov = _json.loads(added[0].provenance_json)
        assert "상한" in prov["transmission_error"]


class TestEachPaidCallIsSavedImmediately:
    """★★★9개 중 7개째에 끊기면 **앞의 7개를 통째로 잃는다** — 실제로 그
    부류의 결함을 이 프로젝트에서 냈다.

    한 호출이 끝날 때마다 바로 남긴다.
    """

    def test_the_callback_saves_before_the_run_ends(self):
        from types import SimpleNamespace

        added = []

        def _fake(client, subs, **kw):
            cb = kw.get("on_batch")
            assert cb is not None, "즉시 저장 콜백을 안 넘겼다"
            cb({"requested": ["rs_a"], "parsed": {"results": []},
                "provenance": {"provider": "openai", "model": "m",
                               "prompt_pack_version": "3.1",
                               "prompt_raw_hash": "a" * 64,
                               "schema_raw_hash": "b" * 64,
                               "payload_hash": "c" * 16,
                               "prompt_locator": "x", "schema_locator": "y",
                               "local_trace_id": "t",
                               "provider_request_id": "r",
                               "transmission_status": "ok"}})
            # ★여기서 죽어도 위에서 저장된 것은 남아 있어야 한다
            raise RuntimeError("주행이 끊겼다")

        step = _step(decided=[{"research_subject_id": "rs_a",
                               "route": "research"}],
                     subjects=[_subj("rs_a")])
        step.db = SimpleNamespace(
            query=lambda m: SimpleNamespace(filter=lambda *a: SimpleNamespace(
                all=lambda: [])),
            add=lambda o: added.append(o), flush=lambda: None)
        with patch("app.modules.pipeline.grounding_claims_search.search_claims",
                   _fake), \
             patch("app.core.openai_keys.openai_client", lambda **k: object()):
            with pytest.raises(RuntimeError, match="끊겼다"):
                step._execute()
        assert len(added) == 1, "끊기기 전에 산 것이 안 남았다"


class TestAllThreeSitesReadTheSamePack:
    """★★★`search_claims(db=self.db)` 는 **활성 DB 팩**을 읽는데 신원·지문을
    `db=None` 으로 읽으면 셋이 갈린다 (Codex).

    ①admission 이 낸 hash 와 저장된 hash 가 달라 **재개마다 다시 사고**
    ②config 지문이 DB 변경을 못 봐 **완료 CP 를 건너뛴다**.
    """

    def test_the_pack_is_loaded_with_the_db(self):
        seen = {}

        def _load(**kw):
            seen["db"] = kw.get("db", "안 넘김")
            return {"module": "m", "version": "v",
                    "pack_manifest_hash": "h",
                    "stems": {"system": {"raw_content_hash": "a",
                                         "content": "x"},
                              "claims_schema": {"raw_content_hash": "b",
                                                "content": {}}}}

        step = _step()
        step.db = object()
        with patch("app.modules.pipeline.grounding_claims_search.load_pack",
                   _load):
            step._pack()
        assert seen["db"] is step.db, "팩을 DB 없이 읽었다"

    def test_the_pack_is_read_once_per_run(self):
        """★두 번 읽으면 그 사이에 바뀐 판에서 신원과 지문이 갈린다."""
        calls = {"n": 0}

        def _load(**kw):
            calls["n"] += 1
            return {"module": "m", "version": "v", "pack_manifest_hash": "h",
                    "stems": {"system": {"raw_content_hash": "a"},
                              "claims_schema": {"raw_content_hash": "b"}}}

        step = _step()
        with patch("app.modules.pipeline.grounding_claims_search.load_pack",
                   _load):
            step._pack()
            step._pack()
            step._config_hash()
            step._research_inputs(era="1983년", region="대한민국")
        assert calls["n"] == 1, f"팩을 {calls['n']}번 읽었다"

    def test_a_db_pack_change_moves_the_config_hash(self):
        """★DB 프롬프트를 고쳤는데 지문이 안 움직이면 완료 CP 를 건너뛴다."""
        def _load_with(h):
            def _f(**kw):
                return {"module": "m", "version": "v",
                        "pack_manifest_hash": h,
                        "stems": {"system": {"raw_content_hash": h},
                                  "claims_schema": {"raw_content_hash": "b"}}}
            return _f

        a, b = _step(), _step()
        with patch("app.modules.pipeline.grounding_claims_search.load_pack",
                   _load_with("옛것")):
            before = a._config_hash()
        with patch("app.modules.pipeline.grounding_claims_search.load_pack",
                   _load_with("새것")):
            after = b._config_hash()
        assert before != after

    def test_the_input_identity_uses_the_same_pack(self):
        """★admission 이 본 것과 저장되는 것이 **같은 팩**이어야 한다."""
        def _load(**kw):
            return {"module": "m", "version": "특정버전",
                    "pack_manifest_hash": "h",
                    "stems": {"system": {"raw_content_hash": "aa"},
                              "claims_schema": {"raw_content_hash": "bb"}}}

        step = _step()
        with patch("app.modules.pipeline.grounding_claims_search.load_pack",
                   _load):
            ri = step._research_inputs(era="1983년", region="대한민국")
        assert ri["claims_pack_version"] == "특정버전"
        assert ri["prompt_raw_hash"] == "aa"


class TestARunWithNothingToResearchIsNormal:
    """★★★조사할 대상이 하나도 없는 판은 **정상**이다 — 예외가 아니다.

    전에는 `_research_targets` 가 그 갈래에서만 3-tuple 을 돌려줘, 호출부가
    풀다가 `ValueError` 로 죽었다. v2 에서 전부 `skip` 인 멀쩡한 판이
    통째로 실패했다 (Codex).
    """

    def test_it_does_not_raise(self):
        step = _step(decided=[{"research_subject_id": "rs_a",
                               "route": "skip"}],
                     subjects=[_subj("rs_a")])
        got = step._research_targets()
        assert len(got) == 5, f"모양이 {len(got)}개다 — 호출부가 죽는다"

    def test_the_public_run_buys_nothing_and_completes(self):
        called = {"n": 0}

        def _never(*a, **k):
            called["n"] += 1
            return {"batches": []}

        step = _step(decided=[{"research_subject_id": "rs_a",
                               "route": "skip"}],
                     subjects=[_subj("rs_a")])
        with patch("app.modules.pipeline.grounding_claims_search.search_claims",
                   _never):
            got = step._execute()
        assert called["n"] == 0, "조사할 게 없는데 샀다"
        assert got["failed_count"] == 0, "정상 판인데 실패로 셌다"
        assert got["data"]["rows"] == []

    def test_an_empty_decided_list_is_also_fine(self):
        step = _step(decided=[], subjects=[])
        got = step._execute()
        assert got["failed_count"] == 0 and got["data"]["rows"] == []


class TestAHardLimitRowIsRecordedAsBoundedRun:
    """★★★크기 hard 상한은 **물리 전송 0** 인 로컬 상한이다 (Codex).

    ```
    provider 0회 → revision status=retryable
                 → gap reason=time_capped · limit_kind=admission_limit
                 → 다음 주행이 **다시 들어갈 수 있다**
    ```
    ★`provider transmission_failed` 로 적으면 「provider 장애」로 오독되고
    고칠 곳이 달라진다.
    """

    def _row(self):
        from app.modules.pipeline import grounding_claims_search as gcs

        seen = {"n": 0}

        class _Client:
            @property
            def responses(self):
                class _R:
                    @staticmethod
                    def create(**kw):
                        seen["n"] += 1
                        raise AssertionError("보내면 안 된다")
                return _R

        out = gcs.search_claims(
            _Client(), [{"research_subject_id": "rs_a", "surface_form": "가",
                         "owner_type": "prop", "source_quote": "문장" * 50_000}],
            model="m", era="1983년", region="대한민국", batch_size=1,
            max_prompt_bytes=80, hard_prompt_bytes=100)
        assert seen["n"] == 0, "안 보내야 하는데 provider 를 불렀다"
        return out["batches"][0]

    def test_it_is_marked_as_an_admission_limit(self):
        row = self._row()
        assert row["not_sent"] is True
        assert row["limit_kind"] == "admission_limit"
        assert row["provenance"]["transmission_status"] == "not_sent"

    def test_the_saved_revision_is_retryable_with_the_right_gap(self):
        import json as _json

        from app.modules.pipeline.grounding_claims import (GAP_TIME_CAPPED,
                                                           STATUS_RETRYABLE)

        added = []

        class _DB:
            @staticmethod
            def add(o):
                added.append(o)

            @staticmethod
            def flush():
                pass

        step = _step()
        step.db = _DB()
        step._save_revisions([self._row()], [_subj("rs_a")], era="1983년",
                             region="대한민국")
        assert added[0].status == STATUS_RETRYABLE
        gaps = _json.loads(added[0].gaps_json)
        assert any(g["reason"] == GAP_TIME_CAPPED
                   and g["limit_kind"] == "admission_limit" for g in gaps), gaps

    def test_the_callback_gets_it_too(self):
        """★안 넘기면 중간에 끊길 때 **안 보낸 사실이 통째로 사라진다**."""
        from app.modules.pipeline import grounding_claims_search as gcs

        got = []
        gcs.search_claims(
            None, [{"research_subject_id": "rs_a", "surface_form": "가",
                    "owner_type": "prop", "source_quote": "문장" * 50_000}],
            model="m", era="1983년", region="대한민국", batch_size=1,
            max_prompt_bytes=80, hard_prompt_bytes=100,
            on_batch=got.append)
        assert len(got) == 1 and got[0]["not_sent"] is True

    def test_it_is_not_counted_as_bought(self):
        """★★안 보낸 행을 「산 것」으로 세면 보고가 **유료 1회**로 오독된다."""
        from app.modules.pipeline import grounding_claims_search as gcs

        out = gcs.search_claims(
            None, [{"research_subject_id": "rs_a", "surface_form": "가",
                    "owner_type": "prop", "source_quote": "문장" * 50_000}],
            model="m", era="1983년", region="대한민국", batch_size=1,
            max_prompt_bytes=80, hard_prompt_bytes=100)
        assert out["logical_calls"] == 1, "계획은 한 호출이 맞다"
        assert out["not_sent_calls"] == 1
        assert out["bought_calls"] == 0, "안 보냈는데 샀다고 셌다"


class TestAHardLimitRowCanBeAdmittedAgain:
    """★★★상한에 걸린 것이 **다음 주행에 다시 들어가는지** 실제
    `plan_admission` 으로 확인한다 (Codex).

    안 들어가면 그 대상은 **영원히 조사 안 된 채** 남는다.
    """

    def test_a_retryable_row_is_admitted(self):
        from app.modules.pipeline.grounding_claims import (plan_admission,
                                                           research_input_hash,
                                                           subject_payload_hash)

        subj = _subj("rs_a")
        step = _step()
        ri = step._research_inputs(era="1983년", region="대한민국")
        sph = subject_payload_hash(subj)
        ih = research_input_hash(research_subject_id="rs_a",
                                 subject_payload_hash=sph, **ri)
        out = plan_admission(
            ["rs_a"], subject_payload_hashes={"rs_a": sph},
            research_inputs=ri,
            done_rows=[{"research_subject_id": "rs_a",
                        "research_input_hash": ih, "status": "retryable"}],
            batch_size=1)
        assert out["admitted"] == ["rs_a"], "상한에 걸린 것이 다시 안 들어간다"
        assert out["already_done"] == []

    def test_a_completed_row_is_not_admitted(self):
        """★positive control — 다 들여보내면 재구매 방지가 없는 것과 같다."""
        from app.modules.pipeline.grounding_claims import (plan_admission,
                                                           research_input_hash,
                                                           subject_payload_hash)

        subj = _subj("rs_a")
        ri = _step()._research_inputs(era="1983년", region="대한민국")
        sph = subject_payload_hash(subj)
        ih = research_input_hash(research_subject_id="rs_a",
                                 subject_payload_hash=sph, **ri)
        out = plan_admission(
            ["rs_a"], subject_payload_hashes={"rs_a": sph},
            research_inputs=ri,
            done_rows=[{"research_subject_id": "rs_a",
                        "research_input_hash": ih, "status": "completed"}],
            batch_size=1)
        assert out["admitted"] == [] and out["already_done"] == ["rs_a"]