"""앞쪽 **판별**(13.68) — 「무엇이 참조 사진을 사야 하는가」.

★이 단계는 **사지 않는다.** 검색·이미지 검색·VLM 호출 0. 사는 곳은 중앙 19.x
한 곳이다 — 두 곳에서 사면 같은 대상을 두 번 산다.

★잠그는 것 넷:

    ① 다섯 owner 가 **전부** 장부에 남는다 (facet 도 잃지 않는다)
    ② 「사야 하나」와 「어디 붙나」는 **다른 칸**이다
    ③ 미확정을 **비대상으로 안 내린다** — 하류가 막힌다
    ④ 정본 신원 셋(id·owner·payload sha)을 **다 보낸다**
"""
import pytest

from app.modules.pipeline import era_research as era
from app.modules.pipeline import grounding_screen as gs
from app.modules.pipeline.grounding_carry import (DISP_CARRIED, DISP_CONTESTED,
                                                  DISP_DEFERRED, DISP_PROMOTED)

OWNERS = ("prop", "character", "location", "location_part", "outlook")


def _subj(sid, owner="prop", **over):
    d = {"research_subject_id": sid, "owner_type": owner,
         "surface_form": f"표면-{sid}", "source_anchor": "A1",
         "source_quote": f"원문-{sid}"}
    d.update(over)
    return d


def _fake_assess(status, plan=None, seen=None, reason=None):
    def _call(**kw):
        if seen is not None:
            seen.append(kw)
        out = {"status": status, "plan": plan}
        if reason:
            out["reason"] = reason
        return out
    return _call


def _run(subjects, dispositions=None, **over):
    kw = dict(dispositions=dispositions or {}, world_facts_block="W",
              cache_get=lambda k: None, cache_put=lambda k, v: None,
              assess_fn=_fake_assess(era.PLAN_NO_SUBJECT))
    kw.update(over)
    return gs.screen_subjects(subjects, **kw)


class TestEveryOwnerStaysInTheLedger:
    """★①**다섯 갈래를 전부 적는다.** 잃으면 「다섯 owner 완료」가 거짓이다."""

    def test_all_five_owners_get_a_row(self):
        subs = [_subj(f"rs_{o}", o) for o in OWNERS]
        out = _run(subs)
        assert [r["owner_type"] for r in out["rows"]] == list(OWNERS)
        assert set(out["by_owner"]) == set(OWNERS)

    def test_the_population_keeps_every_owner_and_refuses_collisions(self):
        """★★`build_population` **끝점** — 조용히 버리는 자리가 없는지.

        앞 판은 빈 id·겹친 id 를 `continue` 로 넘겨서, 검사가 볼 때는 이미
        사라진 뒤였다 (Codex BLOCK-3).
        """
        base = [_subj(f"rs_{o}", o) for o in OWNERS]
        assert len(gs.build_population(base[:3], base, {})) == 5

    def test_a_blank_id_in_the_candidate_side_stops_it(self):
        with pytest.raises(AssertionError, match="candidate 에"):
            gs.build_population([], [_subj("")], {})

    def test_a_duplicate_inside_one_source_stops_it(self):
        with pytest.raises(AssertionError, match="candidate 안에"):
            gs.build_population([], [_subj("rs_a"), _subj("rs_a")], {})

    def test_the_same_id_pointing_at_a_different_thing_stops_it(self):
        """★앞 것을 택하면 뒤 후보가 **소리 없이 사라진다**."""
        with pytest.raises(AssertionError, match="다른 대상"):
            gs.build_population([_subj("rs_a", "prop")],
                                [_subj("rs_a", "location")], {})

    def test_the_same_id_pointing_at_the_same_thing_folds(self):
        one = _subj("rs_a")
        assert gs.build_population([one], [dict(one)], {}) == [one]

    def test_the_ledger_is_measured_by_id_not_by_count(self):
        """★개수만 세면 **항진식**이다 — 겹친 id 를 못 잡는다."""
        with pytest.raises(AssertionError, match="겹친다"):
            _run([_subj("rs_a"), _subj("rs_a", "location")])

    def test_two_blank_ids_do_not_cancel_each_other_out(self):
        """★집합·합계로 보면 **양쪽이 다 비었을 때 통과**한다 (Codex 재현)."""
        with pytest.raises(AssertionError, match="빈 것이"):
            _run([_subj(""), _subj("")])

    def test_a_facet_is_recorded_not_dropped(self):
        out = _run([_subj("rs_a", "location_part")],
                   {"rs_a": DISP_DEFERRED},
                   assess_fn=_fake_assess(era.PLAN_OK,
                                          {"subjects": [{}], "assess_sha": "s"}))
        row = out["rows"][0]
        assert row["screen"] == gs.SCREEN_OBLIGATION
        assert row["binding"] == gs.BIND_DEFERRED_PRODUCER
        assert row["is_facet"] is True


class TestScreenAndBindingAreTwoAxes:
    """★②한 칸에 접으면 「사야 하는데 붙을 데가 없다」를 적을 자리가 없다."""

    @pytest.mark.parametrize("disp,want", [
        (DISP_CARRIED, gs.BIND_ENTITY_ROW),
        (DISP_PROMOTED, gs.BIND_PROMOTED_ROW),
        (DISP_DEFERRED, gs.BIND_DEFERRED_PRODUCER),
        (DISP_CONTESTED, gs.BIND_UNBOUND),
        ("", gs.BIND_UNBOUND),
    ])
    def test_binding_comes_from_the_carry_ledger(self, disp, want):
        assert gs.binding_of("rs_a", {"rs_a": disp}) == want

    def test_an_obligation_with_no_row_is_what_gets_made(self):
        out = _run([_subj("rs_a")], {"rs_a": DISP_PROMOTED},
                   assess_fn=_fake_assess(era.PLAN_OK,
                                          {"subjects": [{}], "assess_sha": "s"}))
        assert gs.promotable_obligations(out) == {"rs_a"}

    def test_an_obligation_that_already_has_a_row_is_not_made(self):
        out = _run([_subj("rs_a")], {"rs_a": DISP_CARRIED},
                   assess_fn=_fake_assess(era.PLAN_OK,
                                          {"subjects": [{}], "assess_sha": "s"}))
        assert gs.promotable_obligations(out) == set()
        assert gs.obligation_ids(out) == {"rs_a"}

    def test_a_facet_obligation_is_not_promotable(self):
        out = _run([_subj("rs_a", "outlook")], {"rs_a": DISP_DEFERRED},
                   assess_fn=_fake_assess(era.PLAN_OK,
                                          {"subjects": [{}], "assess_sha": "s"}))
        assert gs.promotable_obligations(out) == set()
        assert [r["research_subject_id"] for r in gs.facet_obligations(out)] == ["rs_a"]


class TestUnresolvedIsNotNotTarget:
    """★③「못 섰다」를 「비대상」으로 읽으면 참조 없이 내려간다."""

    def test_a_failed_assess_is_unresolved(self):
        out = _run([_subj("rs_a")], assess_fn=_fake_assess(era.PLAN_FAILED,
                                                           reason="boom"))
        assert out["rows"][0]["screen"] == gs.SCREEN_UNRESOLVED
        assert gs.unresolved_ids(out) == {"rs_a"}
        assert gs.obligation_ids(out) == set()

    def test_an_empty_subject_list_is_not_target_and_is_not_a_failure(self):
        out = _run([_subj("rs_a")])
        assert out["rows"][0]["screen"] == gs.SCREEN_NOT_TARGET
        assert gs.unresolved_ids(out) == set()

    def test_an_unknown_status_falls_to_unresolved(self):
        """★모르는 값을 **비대상 쪽으로** 접으면 조용히 안 사게 된다."""
        out = _run([_subj("rs_a")], assess_fn=lambda **kw: {"status": "???"})
        assert out["rows"][0]["screen"] == gs.SCREEN_UNRESOLVED

    def test_a_none_envelope_is_unresolved(self):
        out = _run([_subj("rs_a")], assess_fn=lambda **kw: None)
        assert out["rows"][0]["screen"] == gs.SCREEN_UNRESOLVED

    def test_the_cap_is_capped_not_not_target(self):
        seen = []
        out = _run([_subj("rs_a"), _subj("rs_b")], max_assess=1,
                   assess_fn=_fake_assess(era.PLAN_NO_SUBJECT, seen=seen))
        assert len(seen) == 1, "★상한을 넘겨 사면 승인 범위가 아니다"
        assert out["rows"][1]["screen"] == gs.SCREEN_CAPPED
        assert gs.unresolved_ids(out) == {"rs_b"}

    def test_a_subject_with_no_id_stops_before_buying(self):
        """★**뒤집힌 계약** — 빈 id 는 줄로 남기지 않고 **사기 전에** 선다.

        줄로 남기면 「판별했는데 미확정」처럼 보이는데, 실제로는 producer 가
        신원을 안 준 것이라 고칠 자리가 다르다.
        """
        seen = []
        with pytest.raises(AssertionError, match="빈 것이"):
            _run([_subj("", "prop")],
                 assess_fn=_fake_assess(era.PLAN_NO_SUBJECT, seen=seen))
        assert seen == [], "★붙일 데를 모르는 것에 돈을 쓰지 않는다"

    def test_a_subject_with_no_text_is_not_bought(self):
        seen = []
        out = _run([_subj("rs_a", surface_form="", source_quote="")],
                   assess_fn=_fake_assess(era.PLAN_NO_SUBJECT, seen=seen))
        assert seen == []
        assert out["rows"][0]["reason"] == "no_subject_text"


class TestTheCapActuallyStopsSpending:
    """★★★**상한이 무엇을 막는지 끝점에서 잰다.**

    처음 판은 「캐시를 호출 **뒤에** 읽어」 판별이 방금 써 넣은 값을 적중으로
    셌다. 그래서 `bought` 가 영영 0 이었고 **상한이 아무것도 안 막았다** —
    상한 2 로 걸고 5개를 넣었더니 **5개를 다 샀다**.

    ★조립부(counts)만 보면 안 보인다. **판별이 provider 를 몇 번 탔는지**를
    센다.
    """

    @staticmethod
    def _cache_backed(store, calls):
        """★실제 `assess_plan_cached` 처럼 — 없으면 사고 **캐시에 넣는다**.

        ★키는 **프로덕션 함수**로 만든다. 손으로 조립하면 앞쪽 사전 확인이
        보는 키와 갈리고, 그러면 이 시험이 「공짜인 줄 알았는데 샀다」를
        못 잡는다 — 바로 그 결함을 재는 시험인데.
        """
        def _assess(**kw):
            k = era.assess_cache_key(
                subject_text=kw["subject_text"],
                world_facts_block=kw["world_facts_block"],
                canonical_scope_id=kw.get("canonical_scope_id"),
                canonical_scope_role=kw.get("canonical_scope_role"),
                canonical_scope_sha=kw.get("canonical_scope_sha"))
            if kw["cache_get"](k) is None:
                calls.append(k)
                kw["cache_put"](k, {"subjects": []})
            return {"status": era.PLAN_NO_SUBJECT, "plan": None}
        return _assess

    def _subs(self, n):
        return [_subj(f"rs{i}", surface_form=f"s{i}", source_quote=f"q{i}")
                for i in range(n)]

    def test_it_buys_no_more_than_the_cap(self):
        store, calls = {}, []
        out = _run(self._subs(5), max_assess=2,
                   cache_get=store.get, cache_put=store.__setitem__,
                   assess_fn=self._cache_backed(store, calls))
        assert len(calls) == 2, f"★상한 2 인데 {len(calls)}번 샀다"
        assert out["counts"][gs.SCREEN_CAPPED] == 3
        assert out["assess_bought"] == 2 and out["assess_reused"] == 0

    def test_a_cache_hit_does_not_eat_the_budget(self):
        """★되쓴 것을 상한에 세면 이어받은 판이 아무것도 못 산다."""
        store, calls = {}, []
        _run(self._subs(5), max_assess=2, cache_get=store.get,
             cache_put=store.__setitem__,
             assess_fn=self._cache_backed(store, calls))
        calls.clear()
        out = _run(self._subs(5), max_assess=2, cache_get=store.get,
                   cache_put=store.__setitem__,
                   assess_fn=self._cache_backed(store, calls))
        assert len(calls) == 2, "★앞판이 산 둘은 되쓰고, 새로 둘만 산다"
        assert out["assess_reused"] == 2 and out["assess_bought"] == 2
        assert out["counts"][gs.SCREEN_CAPPED] == 1

    def test_something_that_never_reads_the_cache_counts_as_bought(self):
        """★★캐시를 **안 보는** 구현이 끼면 상한이 통째로 무력해진다.

        늦게 닫히는 것보다 일찍 닫히는 것이 낫다 — 셈은 돈 쪽으로 보수적이다.
        """
        out = _run(self._subs(3), max_assess=2,
                   assess_fn=_fake_assess(era.PLAN_NO_SUBJECT))
        assert out["assess_bought"] == 2
        assert out["counts"][gs.SCREEN_CAPPED] == 1

    def test_something_already_cached_does_not_get_capped(self):
        """★★상한에 **먼저** 걸면 앞판이 사 둔 것까지 `capped` 가 된다.

        공짜로 쓸 수 있는 것을 못 쓰게 되고, 그 대상은 영영 미확정으로
        남아 하류가 막힌다 (Codex). 「살 것인가」를 **사기 전에** 가른다.
        """
        store, calls = {}, []
        pre_bought = self._cache_backed(store, calls)
        # ① 둘째 것만 미리 사 둔다
        _run([self._subs(2)[1]], cache_get=store.get,
             cache_put=store.__setitem__, assess_fn=pre_bought)
        assert len(calls) == 1
        calls.clear()

        # ② 상한 1 — 첫째를 사면 예산이 끝난다. 둘째는 **이미 있다**.
        out = _run(self._subs(2), max_assess=1, cache_get=store.get,
                   cache_put=store.__setitem__, assess_fn=pre_bought)
        assert len(calls) == 1, "★상한 1 을 넘겨 샀다"
        assert out["counts"][gs.SCREEN_CAPPED] == 0, \
            "★이미 있는 것을 `capped` 로 닫았다 — 공짜인데"
        assert out["assess_bought"] == 1 and out["assess_reused"] == 1

    def test_a_walk_memo_return_is_neither_bought_nor_reused(self):
        """★캐시를 **읽지도 않고** 실패로 온 것 — 걷기 memo. 안 샀다."""
        out = _run(self._subs(1),
                   assess_fn=lambda **kw: {"status": era.PLAN_FAILED,
                                           "reason": "same_walk_memo"})
        assert out["assess_bought"] == 0 and out["assess_reused"] == 0
        assert out["rows"][0]["from_cache"] is None

    def test_every_bought_row_is_flushed_before_the_next_buy(self):
        """★★N번째에서 끊기면 앞의 N-1 이 사라진다 — 줄마다 넘긴다."""
        store, calls, flushed = {}, [], []
        _run(self._subs(3), cache_get=store.get, cache_put=store.__setitem__,
             assess_fn=self._cache_backed(store, calls),
             on_row=lambda row, plan: flushed.append(row["research_subject_id"]))
        assert flushed == ["rs0", "rs1", "rs2"]

    def test_a_failing_flush_stops_before_the_next_buy(self):
        """★★**뒤집힌 계약** (Codex BLOCK-2). 삼키면 「호출마다 durable」이
        거짓이다 — 디스크가 찬 채로 스무 번 더 사고, 그 스무 번이 아무 데도
        안 남는다.
        """
        seen = []
        def _boom(row, plan):
            raise RuntimeError("디스크")
        with pytest.raises(RuntimeError, match="디스크"):
            _run(self._subs(5), on_row=_boom,
                 assess_fn=_fake_assess(era.PLAN_NO_SUBJECT, seen=seen))
        assert len(seen) == 1, "★첫 저장이 깨졌는데 더 샀다"


class TestTheCapRefusesJunk:
    @pytest.mark.parametrize("bad", [True, False, "2", 2.0, 0, -1])
    def test_it_only_takes_an_exact_positive_int(self, bad):
        """★`int(x)` 는 True→1 · "2"→2 · 0 · -1 을 전부 통과시킨다."""
        with pytest.raises(ValueError):
            gs.resolve_cap(bad)

    def test_none_means_the_contract_default(self):
        assert gs.resolve_cap(None) == gs.DEFAULT_MAX_ASSESS

    def test_a_plain_int_passes(self):
        assert gs.resolve_cap(7) == 7


class TestItSendsTheCanonicalIdentity:
    """★④셋이 다 있어야 정본이다 — 「**잘못 합치는 것이 중복 조사보다 나쁘다**」."""

    def test_all_three_canonical_slots_go_out(self):
        seen = []
        _run([_subj("rs_a", "location")],
             assess_fn=_fake_assess(era.PLAN_NO_SUBJECT, seen=seen))
        kw = seen[0]
        assert kw["canonical_scope_id"] == "rs_a"
        assert kw["canonical_scope_role"] == "location"
        assert kw["canonical_scope_sha"]

    def test_the_payload_sha_ignores_the_id_and_uses_structured_slots(self):
        """★sha 는 **구조화 칸**만 접는다. id 를 접으면 신원 셋이 한 칸이 된다."""
        same = {"surface_form": "표면", "source_quote": "원문"}
        a = gs.canonical_payload_sha(_subj("rs_a", **same))
        b = gs.canonical_payload_sha(_subj("rs_b", **same))
        assert a == b, "★id 가 달라도 내용이 같으면 같은 payload 다"

    def test_a_different_quote_is_a_different_payload(self):
        a = gs.canonical_payload_sha(_subj("rs_a"))
        b = gs.canonical_payload_sha(_subj("rs_a", source_quote="다른 문장"))
        assert a != b

    def test_a_different_owner_is_a_different_payload(self):
        a = gs.canonical_payload_sha(_subj("rs_a", "prop"))
        b = gs.canonical_payload_sha(_subj("rs_a", "location"))
        assert a != b

    def test_the_subject_text_carries_the_manuscript_quote(self):
        """★판별 근거는 **원문 문장**이다 — LLM 이 상상해 쓴 묘사가 아니다."""
        seen = []
        _run([_subj("rs_a")], assess_fn=_fake_assess(era.PLAN_NO_SUBJECT,
                                                     seen=seen))
        assert "원문-rs_a" in seen[0]["subject_text"]

    def test_the_walk_memo_is_passed_through(self):
        seen = []
        memo = set()
        _run([_subj("rs_a")], failed_memo=memo,
             assess_fn=_fake_assess(era.PLAN_NO_SUBJECT, seen=seen))
        assert seen[0]["failed_memo"] is memo, "★한 걷기 안에서 재시도하면 두 번 산다"


class TestItBuysNothing:
    def test_the_module_never_touches_acquisition(self):
        """★사는 함수를 **부르지 않는다** — 이름으로 확인하는 것이 아니라
        모듈이 무엇을 import 하는지로 본다."""
        import ast
        import inspect

        tree = ast.parse(inspect.getsource(gs))
        called = {n.func.attr for n in ast.walk(tree)
                  if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute)}
        assert "acquire_from_plan_cached" not in called
        assert "assess_and_research_cached" not in called
        assert "research_reference" not in called