"""B-only 주행기의 **문**들. ★유료 0 — `send` 를 안 주면 아무것도 안 나간다.

★**끝점에서 잰다.** 「상한을 세는 함수」를 따로 부르지 않고 실제 `run()` 을
돌려 **`send` 가 몇 번 불렸는지**를 본다.
"""
from __future__ import annotations

import json
from pathlib import Path

import pytest

from tools.grounding_audit import cc_runner as rr

WORLD = "가상의 근대 이후 어느 시기"


class Sender:
    """부른 횟수와 신원을 남기는 `send`. ★provider 를 안 탄다."""

    def __init__(self, fail_at=None):
        self.calls = []
        self.fail_at = fail_at

    def __call__(self, payload, ident):
        self.calls.append(ident)
        if self.fail_at is not None and len(self.calls) == self.fail_at:
            raise RuntimeError("provider 가 끊겼다")
        return {"rows": [], "decisions": []}


def _run(tmp_path, send, **kw):
    kw.setdefault("dispatch_budget", rr.APPROVED_LOGICAL)
    return rr.run(tmp_path / "j.json", send, world_facts=WORLD,
                  approved_slots=None, **kw)


def _journal(tmp_path):
    return json.loads((tmp_path / "j.json").read_text(encoding="utf-8"))


class TestTheCapStandsBeforeTheSend:
    def test_the_send_never_fires_past_the_logical_cap(self, tmp_path):
        s = Sender()
        with pytest.raises(rr.CapExceeded):
            _run(tmp_path, s, cap=2)
        assert len(s.calls) == 2, f"★상한 2인데 {len(s.calls)}번 보냈다"

    def test_the_approved_cap_matches_the_plan_and_is_not_derived(self):
        """★★승인 수를 fixture 에서 **끌어오지 않는다** — 끌어오면 원고가
        길어질 때 승인 범위가 조용히 넓어진다. 대신 여기서 **대조**한다."""
        from tests.grounding.fixtures import synthetic_episode as ep

        assert rr.APPROVED_LOGICAL == len(ep.bundles()) + 1, (
            f"★계획은 구간 {len(ep.bundles())} + merge 1 인데 승인 수는 "
            f"{rr.APPROVED_LOGICAL} 다 — 사람이 다시 정해야 한다")

    def test_the_full_plan_fits_the_approved_cap(self, tmp_path):
        """★positive control — 상한이 너무 좁으면 위 시험은 공짜다."""
        s = Sender()
        got = _run(tmp_path, s, cap=rr.APPROVED_LOGICAL)
        assert len(s.calls) == rr.APPROVED_LOGICAL == got["bought"]


class TestThePhysicalBudgetIsSeparateFromTheLogicalCap:
    """★★논리 상한만으로는 안 막힌다 — **나간 횟수**를 따로 세야 한다."""

    def test_the_budget_stops_the_send_before_it_fires(self, tmp_path):
        s = Sender()
        with pytest.raises(rr.CapExceeded, match="dispatch 예산"):
            _run(tmp_path, s, cap=9, dispatch_budget=2)
        assert len(s.calls) == 2

    def test_the_budget_counts_the_journal_not_just_this_run(self, tmp_path):
        """★★예산은 **장부 전체**로 센다. 이 판만 세면 판을 나누는 것만으로
        예산이 무한이 된다.

        ★한 판 앞서 나는 여기에 `assert False not in [True]` 를 적었다 —
        **아무것도 안 재면서 초록**이다. 장부에 앞 판 흔적 4건을 미리 넣고
        예산 5로 돌리면, 이 판은 **1건만** 보낼 수 있어야 한다.
        (시험용 손잡이를 코드에 넣지 않으려고 장부 쪽에서 짠다.)
        """
        s0 = Sender()
        first = _run(tmp_path, s0)               # 잠금을 제대로 만들어 둔다
        j = _journal(tmp_path)
        j["calls"] = [{"identity": f"앞판{i}", "status": "ok",
                       "response": {"rows": [], "decisions": []}}
                      for i in range(first["lock"]["dispatch_budget"] - 1)]
        (tmp_path / "j.json").write_text(
            json.dumps(j, ensure_ascii=False), encoding="utf-8")

        from tools.grounding_audit.cc_runner import Journal

        assert Journal(tmp_path / "j.json").dispatch_count() \
            == first["lock"]["dispatch_budget"] - 1
        s2 = Sender()
        with pytest.raises(rr.CapExceeded, match="dispatch 예산"):
            rr.run(tmp_path / "j.json", s2, world_facts=WORLD,
                   approved_slots=None, cap=first["lock"]["logical_cap"],
                   dispatch_budget=first["lock"]["dispatch_budget"])
        assert len(s2.calls) == 1, (
            f"★앞 판 것을 안 세어 {len(s2.calls)}건이 더 나갔다")

    def test_the_names_say_what_is_actually_counted(self):
        """★★우리가 세는 것은 **dispatch**(=`call_structured` 호출)다.

        물리 시도는 그 안에서 키 슬롯 loop 로 최대 슬롯 배까지 간다
        (`llm_client.py:427-439`, Router **밖**이라 `num_retries=0` 이 안 닫는다).
        이름이 「물리 예산」이면 세는 것과 이름이 어긋난다 (Codex NON-BLOCK).
        """
        assert rr.dispatch_budget_for(5) == 5
        assert rr.physical_upper_bound(5, rr.APPROVED_SLOTS) == 10


class TestTheSlotCountIsCheckedBeforeAnything:
    def test_a_different_slot_count_stops_with_zero_sends(
            self, tmp_path, monkeypatch):
        from app.core import openai_keys

        monkeypatch.setattr(openai_keys, "slot_count", lambda: 7)
        s = Sender()
        with pytest.raises(rr.SlotMismatch):
            rr.run(tmp_path / "j.json", s, world_facts=WORLD, approved_slots=2)
        assert s.calls == [], "★슬롯이 다른데 보냈다"

    def test_the_matching_slot_count_lets_it_through(
            self, tmp_path, monkeypatch):
        from app.core import openai_keys

        monkeypatch.setattr(openai_keys, "slot_count", lambda: 2)
        s = Sender()
        rr.run(tmp_path / "j.json", s, world_facts=WORLD, approved_slots=2)
        assert len(s.calls) == rr.APPROVED_LOGICAL


class TestAnUncertainSendIsNotBoughtAgain:
    """★★★「보냈는데 답을 못 받은 것」은 **샀을 수도 있다.**

    앞 판 주행기는 실패한 호출을 장부에 안 남겨서, 재개하면 같은 것을 **다시
    보냈다** — 총 전송이 계획 5를 넘고 물리 상한 10도 넘을 수 있었다 (Codex).
    """

    def test_the_failed_send_is_journaled_immediately(self, tmp_path):
        with pytest.raises(RuntimeError):
            _run(tmp_path, Sender(fail_at=2))
        calls = _journal(tmp_path)["calls"]
        assert len(calls) == 2, f"★나간 2건이 다 안 남았다 ({len(calls)})"
        un = [c for c in calls if c["status"] == "uncertain"]
        assert len(un) == 1 and un[0].get("error")

    def test_resuming_refuses_to_re_buy_it(self, tmp_path):
        with pytest.raises(RuntimeError):
            _run(tmp_path, Sender(fail_at=2))
        s2 = Sender()
        with pytest.raises(rr.NeedsDecision):
            _run(tmp_path, s2)
        assert s2.calls == [], "★불확실한 것을 자동으로 다시 샀다"

    def test_a_clean_run_resumes_without_buying(self, tmp_path):
        """★positive control — 막기만 하고 **이어 가는 길**이 없으면 못 쓴다."""
        _run(tmp_path, Sender())
        got = _run(tmp_path, Sender())
        assert (got["bought"], got["reused"]) == (0, rr.APPROVED_LOGICAL)

    def test_the_sent_count_includes_the_uncertain_one(self, tmp_path):
        with pytest.raises(RuntimeError):
            _run(tmp_path, Sender(fail_at=2))
        from tools.grounding_audit.cc_runner import Journal

        assert Journal(tmp_path / "j.json").dispatch_count() == 2


class TestTheExperimentLockStopsDrift:
    """★앞 판은 `PINNED` 가 preflight **문구에만** 있고 주행기가 안 썼다."""

    def test_the_lock_records_what_the_run_was_bought_under(self, tmp_path):
        got = _run(tmp_path, Sender())
        for k in ("pack_version", "pack_hash", "world_hash", "model_alias",
                  "model_physical", "request_contract", "slots",
                  "logical_cap", "dispatch_budget", "physical_upper_bound",
                  "opik_trace_v2"):
            assert k in got["lock"], f"★잠금에 {k} 가 없다"
        assert _journal(tmp_path)["lock"] == got["lock"]

    def test_a_changed_world_stops_with_zero_sends(self, tmp_path):
        _run(tmp_path, Sender())
        s = Sender()
        with pytest.raises(rr.LockDrift):
            rr.run(tmp_path / "j.json", s, world_facts="다른 세계관",
                   approved_slots=None, dispatch_budget=5)
        assert s.calls == []

    def test_a_changed_model_stops_with_zero_sends(
            self, tmp_path, monkeypatch):
        _run(tmp_path, Sender())
        monkeypatch.setattr(rr, "physical_model", lambda alias=None: "다른-모델")
        s = Sender()
        with pytest.raises(rr.LockDrift):
            _run(tmp_path, s)
        assert s.calls == []

    def test_a_changed_cap_stops_with_zero_sends(self, tmp_path):
        _run(tmp_path, Sender(), cap=5)
        s = Sender()
        with pytest.raises(rr.LockDrift):
            _run(tmp_path, s, cap=4)
        assert s.calls == []


class TestIdentityIsPayloadPlusModelPlusContract:
    def test_the_same_payload_on_a_different_model_is_not_reused(self):
        a = rr.identity("s", "u", {}, "gpt", "물리-A")
        b = rr.identity("s", "u", {}, "gpt", "물리-B")
        assert a != b

    def test_the_request_contract_is_folded_in(self):
        """★재시도를 켠 판과 끈 판이 **같은 신원**이면 안 된다."""
        a = rr.identity("s", "u", {}, "gpt", "p", {"num_retries": 0})
        b = rr.identity("s", "u", {}, "gpt", "p", {"num_retries": 3})
        assert a != b

    def test_the_schema_is_folded_in(self):
        assert rr.identity("s", "u", {"type": "object"}, "g", "p") \
            != rr.identity("s", "u", {"type": "array"}, "g", "p")

    def test_every_call_in_one_run_has_its_own_identity(self, tmp_path):
        s = Sender()
        _run(tmp_path, s)
        assert len(set(s.calls)) == len(s.calls), "★신원이 겹쳤다"


class TestNothingLeavesWithoutASender:
    def test_the_dry_send_makes_nothing_up(self):
        """★그럴듯한 답을 지어내면 재는 것은 **내 상상**이지 모델이 아니다."""
        assert rr._dry_send({"system": "", "parts": [{"text": ""}],
                             "schema": {}}, "i") == {"rows": [],
                                                     "decisions": []}

    def test_the_live_sender_consumes_the_pinned_contract(self, monkeypatch):
        """★★★`PINNED` 이 **문구에만** 있으면 아무것도 안 잠근다 (Codex).

        실제로 `call_structured` 에 `num_retries`·`enable_fallback` 이
        **인자로** 넘어가는지 끝점에서 본다.
        """
        seen = {}

        def fake(*a, **kw):
            seen.update(kw)
            return {"rows": []}

        import app.modules.llm.llm_client as lc
        import app.modules.llm.opik_trace as ot

        # ★부모 trace 가 없으면 이제 **안 산다** — 열린 것으로 세운다
        monkeypatch.setattr(
            ot, "open_trace",
            lambda **_k: __import__("contextlib").nullcontext(object()))

        monkeypatch.setattr(lc, "call_structured", fake)
        rr.live_send({"system": "s", "parts": [{"text": "u"}],
                      "schema": {"type": "object"}}, "ident-1")
        assert seen["num_retries"] == rr.PINNED["num_retries"] == 0
        assert seen["enable_fallback"] is rr.PINNED["enable_fallback"] is False

    def test_the_identity_cannot_ride_a_tag_and_we_prove_it(self):
        """★★★**끝점.** 신원을 tag 로 실으면 **걸러진다** — 두 번 막힌다.

        ① litellm 이 `metadata["opik"]` 에서 읽는 것은 넷뿐이라 자유 키가
           버려진다(실측: trace 에 `tags: ['gemini','op:audit']` 만 남았다).
        ② tag 로 옮겨도 **축 whitelist** 가 거른다. 그 문은 고카디널리티
           태그를 막으려고 있고, 호출 신원이 정확히 그것이다 — 넓히면
           안 되는 가드다. 그래서 부모 trace 의 metadata 로 간다.
        """
        from app.modules.llm.llm_client import _build_opik_metadata

        built = _build_opik_metadata(
            rr.LIVE_STEP, {"tags": [rr.LIVE_TAG, "cc-id:ident-1"]})
        tags = built["opik"]["tags"]
        assert rr.LIVE_TAG in tags, f"★실험 표식마저 사라졌다: {tags}"
        assert "cc-id:ident-1" not in tags, (
            "★신원 tag 가 통과했다 — 축 가드가 느슨해졌다는 뜻이다")

    def test_a_free_form_metadata_key_also_would_not_have_survived(self):
        """★positive control — 첫 판 방식(자유 키)도 실제로 못 간다."""
        from app.modules.llm.llm_client import _build_opik_metadata

        built = _build_opik_metadata(
            rr.LIVE_STEP, {"tag": rr.LIVE_TAG, "call_identity": "ident-1"})
        assert rr.LIVE_TAG not in built["opik"]["tags"]

    def test_the_identity_rides_the_parent_trace_metadata(self, monkeypatch):
        """★그래서 `open_trace` 로 부모를 열고 **그 metadata** 에 싣는다."""
        seen, bought = {}, []

        def fake_open(**kw):
            from contextlib import nullcontext

            seen.update(kw)
            return nullcontext(object())        # ★열렸다

        import app.modules.llm.llm_client as lc
        import app.modules.llm.opik_trace as ot

        monkeypatch.setattr(ot, "open_trace", fake_open)
        monkeypatch.setattr(lc, "call_structured",
                            lambda *a, **k: bought.append(1) or {"rows": []})
        rr.live_send({"system": "s", "parts": [{"text": "u"}],
                      "schema": {}}, "ident-1")
        assert seen["metadata"][rr.ID_META_KEY] == "ident-1"
        assert seen["tags"] == [rr.LIVE_TAG]
        assert seen["thread_id"] == rr.LIVE_THREAD
        assert len(bought) == 1

    def test_no_parent_trace_means_no_purchase(self, monkeypatch):
        """★★★**기록을 못 남기면 안 산다.**

        `open_trace` 는 설정이 꺼졌거나 client 생성이 실패하면 **예외 없이
        None** 을 준다 — production 에서는 그게 맞다(기록이 본 작업을 막으면
        안 된다). 그런데 이 실험은 **기록이 목적**이라 반대다.

        ★한 판 앞서 내 시험은 `nullcontext(None)` 을 주고도 provider fake 가
        불리는 것을 **초록으로 잠갔다** — 틀린 동작을 못박은 것이다 (Codex).
        """
        bought = []

        def fake_open(**_kw):
            from contextlib import nullcontext

            return nullcontext(None)            # ★못 열렸다

        import app.modules.llm.llm_client as lc
        import app.modules.llm.opik_trace as ot

        monkeypatch.setattr(ot, "open_trace", fake_open)
        monkeypatch.setattr(lc, "call_structured",
                            lambda *a, **k: bought.append(1) or {"rows": []})
        with pytest.raises(rr.TraceUnavailable):
            rr.live_send({"system": "s", "parts": [{"text": "u"}],
                          "schema": {}}, "ident-1")
        assert bought == [], "★기록 없이 샀다"

    def test_the_lock_folds_whether_tracing_is_even_on(self):
        """★설정이 꺼져 있으면 **장부에 구매 줄을 적기 전에** 걸린다."""
        lock = rr.experiment_lock("w", 2, 5, 5)
        assert "opik_trace_v2" in lock

    def test_tracing_turned_off_drifts_the_lock(self, tmp_path, monkeypatch):
        _run(tmp_path, Sender())
        from app.core.config import settings

        cur = bool(getattr(settings, "opik_trace_v2_enabled", False))
        monkeypatch.setattr(settings, "opik_trace_v2_enabled", not cur)
        s = Sender()
        with pytest.raises(rr.LockDrift):
            _run(tmp_path, s)
        assert s.calls == []

    def test_the_live_sender_asks_for_the_approved_model(self, monkeypatch):
        seen = {}

        def fake(*a, **kw):
            seen.update(kw)
            return {"rows": []}

        import app.modules.llm.llm_client as lc
        import app.modules.llm.opik_trace as ot

        # ★부모 trace 가 없으면 이제 **안 산다** — 열린 것으로 세운다
        monkeypatch.setattr(
            ot, "open_trace",
            lambda **_k: __import__("contextlib").nullcontext(object()))

        monkeypatch.setattr(lc, "call_structured", fake)
        rr.live_send({"system": "s", "parts": [{"text": "u"}],
                      "schema": {}}, "i")
        assert seen["project_config"][rr.LIVE_STEP]["model"] == rr.MODEL_ALIAS


@pytest.fixture
def dry_main():
    """`main()` 을 부르되 **전역 문을 반드시 되돌린다**.

    ★`seal_outbound()` 는 프로세스 전역이라, 시험 안에서 한 번 열면 그 뒤
    모든 시험의 소켓이 막힌다 — 실제로 이 시험이 다른 파일 네 건을 깨뜨렸다.
    """
    import sys

    from tools.grounding_audit.call_payload_table import unseal_outbound

    def go(path):
        old = sys.argv
        try:
            sys.argv = ["x", "--dry", str(path)]
            return rr.main()
        finally:
            sys.argv = old
            unseal_outbound()

    yield go
    unseal_outbound()


class TestTheRunIsScoredNotJustSurvived:
    """★★★앞 판 `--live` 는 줄인 counts 만 찍고 채점기를 **안 불렀다** —
    전부 미확정이어도 `exit 0` 이었다. 「샀고 안 죽었다」가 「통과」로 읽힌다.
    """

    def test_an_empty_run_is_scored_as_failing(self, tmp_path, dry_main):
        assert dry_main(tmp_path / "j.json") == 1, "★빈 산출인데 통과로 끝났다"

    def test_the_review_artifact_carries_the_basis_for_the_two_axes(
            self, tmp_path, dry_main):
        dry_main(tmp_path / "j.json")
        d = json.loads((tmp_path / "j_score.json").read_text(encoding="utf-8"))
        hr = d["human_review"]
        assert d["score"]["final_candidate"] is None
        assert hr["required"] is True and hr["verdict"] is None
        assert d["lock"]["pack_version"]
        # ★★판정 대상 **전부**가 올라와야 한다 — 기계가 「맞다」고 한 것도.
        from tests.grounding.fixtures import synthetic_episode as ep

        got = {x["target"] for x in hr["two_axis_judgements"]}
        assert got == set(ep.AXIS_BASIS), f"★빠진 축이 있다: {got}"
        for x in hr["two_axis_judgements"]:
            assert x["basis"].strip() and x["human_verdict"] is None
            assert "expected_hard" in x and "expected_notice" in x

    def test_the_world_facts_come_from_the_fixture(self):
        """★세계관을 주행기에 적으면 두 벌이 된다 — 두 축의 정답 근거와
        실제 입력이 갈린다 (Codex BLOCK-1)."""
        from tests.grounding.fixtures import synthetic_episode as ep

        calls = rr.plan_calls(ep.WORLD_FACTS)
        assert ep.WORLD_FACTS.strip() in calls[0]["payload"]["parts"][0]["text"]

    def test_the_world_facts_give_a_basis_for_every_judged_target(self):
        """★근거 없는 표적이 있으면 그 축의 정답을 **사람도 못 정한다**."""
        from tests.grounding.fixtures import synthetic_episode as ep

        judged = [t["key"] for t in ep.EXPECTED_TARGETS
                  if t.get("exception_axis")]
        for k in judged:
            assert k in ep.AXIS_BASIS, f"★{k} 의 두 축 근거가 없다"
            assert ep.AXIS_BASIS[k]["basis"].strip()


class TestTheOutboundSealIsReversible:
    """★★전역 문을 못 되돌리면 **한 번 연 시험이 뒤의 모든 시험을 막는다.**

    실제로 그랬다 — 이 파일이 `test_grounding_claims_support.py` 4건을
    깨뜨렸다(2026-08-31). 도구 경로는 안 되돌리고, 시험만 되돌린다.
    """

    def test_seal_then_unseal_restores_the_socket(self):
        import socket

        from tools.grounding_audit.call_payload_table import (seal_outbound,
                                                              unseal_outbound)
        before = socket.socket.connect
        seal_outbound()
        assert socket.socket.connect is not before, "★안 잠겼다"
        unseal_outbound()
        assert socket.socket.connect is before, "★안 풀렸다"

    def test_unseal_without_seal_is_harmless(self):
        from tools.grounding_audit.call_payload_table import unseal_outbound

        unseal_outbound()
        unseal_outbound()


class TestExitCodeSeparatesCliSuccessFromAcceptance:
    """★★★「돌았고 안 죽었다」가 「통과」로 읽히면 안 된다 (Codex).

    자동은 `final_candidate` 를 **못 만든다**. 그래서 `0` 은 사람 판정이
    적힌 뒤에만 나야 한다.
    """

    def test_machine_fail_is_one(self):
        assert rr._exit_code({"mechanical_candidate": False,
                              "final_candidate": None}) \
            == rr.EXIT_MACHINE_FAIL

    def test_machine_pass_without_a_human_is_not_zero(self):
        got = rr._exit_code({"mechanical_candidate": True,
                             "final_candidate": None})
        assert got == rr.EXIT_HUMAN_REVIEW_REQUIRED != rr.EXIT_PASS

    def test_zero_needs_a_human_verdict(self):
        assert rr._exit_code({"mechanical_candidate": True,
                              "final_candidate": True}) == rr.EXIT_PASS

    def test_the_scorer_never_produces_a_final_candidate(self):
        """★자동 경로에서 `final_candidate` 가 나면 위 문이 무너진다."""
        import inspect

        from tools.grounding_audit import cc_scorer as sc

        src = inspect.getsource(sc.score)
        assert '"final_candidate": None' in src


class TestTracingOffStopsBeforeTheJournalRecordsAPurchase:
    """★sender 에서 서면 장부에 `uncertain` 이 남아 「샀을지도 모른다」로
    읽힌다 — 실제로는 안 샀다 (Codex NON-BLOCK)."""

    def test_no_journal_purchase_row_when_tracing_is_off(
            self, tmp_path, monkeypatch):
        from app.core.config import settings

        monkeypatch.setattr(settings, "opik_trace_v2_enabled", False)
        s = Sender()
        with pytest.raises(rr.TraceUnavailable):
            _run(tmp_path, s)
        assert s.calls == []
        d = json.loads((tmp_path / "j.json").read_text(encoding="utf-8"))
        assert d["calls"] == [], f"★안 샀는데 장부에 남았다: {d['calls']}"


class TestQuarantineIsRowLevelAndContributesNothing:
    """★★★Codex 가 지정한 여덟 끝점 (2026-08-31) — 2026-09-02 실측으로 ①만 뒤집었다.

    유료 주행이 merge 직전에 섰다 — 모델이 인용 하나를 **다듬어 썼다**
    (「나무 고리가 하나 달려 있다」 → 「나무 고리 하나」). 185분의 2였다.
    앞 판은 그래서 **행째** 격리했는데, 다음 두 유료 주행에서 22행 중 9행 · 29행 중
    12행이 「언급 번호 하나 초과」·「씬 밖 샷 결속 하나」로 통째로 사라져 장소 셋·
    주인공·부분 둘이 등록을 못 했다. 이제 **검증된 자리만 남기고 뺀 것을 적는다**.
    지어낸 인용은 여전히 한 자리도 안 남고(원문에서 찾은 자리만 남는다), 검증된
    언급이 하나도 없거나 근거 문장을 지어낸 행은 지금처럼 행째 격리다.
    """

    def _segs(self):
        from tests.grounding.fixtures import synthetic_episode as ep

        return ep.segment_texts()

    def _row(self, quote, evidence=(), surface="x"):
        return {"owner_type": "prop", "surface_form": surface,
                "mentions": [{"mention_quote": quote, "occurrence_index": 1}],
                "evidence_quotes": list(evidence),
                "hard_to_generate": True, "viewers_would_notice": True,
                "visual_brief": "", "search_terms_native": ["가", "나"],
                "language_lock_native": "어느 말로"}

    def _mixed(self):
        """언급 둘 중 **하나만** 틀린 행 — 일부 살리기의 유혹이 있는 자리."""
        return {"owner_type": "prop", "surface_form": "섞인 행",
                "mentions": [{"mention_quote": "가방", "occurrence_index": 1},
                             {"mention_quote": "원고에 없는 말",
                              "occurrence_index": 1}],
                "evidence_quotes": [], "hard_to_generate": True,
                "viewers_would_notice": True, "visual_brief": "",
                "search_terms_native": [], "language_lock_native": ""}

    def _resolve(self, rows):
        from app.modules.pipeline import grounding_chunk as gc

        return gc.resolve_rows(rows, chunk_id="c0",
                               segment_ids=["scene-1", "scene-2"],
                               segments=self._segs())

    # ① 한 언급만 틀리면 **그 언급만** 빠지고 사유가 행에 남는다 (2026-09-02 뒤집음)
    def test_one_invalid_mention_is_dropped_and_recorded(self):
        from app.modules.pipeline import grounding_chunk as gc

        got = self._resolve([self._mixed()])
        assert got["quarantined"] == []
        r = got["rows"][0]
        assert [o["source_quote"] for o in r["occurrences"]] == ["가방"], "★지어낸 인용이 자리를 받았다"
        assert r["salvage_problems"][0]["kind"] == gc.Q_MENTION
        assert r["salvage_problems"][0]["quote"] == "원고에 없는 말"

    # ①-b 검증된 언급이 **하나도** 없으면 여전히 행째 격리
    def test_no_verified_mention_still_quarantines(self):
        from app.modules.pipeline import grounding_chunk as gc

        got = self._resolve([self._row("원고에 없는 말")])
        assert got["rows"] == [] and len(got["quarantined"]) == 1
        assert got["quarantined"][0]["problems"][0]["kind"] == gc.Q_MENTION

    # ② 전부 틀린 행도 **명시 격리** · 조용한 삭제 0
    def test_an_all_invalid_row_is_recorded_not_dropped(self):
        got = self._resolve([self._row("원고에 없는 말")])
        assert got["rows"] == [] and len(got["quarantined"]) == 1
        q = got["quarantined"][0]
        for k in ("chunk_id", "row_index", "local_id", "surface_form",
                  "problems", "raw_mentions", "processing_contract"):
            assert k in q, f"★장부에 {k} 가 없다"

    # ③ 격리 행은 merge·출현·등록·최종 ID·자동 합격에 기여 0
    def test_a_quarantined_row_touches_nothing_downstream(self):
        from app.modules.pipeline import grounding_chunk as gc
        from app.modules.pipeline import grounding_chunk_merge as cm

        got = self._resolve([self._row("가방", surface="깨끗"),
                             self._row("원고에 없는 말", surface="섞인 행")])
        assert len(got["rows"]) == 1
        sent = gc.build_merge_payload(got["rows"])["parts"][0]["text"]
        assert "섞인 행" not in sent, "★격리 행이 merge payload 에 실렸다"
        red = cm.reduce_episode(got["rows"], [], segments=self._segs())
        assert "c0#1" not in red["registered"], "★격리 행이 등록 장부에 있다"
        assert "c0#1" not in red["final_ids"], "★격리 행이 번호를 받았다"

    # ④ 격리 0인 판은 **비회귀**
    def test_a_clean_run_is_unchanged(self):
        got = self._resolve([self._row("가방", surface="깨끗")])
        assert got["quarantined"] == []
        r = got["rows"][0]
        assert r["local_id"] == "c0#0" and len(r["occurrences"]) == 1
        assert r["search_terms_native"] == ["가", "나"]

    # ⑤ 근거가 틀린 행의 두 축·의무는 살아남지 못한다
    def test_bad_evidence_kills_the_rows_axes_and_obligation(self):
        got = self._resolve([self._row("가방", evidence=["원고에 없는 근거"])])
        assert got["rows"] == [], "★근거가 틀린데 두 축이 살아남았다"
        from app.modules.pipeline import grounding_chunk as gc

        assert got["quarantined"][0]["problems"][0]["kind"] == gc.Q_EVIDENCE

    # ⑦ 채점은 미확정 · 최종 없음
    def test_the_score_marks_it_unresolved(self):
        from tests.grounding.fixtures import synthetic_episode as ep
        from tools.grounding_audit import cc_scorer as sc

        got = sc.score(ep.EXPECTED_TARGETS, ep.target_spans(), [],
                       segments=self._segs(), relations=[], registered={},
                       quarantined=[{"local_id": "c0#1"}])
        assert not got["mechanical_candidate"]
        assert got["final_candidate"] is None
        assert "mention_resolution" in got["needs_human"]

    def test_no_quarantine_leaves_that_axis_ok(self):
        """★positive control — 격리가 없으면 그 축은 통과여야 한다."""
        from tests.grounding.fixtures import synthetic_episode as ep
        from tools.grounding_audit import cc_scorer as sc

        got = sc.score(ep.EXPECTED_TARGETS, ep.target_spans(), [],
                       segments=self._segs(), relations=[], registered={},
                       quarantined=[])
        v = [f for f in got["findings"]
             if f["axis"] == "mention_resolution"][0]
        assert v["verdict"] == sc.OK

    # ⑧ 후처리 계약은 파생 장부에만 · 획득 신원엔 안 접힌다
    def test_the_processing_contract_never_touches_acquisition_identity(self):
        from app.modules.pipeline import grounding_chunk as gc

        a = rr.identity("s", "u", {}, "gpt", "p")
        assert gc.PROCESSING_CONTRACT_VERSION not in a
        got = self._resolve([self._row("가방")])
        assert got["processing_contract"] == gc.PROCESSING_CONTRACT_VERSION


class TestTheFrozenPaidRunBelongsToTheOldContract:
    """★★★**뒤집힌 시험** — 얼어붙은 유료 판은 **새 계약으로 못 쓴다**.

    그 판은 팩 2.x · **씬 단위** 반복축으로 돌았다. 지금 계약은 팩 3.x ·
    **샷 단위**다. Codex 가 못박은 대로 「지금 3회는 새 shot-aware 요청의
    재사용 근거가 아니다」 — 그래서 잠금이 **어긋나야 맞다**.

    앞 판 시험은 「재사용 2 · 신규 1」을 잠갔는데, 그건 **그때 계약**의 이야기다.
    계약이 바뀐 뒤에도 그게 통과하면 **두 계약이 섞인 것**이다.
    """

    PAID = Path(__file__).resolve().parents[3] \
        / "artifact" / "20260831_cc_preflight" / "live2_j.json"

    def test_the_saved_lock_no_longer_matches(self):
        from app.core import openai_keys
        from tests.grounding.fixtures import synthetic_episode as ep

        if not self.PAID.exists():
            pytest.skip("얼어붙은 유료 장부가 없다")
        saved = json.loads(self.PAID.read_text(encoding="utf-8"))["lock"]
        now = rr.experiment_lock(ep.WORLD_FACTS, openai_keys.slot_count(),
                                 rr.APPROVED_LOGICAL,
                                 rr.dispatch_budget_for(rr.APPROVED_LOGICAL))
        assert saved.get("pack_version") != now.get("pack_version"), (
            "★팩이 그대로다 — 샷 계약이 안 들어갔거나 두 계약이 섞였다")

    def test_resuming_it_stops_instead_of_mixing_contracts(self, tmp_path,
                                                           monkeypatch):
        """★섞어 쓰려 하면 **provider 0회로 선다**."""
        import shutil

        from app.core import openai_keys
        from tests.grounding.fixtures import synthetic_episode as ep

        if not self.PAID.exists():
            pytest.skip("얼어붙은 유료 장부가 없다")
        jp = tmp_path / "j.json"
        shutil.copy(self.PAID, jp)
        monkeypatch.setattr(openai_keys, "slot_count",
                            lambda: rr.APPROVED_SLOTS)
        s = Sender()
        with pytest.raises(rr.LockDrift):
            rr.run(jp, s, world_facts=ep.WORLD_FACTS,
                   approved_slots=rr.APPROVED_SLOTS,
                   cap=rr.APPROVED_LOGICAL,
                   dispatch_budget=rr.APPROVED_LOGICAL)
        assert s.calls == [], "★계약이 다른데 샀다"


class TestReplayCannotBuyAtAll:
    """★★재채점은 「안 샀다」를 **믿음이 아니라 구조**로 만든다 —
    `replay` 에는 sender 인자가 **없다**."""

    PAID = Path(__file__).resolve().parents[3] \
        / "artifact" / "20260831_cc_preflight" / "live2_j.json"

    def test_replay_takes_no_sender(self):
        import inspect

        sig = inspect.signature(rr.replay)
        # ★핵심은 **보낼 길이 없다**는 것이다. 계획·원문·샷 목록은 받아도
        #  되지만(합성과 실제가 같은 재생기를 써야 한다) sender 는 안 된다.
        for banned in ("send", "sender", "provider", "call"):
            assert banned not in sig.parameters, f"★보낼 길이 남았다: {banned}"
        assert "journal_path" in sig.parameters

    def test_replaying_an_old_contract_journal_stops(self):
        """★옛 계약 장부는 새 payload 와 신원이 달라 **찾지 못하고 선다** —
        몰래 사지 않는다."""
        from tests.grounding.fixtures import synthetic_episode as ep

        if not self.PAID.exists():
            pytest.skip("얼어붙은 유료 장부가 없다")
        with pytest.raises(LookupError, match="사지 않는다"):
            rr.replay(self.PAID, world_facts=ep.WORLD_FACTS)

    def test_a_missing_response_stops_instead_of_buying(self, tmp_path):
        """★장부에 없는 호출이 필요하면 **선다** — 몰래 사지 않는다."""
        from tests.grounding.fixtures import synthetic_episode as ep

        if not self.PAID.exists():
            pytest.skip("얼어붙은 유료 장부가 없다")
        d = json.loads(self.PAID.read_text(encoding="utf-8"))
        d["calls"] = d["calls"][:1]           # 하나를 뺀다
        jp = tmp_path / "j.json"
        jp.write_text(json.dumps(d, ensure_ascii=False), encoding="utf-8")
        with pytest.raises(LookupError, match="사지 않는다"):
            rr.replay(jp, world_facts=ep.WORLD_FACTS)

    def test_the_archived_rescore_is_kept_as_a_record(self):
        """★옛 계약의 재채점 산출은 **개발 기록으로 남는다** — 지우지 않는다."""
        arch = (Path(__file__).resolve().parents[3] / "artifact"
                / "20260831_cc_preflight"
                / "live2_score_posthoc_oracle_v2.json")
        if not arch.exists():
            pytest.skip("보관된 재채점본이 없다")
        d = json.loads(arch.read_text(encoding="utf-8"))
        assert d["provider_calls"] == 0
        assert d["score"]["final_candidate"] is None
        assert "post-hoc development diagnostic" in d["note"]


class TestStampFollowsTheContractItActuallyBoughtWith:
    """★★조회는 **장부에 적힌 계약**으로 하는데 지문만 **지금 `PINNED`** 으로
    찍던 것 (Codex NON-BLOCK, 2026-08-31).

    지금은 두 값이 같아서 안 드러난다. 계약을 바꾼 뒤 옛 장부를 다시 읽으면
    **다른 계약으로 산 응답에 새 계약 지문이 찍힌다** — D 의 하류 무효화가
    「같은 지문이니 그대로 두자」로 엉뚱한 것을 살려 둔다.

    ★그래서 **계약이 다른 장부**를 만들어 재생시킨다. 규칙을 두 곳에 적은 것이
    병이었으므로, 시험도 `run`/`replay` 를 갈라 보지 않고 **지문이 무엇을 따르나**
    하나만 본다.
    """

    OLD = {"num_retries": 2, "enable_fallback": True}   # ★지금과 다른 계약

    def _journal(self, tmp_path, payloads):
        """옛 계약으로 산 장부를 손으로 짓는다."""
        phys = rr.physical_model()
        calls = [{
            "identity": rr.identity(p["system"], p["parts"][0]["text"],
                                    p["schema"], rr.MODEL_ALIAS, phys,
                                    self.OLD),
            "status": "ok", "model_alias": rr.MODEL_ALIAS,
            "model_physical": phys, "bytes": 0, "response": r,
        } for p, r in payloads]
        jp = tmp_path / "old.json"
        jp.write_text(json.dumps({
            "lock": {"model_physical": phys, "request_contract": self.OLD},
            "nonce": "0" * 12, "calls": calls,
        }, ensure_ascii=False), encoding="utf-8")
        return jp, phys

    def test_replay_stamps_with_the_saved_contract_not_the_current_one(
            self, tmp_path):
        from app.modules.pipeline import grounding_chunk as gc

        segs = {"scene-1": "여기에 원문이 있다."}
        payload = gc.build_chunk_payload(["scene-1"], segs, "세계 규칙")
        plan = [{"chunk_id": "c0", "segment_ids": ["scene-1"],
                 "payload": payload}]
        jp, phys = self._journal(tmp_path, [
            (payload, {"rows": []}),
            (gc.build_merge_payload([]), {"decisions": []}),
        ])

        got = rr.replay(jp, world_facts="세계 규칙", plan=plan, segments=segs)

        assert got["processing_stamps"]["c0"] == \
            rr.stamp_of(payload, phys, self.OLD), \
            "★산 계약이 아니라 다른 계약으로 지문을 찍었다"
        assert got["processing_stamps"]["c0"] != \
            rr.stamp_of(payload, phys, rr.PINNED), \
            "★계약이 다른데 지문이 같다 — 지문이 계약을 안 접는다"

    def test_stamp_is_computed_in_exactly_one_place(self):
        """★같은 규칙을 두 곳에 적으면 한쪽만 고쳐진다 — 실제로 그랬다."""
        import ast
        import inspect

        src = inspect.getsource(rr)
        tree = ast.parse(src)
        hits = [n for n in ast.walk(tree)
                if isinstance(n, ast.Call)
                and isinstance(n.func, ast.Attribute)
                and n.func.attr == "processing_stamp"]
        assert len(hits) == 1, \
            f"★`processing_stamp` 을 부르는 자리가 {len(hits)} 곳이다 — 한 곳이어야"