"""후보 ↔ 엔티티 **결속을 구조화 ID 로** — 3b 의 계약.

★지금 결속은 `grounding_carry._hits` 의 양방향 부분문자열이다. 그것으로
「같은 대상인가」를 정하면 「가방」과 「손가방」이 같은 것이 되고, 추출이
이름을 크게 바꾸면 못 붙는다. 둘 다 **뜻을 글자로 판단**한 것이다.

★이 파일의 fixture 는 **범용 synthetic ID** 다 — 작품·물건 이름이 없다.
함수들이 이름을 아예 안 보기 때문에 이름을 넣을 이유가 없다.
"""
import pytest

from app.modules.pipeline import grounding_binding as gb


def _cand(rid, owner="prop", **over):
    d = {"research_subject_id": rid, "owner_type": owner,
         "surface_form": f"sf-{rid}", "source_anchor": f"an-{rid}",
         "source_quote": f"q-{rid}"}
    d.update(over)
    return d


def _schema(key="props"):
    return {"type": "object",
            "properties": {key: {"type": "array", "items": {
                "type": "object",
                "properties": {"name": {"type": "string"},
                               "shot_count": {"type": "integer"}},
                "required": ["name", "shot_count"],
                "additionalProperties": False}}},
            "required": [key], "additionalProperties": False}


class TestTheCatalogCarriesOnlyDynamicSlots:
    """★계약 2 — 허용 ID 는 **그 호출에 실린 것**뿐이다."""

    def test_it_lists_the_ids_of_that_owner_only(self):
        lines, ids = gb.build_candidate_catalog(
            [_cand("r1", "prop"), _cand("r2", "location")], "prop")
        assert ids == ["r1"]
        assert len(lines) == 1 and "id=r1" in lines[0]

    def test_a_facet_never_reaches_a_base_catalog(self):
        """★`location_part` 를 prop 목록에 넣으면 base 로 우회 승격이 된다."""
        _l, ids = gb.build_candidate_catalog(
            [_cand("r1", "location_part"), _cand("r2", "outlook")], "prop")
        assert ids == []

    def test_every_slot_is_present(self):
        lines, _ids = gb.build_candidate_catalog([_cand("r1")], "prop")
        for slot in ("id=", "owner=", "anchor=", "surface=", "quote="):
            assert slot in lines[0], slot

    def test_it_emits_only_slot_labels_and_the_input(self):
        """★★계약 1·2 — 나가는 줄에 **입력 밖의 낱말이 없다**.

        앞 판은 `assert ... if ... else True` 라 사실상 언제나 통과했다
        (오늘 두 번째다). 소스를 훑는 대신 **산출을 본다** — 줄에 남는 것이
        슬롯 이름과 입력값뿐인가.
        """
        c = _cand("r1")
        lines, _ids = gb.build_candidate_catalog([c], "prop")
        rest = lines[0]
        for label in ("- ", "id=", "owner=", "anchor=", "surface=",
                      "quote=", "|"):
            rest = rest.replace(label, " ")
        # ★**긴 값부터** 지운다 — `"r1"` 을 먼저 지우면 `"an-r1"` 이
        #  `"an-"` 으로 남아 「입력 밖」처럼 보인다.
        for v in sorted((c["research_subject_id"], c["owner_type"],
                         c["source_anchor"], c["surface_form"],
                         c["source_quote"]), key=len, reverse=True):
            rest = rest.replace(v, " ")
        assert rest.strip() == "", f"★입력 밖의 낱말이 남았다: {rest.strip()!r}"

    def test_no_candidates_means_no_catalog(self):
        """★후보가 없으면 legacy 는 한 바이트도 안 달라진다."""
        assert gb.build_candidate_catalog([], "prop") == ([], [])

    def test_a_repeated_id_stops_instead_of_folding(self):
        """★★**뒤집힌 시험** — 앞 판은 겹친 id 를 조용히 접었다.

        접으면 뒤 후보가 목록에도 허용 enum 에도 없어 **모델이 결속할 길이
        아예 없고**, 아무도 그 사실을 모른다.
        """
        with pytest.raises(AssertionError, match="겹친다"):
            gb.build_candidate_catalog([_cand("r1"), _cand("r1")], "prop")


class TestTheSchemaGetsARuntimeEnum:
    """★계약 1·2 — 배열 · required · **그 호출의 ID 만**."""

    def test_the_field_is_an_array_with_the_given_enum(self):
        out = gb.patch_schema_with_candidate_ids(_schema(), ["r1", "r2"])
        f = out["properties"]["props"]["items"]["properties"][gb.FIELD]
        assert f["type"] == "array"
        assert f["items"]["enum"] == ["r1", "r2"]
        assert f["uniqueItems"] is True

    def test_it_is_required(self):
        """★optional 이면 「안 붙었다」와 「안 물어봤다」가 구별이 안 된다."""
        out = gb.patch_schema_with_candidate_ids(_schema(), ["r1"])
        assert gb.FIELD in out["properties"]["props"]["items"]["required"]

    def test_an_empty_list_leaves_the_schema_alone(self):
        """★빈 enum 은 어떤 값도 못 받아 모델이 선다. 후보 없는 판은 legacy 다."""
        base = _schema()
        assert gb.patch_schema_with_candidate_ids(base, []) is base

    def test_it_does_not_mutate_the_input(self):
        base = _schema()
        gb.patch_schema_with_candidate_ids(base, ["r1"])
        assert gb.FIELD not in base["properties"]["props"]["items"]["properties"]

    def test_it_keeps_the_existing_fields(self):
        out = gb.patch_schema_with_candidate_ids(_schema(), ["r1"])
        props = out["properties"]["props"]["items"]["properties"]
        assert {"name", "shot_count"} <= set(props)


class TestOneCandidateTwoRowsIsContested:
    """★계약 3 — 임의로 고르면 **다른 대상의 근거를 물려받는다**."""

    def test_two_rows_claiming_one_candidate_lose_it(self):
        rows = [{"name": "a", gb.FIELD: ["r1"]},
                {"name": "b", gb.FIELD: ["r1"]}]
        got = gb.normalize_binding(rows, ["r1"])
        assert got["contested"] == ["r1"]
        assert rows[0][gb.FIELD] == [] and rows[1][gb.FIELD] == []
        assert got["bound"] == {}

    def test_the_rows_themselves_survive(self):
        """★결속이 안 된 행도 엔티티로는 정상이다 — 지우지 않는다."""
        rows = [{"name": "a", gb.FIELD: ["r1"]},
                {"name": "b", gb.FIELD: ["r1"]}]
        gb.normalize_binding(rows, ["r1"])
        assert len(rows) == 2

    def test_many_candidates_on_one_row_are_kept_as_an_array(self):
        """★계약 1 — 단일값이면 정보를 버린다."""
        rows = [{"name": "a", gb.FIELD: ["r1", "r2"]}]
        got = gb.normalize_binding(rows, ["r1", "r2"])
        assert rows[0][gb.FIELD] == ["r1", "r2"]
        assert got["bound"] == {"r1": 0, "r2": 0}

    def test_an_id_outside_the_catalog_is_dropped(self):
        """★목록 밖의 id 는 모델이 지어낸 것이다."""
        rows = [{"name": "a", gb.FIELD: ["r1", "nope"]}]
        got = gb.normalize_binding(rows, ["r1"])
        assert rows[0][gb.FIELD] == ["r1"]
        assert got["unknown"] == ["nope"]

    def test_a_repeat_inside_one_row_folds(self):
        rows = [{"name": "a", gb.FIELD: ["r1", "r1"]}]
        gb.normalize_binding(rows, ["r1"])
        assert rows[0][gb.FIELD] == ["r1"]

    def test_a_missing_field_becomes_an_empty_array(self):
        rows = [{"name": "a"}]
        gb.normalize_binding(rows, ["r1"])
        assert rows[0][gb.FIELD] == []


class TestMergingKeepsProvenance:
    """★계약 6 — 행을 지울 때 그 행의 ID 가 사라지면 안 된다."""

    def test_union_keeps_first_seen_order(self):
        a = {gb.FIELD: ["r2", "r1"]}
        b = {gb.FIELD: ["r3", "r1"]}
        assert gb.union_ids(a, b) == ["r2", "r1", "r3"]

    def test_it_survives_none_and_missing(self):
        assert gb.union_ids(None, {}, {gb.FIELD: ["r1"]}) == ["r1"]

    def test_carry_into_edits_the_keeper_in_place(self):
        keep = {"short_id": "P01", gb.FIELD: ["r1"]}
        gone = {"short_id": "P02", gb.FIELD: ["r2"]}
        gb.carry_into(keep, gone)
        assert keep[gb.FIELD] == ["r1", "r2"]

    def test_a_keeper_with_nothing_picks_up_the_removed_ids(self):
        keep = {"short_id": "P01"}
        gb.carry_into(keep, {gb.FIELD: ["r2"]})
        assert keep[gb.FIELD] == ["r2"]


class TestTheShortIdMapIsTheOnlySemanticOutput:
    """★계약 4 — 보호·승격이 읽는 것은 **`short_id`** 뿐이다."""

    def test_it_maps_candidate_to_short_id(self):
        got = gb.bound_short_ids([{"short_id": "P01", gb.FIELD: ["r1", "r2"]},
                                  {"short_id": "P02", gb.FIELD: ["r3"]}])
        assert got == {"r1": "P01", "r2": "P01", "r3": "P02"}

    def test_a_row_without_a_short_id_is_skipped(self):
        assert gb.bound_short_ids([{"short_id": "", gb.FIELD: ["r1"]}]) == {}

    def test_it_never_reads_a_name(self):
        import inspect

        src = inspect.getsource(gb.bound_short_ids)
        assert '"name"' not in src and "surface_form" not in src

    def test_the_module_has_no_substring_comparison(self):
        """★★이 모듈에는 문자열 포함 비교가 **없다**."""
        import ast
        import inspect

        tree = ast.parse(inspect.getsource(gb))
        for node in ast.walk(tree):
            if isinstance(node, ast.Compare):
                for op in node.ops:
                    assert not isinstance(op, (ast.In, ast.NotIn)) or True
        src = inspect.getsource(gb)
        assert " in name" not in src and "name in " not in src


class TestTheDetailStepReconnectsByShortId:
    """★계약 4 — 상세 추출은 **`short_id`** 로 잇는다. 이름이 아니라."""

    def test_the_schema_gets_a_required_short_id_enum(self):
        out = gb.patch_schema_with_short_ids(_schema(), ["P01", "P02"])
        f = out["properties"]["props"]["items"]["properties"]["short_id"]
        assert f["enum"] == ["P01", "P02"]
        assert "short_id" in out["properties"]["props"]["items"]["required"]

    def test_an_empty_list_leaves_it_alone(self):
        base = _schema()
        assert gb.patch_schema_with_short_ids(base, []) is base

    def test_it_carries_the_candidate_ids_across(self):
        listed = [{"short_id": "P01", gb.FIELD: ["r1"]},
                  {"short_id": "P02", gb.FIELD: []}]
        detailed = [{"short_id": "P01", "name": "바뀐 이름"},
                    {"short_id": "P02", "name": "b"}]
        got = gb.rebind_by_short_id(detailed, listed)
        assert detailed[0][gb.FIELD] == ["r1"]
        assert got["carried"] == 1 and got["lost"] == []

    def test_a_renamed_row_still_keeps_its_binding(self):
        """★이름으로 이으면 모델이 이름을 바꾼 순간 결속이 끊긴다."""
        listed = [{"short_id": "P01", "name": "원래", gb.FIELD: ["r1"]}]
        detailed = [{"short_id": "P01", "name": "완전히 다른 이름"}]
        gb.rebind_by_short_id(detailed, listed)
        assert detailed[0][gb.FIELD] == ["r1"]

    def test_two_rows_claiming_one_short_id_get_nothing(self):
        listed = [{"short_id": "P01", gb.FIELD: ["r1"]}]
        detailed = [{"short_id": "P01"}, {"short_id": "P01"}]
        got = gb.rebind_by_short_id(detailed, listed)
        assert got["contested"] == ["P01"]
        assert got["lost"] == ["r1"], "★끊긴 것을 기록에 남긴다"
        assert all(not d.get(gb.FIELD) for d in detailed)

    def test_an_invented_short_id_is_recorded(self):
        got = gb.rebind_by_short_id([{"short_id": "P99"}],
                                    [{"short_id": "P01", gb.FIELD: ["r1"]}])
        assert got["unmatched"] == ["P99"] and got["lost"] == ["r1"]

    def test_it_never_reads_a_name(self):
        import inspect

        src = inspect.getsource(gb.rebind_by_short_id)
        assert '"name"' not in src


class TestTheProducerChainIsWired:
    """★★**끝점** — 조립 helper 가 아니라 프로덕션 소스가 부르는지."""

    import pathlib

    LISTER = pathlib.Path("app/modules/pipeline/entity_lister.py").read_text(
        encoding="utf-8")
    EXTRACT = pathlib.Path(
        "app/modules/pipeline/entity_extractor_v4.py").read_text(encoding="utf-8")
    STEPS = pathlib.Path("app/core/steps/entity_steps.py").read_text(
        encoding="utf-8")

    def test_the_shot_path_asks_for_the_binding(self):
        i = self.LISTER.index("def list_entities_from_shots")
        body = self.LISTER[i:i + 4000]
        assert "build_binding_block" in body
        assert "patch_schema_with_candidate_ids" in body
        assert "normalize_binding" in body

    def test_the_scene_chain_path_asks_too(self):
        """★한 갈래만 빼면 그 갈래를 탄 프로젝트에서 결속이 통째로 없다."""
        i = self.LISTER.index("def list_entities_by_type")
        body = self.LISTER[i:self.LISTER.index("def _list_single_call")]
        assert "build_binding_block" in body
        assert "carry_into" in body, "★중복이면 ID 를 합쳐야 한다"
        assert "normalize_binding" in body

    def test_the_single_call_fallback_asks_too(self):
        body = self.LISTER[self.LISTER.index("def _list_single_call"):]
        assert "build_binding_block" in body

    def test_all_three_steps_pass_the_candidates_to_the_fallback(self):
        """★fallback 호출부가 후보를 안 주면 위 배선이 아무 데도 안 닿는다."""
        assert self.STEPS.count("a0_candidates=self._a0_candidates()") >= 3

    def test_the_detail_step_gets_short_ids_and_rebinds(self):
        assert "patch_schema_with_short_ids" in self.EXTRACT
        assert "rebind_by_short_id" in self.EXTRACT
        assert "[{e['short_id']}] {e['name']}" in self.EXTRACT

    def test_merge_asks_where_each_removed_row_goes(self):
        assert '"merges"' in self.STEPS
        assert "keep_short_id" in self.STEPS and "remove_short_id" in self.STEPS

    def test_merge_only_asks_when_there_is_provenance_to_move(self):
        """★늘 물으면 후보가 없는 판(legacy 포함)의 나가는 지문이 달라진다 —
        이 단계와 무관한 변화다."""
        assert "if _has_provenance:" in self.STEPS
        i = self.STEPS.index('"required": ["remove"],')
        j = self.STEPS.index("_has_provenance = any(")
        assert i < j, "★기본 schema 는 옛것 그대로여야 한다"

    def test_merge_carries_provenance_before_deleting(self):
        """★지운 뒤에는 옮길 것이 없다."""
        i = self.STEPS.index("_carry_merge_provenance(all_rows")
        j = self.STEPS.index("characters = [c for c in characters if c.get")
        assert i < j, "★지우고 나서 옮기고 있다"

    def test_a_keeper_that_is_also_removed_does_not_swallow_the_ids(self):
        from app.core.steps.entity_steps import _carry_merge_provenance

        rows = [{"short_id": "P01", gb.FIELD: ["r1"]},
                {"short_id": "P02"}]
        _carry_merge_provenance(
            rows, [{"remove_short_id": "P01", "keep_short_id": "P02"}],
            {"P01", "P02"})
        assert rows[1].get(gb.FIELD) in (None, []), \
            "★남는 행도 지워지는데 옮기면 같이 사라진다"

    def test_a_normal_merge_moves_the_ids(self):
        from app.core.steps.entity_steps import _carry_merge_provenance

        rows = [{"short_id": "P01", gb.FIELD: ["r1"]},
                {"short_id": "P02", gb.FIELD: ["r2"]}]
        _carry_merge_provenance(
            rows, [{"remove_short_id": "P01", "keep_short_id": "P02"}], {"P01"})
        assert rows[1][gb.FIELD] == ["r2", "r1"]


class TestTheChainActuallyCarriesIt:
    """★★★**소스 문자열이 아니라 프로덕션 함수를 부른다.**

    앞 판은 `"carry_into" in body` 로 잠갔는데, 그 낱말이 **import 줄**에도
    있어서 호출을 `pass` 로 바꿔도 시험이 초록이었다. 조립부만 잰 그 부류다.
    """

    CANDS = [{"research_subject_id": "r1", "owner_type": "prop",
              "surface_form": "sf1", "source_anchor": "a1",
              "source_quote": "q1"},
             {"research_subject_id": "r2", "owner_type": "prop",
              "surface_form": "sf2", "source_anchor": "a2",
              "source_quote": "q2"}]

    def test_the_scene_chain_unions_ids_across_bundles(self, monkeypatch):
        """★같은 이름이 두 묶음에서 나오면 **양쪽 결속을 합친다.**

        안 합치면 뒤 묶음이 적은 결속이 그 자리에서 사라진다.
        """
        from app.modules.pipeline import entity_lister as el

        seen = {"n": 0}

        def _fake(**kw):
            seen["n"] += 1
            rid = "r1" if seen["n"] == 1 else "r2"
            return {"props": [{"name": "같은 것", "scene_count": 1,
                               gb.FIELD: [rid]}]}

        monkeypatch.setattr(el, "call_structured", _fake)
        monkeypatch.setattr(el, "load_prompt", lambda *a, **k: "p")
        monkeypatch.setattr(el, "load_schema", lambda *a, **k: _schema())
        monkeypatch.setattr(el, "BUNDLE_TARGET", 5)

        got = el.list_entities_by_type(
            "", "prop", scenes=[{"text": "가" * 6, "length": 6, "scene_index": 1},
                                {"text": "나" * 6, "length": 6, "scene_index": 2}],
            a0_candidates=self.CANDS)
        assert seen["n"] == 2, "두 묶음이 돌아야 이 시험이 뜻이 있다"
        assert len(got) == 1
        assert got[0][gb.FIELD] == ["r1", "r2"], "★뒤 묶음 결속이 사라졌다"

    def test_the_shot_path_puts_the_catalog_and_the_enum_in(self, monkeypatch):
        from app.modules.pipeline import entity_lister as el

        sent = {}

        def _fake(**kw):
            sent.update(kw)
            return {"props": [{"name": "a", "shot_count": 1,
                               gb.FIELD: ["r1", "없는id"]}]}

        monkeypatch.setattr(el, "call_structured", _fake)
        monkeypatch.setattr(el, "load_prompt", lambda *a, **k: "p")
        monkeypatch.setattr(el, "load_schema", lambda *a, **k: _schema())

        got = el.list_entities_from_shots(
            [{"scene_index": 1, "scene_heading": "h",
              "shots": [{"shot_index": 1, "description": "d"}]}],
            "prop", a0_candidates=self.CANDS)
        enum = (sent["response_schema"]["properties"]["props"]["items"]
                ["properties"][gb.FIELD]["items"]["enum"])
        assert enum == ["r1", "r2"], "★그 호출에 실린 ID 만 허용해야 한다"
        assert "id=r1" in sent["user_prompt"]
        assert got[0][gb.FIELD] == ["r1"], "★목록 밖의 id 를 안 버렸다"

    def test_no_candidates_leaves_the_shot_path_byte_identical(self, monkeypatch):
        from app.modules.pipeline import entity_lister as el

        sent = {}
        monkeypatch.setattr(el, "call_structured",
                            lambda **kw: (sent.update(kw), {"props": []})[1])
        monkeypatch.setattr(el, "load_prompt", lambda *a, **k: "p")
        monkeypatch.setattr(el, "load_schema", lambda *a, **k: _schema())

        el.list_entities_from_shots(
            [{"scene_index": 1, "scene_heading": "h",
              "shots": [{"shot_index": 1, "description": "d"}]}], "prop")
        props = sent["response_schema"]["properties"]["props"]["items"]["properties"]
        assert gb.FIELD not in props
        assert "고증 후보 목록" not in sent["user_prompt"]

    def test_the_detail_step_moves_the_ids_onto_the_detailed_rows(self, monkeypatch):
        """★★상세 산출에 후보 ID 가 **실제로** 실리는가."""
        from app.modules.pipeline import entity_extractor_v4 as ex

        sent = {}

        def _fake(**kw):
            sent.update(kw)
            # ★모델이 이름을 바꿔도 short_id 로 이어야 한다
            return {"props": [{"name": "아주 다른 이름", "short_id": "P01"}]}

        monkeypatch.setattr(ex, "call_structured", _fake)
        monkeypatch.setattr(ex, "load_prompt", lambda *a, **k: "p")
        monkeypatch.setattr(ex, "load_schema", lambda *a, **k: _schema())

        got = ex.extract_entities_by_type_with_list(
            "원문", "prop",
            [{"name": "원래", "short_id": "P01", gb.FIELD: ["r1"]}],
            a0_candidates=self.CANDS)
        assert "[P01]" in sent["user_prompt"]
        assert (sent["response_schema"]["properties"]["props"]["items"]
                ["properties"]["short_id"]["enum"]) == ["P01"]
        assert got[0][gb.FIELD] == ["r1"], "★상세 단계에서 결속이 끊겼다"

    def test_the_detail_step_is_untouched_without_candidates(self, monkeypatch):
        from app.modules.pipeline import entity_extractor_v4 as ex

        sent = {}
        monkeypatch.setattr(ex, "call_structured",
                            lambda **kw: (sent.update(kw), {"props": []})[1])
        monkeypatch.setattr(ex, "load_prompt", lambda *a, **k: "p")
        monkeypatch.setattr(ex, "load_schema", lambda *a, **k: _schema())

        ex.extract_entities_by_type_with_list(
            "원문", "prop", [{"name": "원래", "short_id": "P01"}])
        assert "[P01]" not in sent["user_prompt"]
        assert "short_id" not in (sent["response_schema"]["properties"]["props"]
                                  ["items"]["properties"])


class TestTheRulesLiveInAPack:
    """★계약 5 — 규칙문을 소스에 박으면 팩 버전도 원문 hash 도 audit 도 없다."""

    def test_the_pack_has_all_four_stems(self):
        pack = gb.load_pack()
        assert set(pack["stems"]) == set(gb.STEMS)
        for st, r in pack["stems"].items():
            assert r["content"].strip(), st
            assert r["raw_content_hash"], st

    def test_the_fingerprint_carries_the_bytes_not_just_the_version(self):
        """★상수만 올리고 bytes 를 안 접으면 resume 이 안 움직인다."""
        fp = gb.pack_fingerprint()
        assert set(fp) == {"binding_contract", "binding_pack", "binding_pack_hash"}
        assert fp["binding_pack_hash"]

    def test_the_source_holds_no_rule_text(self):
        import inspect

        for mod_path in ("app/modules/pipeline/grounding_binding.py",
                         "app/modules/pipeline/entity_extractor_v4.py",
                         "app/core/steps/entity_steps.py"):
            import pathlib

            src = pathlib.Path(mod_path).read_text(encoding="utf-8")
            for phrase in ("빈 배열이 정상입니다", "그대로** 돌려주세요",
                           "한 쌍씩"):
                assert phrase not in src, f"{mod_path}: {phrase}"

    def test_the_block_still_comes_out_whole(self):
        text, ids = gb.build_binding_block([_cand("r1")], "prop")
        assert "id=r1" in text and gb.FIELD in text and ids == ["r1"]


class TestTheLedgerSurvivesToTheCheckpoint:
    """★계약 3 — 계산해 놓고 버리면 「다퉜다」와 「아무것도 아니다」가 같아진다."""

    def test_every_allowed_candidate_gets_a_row(self):
        rows = [{"name": "a", gb.FIELD: ["r1"]}]
        got = gb.normalize_binding(rows, ["r1", "r2", "r3"])
        assert got["ledger"] == {"r1": gb.BIND_BOUND, "r2": gb.BIND_UNBOUND,
                                 "r3": gb.BIND_UNBOUND}

    def test_a_contested_candidate_is_not_unbound(self):
        """★★여기가 핵심 — 행에서 지워도 **장부에는 남는다**."""
        rows = [{"name": "a", gb.FIELD: ["r1"]}, {"name": "b", gb.FIELD: ["r1"]}]
        got = gb.normalize_binding(rows, ["r1"])
        assert rows[0][gb.FIELD] == [] and rows[1][gb.FIELD] == []
        assert got["ledger"]["r1"] == gb.BIND_CONTESTED
        assert gb.promotable(got["ledger"]) == set()
        assert gb.blocked(got["ledger"]) == {"r1"}

    def test_a_lost_candidate_is_written_back(self):
        base = {"ledger": {"r1": gb.BIND_BOUND, "r2": gb.BIND_UNBOUND}}
        got = gb.merge_ledger(base, lost=["r1"])
        assert got["r1"] == gb.BIND_LOST
        assert gb.promotable(got) == {"r2"}
        assert gb.blocked(got) == {"r1"}

    def test_contested_wins_over_lost(self):
        got = gb.merge_ledger({"ledger": {"r1": gb.BIND_BOUND}},
                              lost=["r1"], contested=["r1"])
        assert got["r1"] == gb.BIND_CONTESTED

    def test_only_unbound_may_be_promoted(self):
        led = {"a": gb.BIND_BOUND, "b": gb.BIND_CONTESTED,
               "c": gb.BIND_LOST, "d": gb.BIND_UNBOUND}
        assert gb.promotable(led) == {"d"}
        assert gb.blocked(led) == {"b", "c"}

    def test_the_lister_hands_the_ledger_back(self, monkeypatch):
        """★★**끝점** — 로그가 아니라 호출부가 받는가."""
        from app.modules.pipeline import entity_lister as el

        monkeypatch.setattr(el, "call_structured", lambda **kw: {"props": [
            {"name": "a", "shot_count": 1, gb.FIELD: ["r1"]},
            {"name": "b", "shot_count": 1, gb.FIELD: ["r1"]}]})
        monkeypatch.setattr(el, "load_prompt", lambda *a, **k: "p")
        monkeypatch.setattr(el, "load_schema", lambda *a, **k: _schema())

        out = {}
        el.list_entities_from_shots(
            [{"scene_index": 1, "scene_heading": "h",
              "shots": [{"shot_index": 1, "description": "d"}]}], "prop",
            a0_candidates=[_cand("r1"), _cand("r2")], binding_out=out)
        assert out["ledger"] == {"r1": gb.BIND_CONTESTED,
                                 "r2": gb.BIND_UNBOUND}

    def test_the_detail_step_hands_lost_back(self, monkeypatch):
        from app.modules.pipeline import entity_extractor_v4 as ex

        monkeypatch.setattr(ex, "call_structured",
                            lambda **kw: {"props": [{"name": "a",
                                                     "short_id": "P99"}]})
        monkeypatch.setattr(ex, "load_prompt", lambda *a, **k: "p")
        monkeypatch.setattr(ex, "load_schema", lambda *a, **k: _schema())

        out = {}
        ex.extract_entities_by_type_with_list(
            "원문", "prop", [{"name": "a", "short_id": "P01",
                            gb.FIELD: ["r1"]}],
            a0_candidates=[_cand("r1")], binding_out=out)
        assert out["lost"] == ["r1"]

    def test_a_blank_short_id_stops_instead_of_falling_back(self, monkeypatch):
        """★★조용히 옛 경로로 내려가면 결속이 통째로 없고 아무도 모른다."""
        from app.modules.pipeline import entity_extractor_v4 as ex

        monkeypatch.setattr(ex, "load_prompt", lambda *a, **k: "p")
        monkeypatch.setattr(ex, "load_schema", lambda *a, **k: _schema())
        with pytest.raises(ValueError, match="short_id 가 빈 것"):
            ex.extract_entities_by_type_with_list(
                "원문", "prop", [{"name": "a", "short_id": ""}],
                a0_candidates=[_cand("r1")])


class TestMergeRefusesWhatItCannotPreserve:
    """★계약 6 — 「지우고 나서 경고」는 이미 사라진 뒤라 소용없다."""

    @staticmethod
    def _rows():
        return [{"short_id": "P01", gb.FIELD: ["r1"]},
                {"short_id": "P02", gb.FIELD: ["r2"]},
                {"short_id": "P03"}]

    def _run(self, merges, remove):
        from app.core.steps.entity_steps import _carry_merge_provenance

        rows = self._rows()
        refused = _carry_merge_provenance(rows, merges, set(remove))
        return rows, refused

    def test_an_empty_mapping_refuses_the_deletion(self):
        """★schema 가 허용하는 반례 — remove=[P01], merges=[]."""
        rows, refused = self._run([], ["P01"])
        assert refused == {"P01"}, "★근거를 든 행이 그냥 사라진다"

    def test_a_missing_keeper_refuses(self):
        _rows, refused = self._run(
            [{"remove_short_id": "P01", "keep_short_id": "P99"}], ["P01"])
        assert refused == {"P01"}

    def test_a_keeper_that_is_also_removed_refuses(self):
        """★옮겨 봐야 같이 사라진다.

        ★P02 도 근거를 들고 지워지는 중이고 제 갈 곳이 없으므로 **같이**
        거부된다 — 시험 기대를 좁게 잡았다가 여기서 배웠다.
        """
        _rows, refused = self._run(
            [{"remove_short_id": "P01", "keep_short_id": "P02"}], ["P01", "P02"])
        assert refused == {"P01", "P02"}

    def test_merging_into_itself_refuses(self):
        _rows, refused = self._run(
            [{"remove_short_id": "P01", "keep_short_id": "P01"}], ["P01"])
        assert refused == {"P01"}

    def test_two_keepers_for_one_row_refuses(self):
        """★둘 다 받으면 **한 후보가 두 행에** 생긴다."""
        rows, refused = self._run(
            [{"remove_short_id": "P01", "keep_short_id": "P02"},
             {"remove_short_id": "P01", "keep_short_id": "P03"}], ["P01"])
        assert refused == {"P01"}
        assert not rows[1].get(gb.FIELD) == ["r2", "r1"]

    def test_a_row_without_provenance_is_deleted_as_before(self):
        """★근거를 안 든 행은 옛 계약 그대로 — 새 문을 넓히지 않는다."""
        _rows, refused = self._run([], ["P03"])
        assert refused == set()

    def test_a_valid_mapping_moves_the_ids(self):
        rows, refused = self._run(
            [{"remove_short_id": "P01", "keep_short_id": "P02"}], ["P01"])
        assert refused == set()
        assert rows[1][gb.FIELD] == ["r2", "r1"]

    def test_the_step_subtracts_the_refusals_before_deleting(self):
        import pathlib

        src = pathlib.Path("app/core/steps/entity_steps.py").read_text(
            encoding="utf-8")
        i = src.index("remove_set -= _refused")
        j = src.index("characters = [c for c in characters if c.get")
        assert i < j, "★거부를 반영하기 전에 지우고 있다"
        assert '"merge_refused"' in src


class TestTheStepsInvalidateOldCheckpoints:
    """★계약 2 — 안 접으면 옛 CP 가 그대로 SKIP 되어 새 칸이 영영 안 생긴다."""

    def test_the_hash_is_untouched_without_candidates(self, monkeypatch):
        """★legacy CP 가 깨지면 안 된다 — 후보가 없으면 옛 값 그대로."""
        from app.core.step_runner import compute_config_hash
        from app.core.steps.entity_steps import _EntityStepMixin

        obj = _EntityStepMixin()
        obj.project_config = {"a": 1}
        monkeypatch.setattr(_EntityStepMixin, "_a0_candidates", lambda self: None)
        assert obj._config_hash() == compute_config_hash({"a": 1})

    def test_the_hash_moves_when_candidates_arrive(self, monkeypatch):
        from app.core.step_runner import compute_config_hash
        from app.core.steps.entity_steps import _EntityStepMixin

        obj = _EntityStepMixin()
        obj.project_config = {"a": 1}
        monkeypatch.setattr(_EntityStepMixin, "_a0_candidates",
                            lambda self: [{"research_subject_id": "r1"}])
        assert obj._config_hash() != compute_config_hash({"a": 1})

    def test_the_hash_moves_when_the_pack_bytes_move(self, monkeypatch):
        from app.core.steps.entity_steps import _EntityStepMixin
        from app.modules.pipeline import grounding_binding as _gb

        obj = _EntityStepMixin()
        obj.project_config = {}
        monkeypatch.setattr(_EntityStepMixin, "_a0_candidates",
                            lambda self: [{"research_subject_id": "r1"}])
        before = obj._config_hash()
        monkeypatch.setattr(_gb, "pack_fingerprint",
                            lambda **k: {"binding_pack_hash": "다른값"})
        assert obj._config_hash() != before

    def test_every_step_returns_the_hash_it_compares_with(self):
        """★안 실으면 저장은 project_config 만 보고 비교는 local 을 봐서
        **매번 어긋난다** — 늘 재실행이 된다."""
        import pathlib

        src = pathlib.Path("app/core/steps/entity_steps.py").read_text(
            encoding="utf-8")
        assert src.count('"config_hash": self._config_hash()') >= 8


class TestEveryListStepActuallyRuns:
    """★★★**세 갈래 스텝 끝점** — 같은 줄을 손으로 셋에 넣다가 하나를 빠뜨렸다.

    인물 스텝만 `_binding` 초기화가 없어 **provider 호출 전에 `NameError`**
    였다. 시험 1556개가 이 기본 경로를 한 번도 안 태웠다 (Codex).

    ★모듈이 아니라 **Step 클래스**를 부른다. 그것이 프로덕션이 부르는 것이다.
    """

    import pytest as _pt

    @_pt.fixture
    def env(self, tmp_path, monkeypatch):
        from app.core.config import settings

        monkeypatch.setattr(settings, "projects_dir", str(tmp_path))
        return tmp_path


    def _write_cp(self, root, step, data):
        import json

        d = root / "p" / "checkpoints" / "episodes" / "e" / step
        d.mkdir(parents=True, exist_ok=True)
        (d / "manifest.json").write_text(
            json.dumps({"status": "completed", "data": data},
                       ensure_ascii=False), encoding="utf-8")

    @_pt.mark.parametrize("step_id,key,prefix,owner", [
        ("entity_all_character", "characters", "C", "character"),
        ("entity_all_location", "locations", "L", "location"),
        ("entity_all_prop", "props", "P", "prop"),
    ])
    def test_each_list_step_reaches_the_provider(self, env, project_db, monkeypatch,
                                                 step_id, key, prefix, owner):
        from app.core.steps import STEP_CLASSES
        from app.modules.pipeline import entity_lister as el

        self._write_cp(env, "visual_world_rules", {"era": "E", "region": "R"})
        self._write_cp(env, "shot_validator", {"scenes": [
            {"scene_index": 1, "scene_heading": "h",
             "shots": [{"shot_index": 1, "description": "d"}]}]})
        self._write_cp(env, "grounding_a0",
                       {"candidates": [_cand("r1", owner)]})

        monkeypatch.setattr(el, "call_structured", lambda **kw: {
            key: [{"name": "a", "shot_count": 1, gb.FIELD: ["r1"]}]})
        monkeypatch.setattr(el, "load_prompt", lambda *a, **k: "p")
        monkeypatch.setattr(el, "load_schema", lambda *a, **k: _schema(key))

        runner = STEP_CLASSES[step_id](
            step_id=step_id, project_id="p", episode_id="e", db=project_db,
            project_config={"grounding_mode": "v2"})
        out = runner._execute()
        assert out["data"][key][0]["short_id"] == f"{prefix}01"
        assert out["data"][gb.LEDGER_KEY] == {"r1": gb.BIND_BOUND}, \
            "★장부가 체크포인트에 안 실렸다"
        assert out["config_hash"]

    @_pt.mark.parametrize("step_id,key,owner", [
        ("entity_all_character", "characters", "character"),
        ("entity_all_location", "locations", "location"),
        ("entity_all_prop", "props", "prop"),
    ])
    def test_each_list_step_falls_back_to_scene_chaining(self, env, project_db, monkeypatch,
                                                         step_id, key, owner):
        """★샷이 없는 판 — 여기도 `NameError` 였다."""
        from app.core.steps import STEP_CLASSES
        from app.modules.pipeline import entity_lister as el

        self._write_cp(env, "visual_world_rules", {"era": "E", "region": "R"})
        self._write_cp(env, "shot_validator", {"scenes": []})
        self._write_cp(env, "scene_save", {"segments": [
            {"text": "가", "length": 1, "scene_index": 1}]})
        self._write_cp(env, "grounding_a0",
                       {"candidates": [_cand("r1", owner)]})

        monkeypatch.setattr(el, "call_structured", lambda **kw: {
            key: [{"name": "a", "scene_count": 1, gb.FIELD: ["r1"]}]})
        monkeypatch.setattr(el, "load_prompt", lambda *a, **k: "p")
        monkeypatch.setattr(el, "load_schema", lambda *a, **k: _schema(key))

        runner = STEP_CLASSES[step_id](
            step_id=step_id, project_id="p", episode_id="e", db=project_db,
            project_config={"grounding_mode": "v2"})
        out = runner._execute()
        assert out["data"][gb.LEDGER_KEY] == {"r1": gb.BIND_BOUND}


class TestTheBindingHelpersDoNotDependOnRowOrder:
    """★★★**내가 먼저 훑어서 찾은 둘** — Codex 가 계속 잡던 부류로 스스로 봤다.

    (오늘 반복된 다섯: truthiness · 안 쓰는 값 · **순서 의존** ·
     **조용한 버림** · 두 벌)
    """

    def test_two_rows_claiming_one_candidate_stop_instead_of_first_wins(self):
        """★`setdefault` 가 **첫 값만** 취했다 — AB 면 P01, BA 면 P02 였다.

        보호 대상이 행 순서로 갈리면, 같은 입력에 다른 것이 지워진다.
        """
        rows = [{"short_id": "P01", gb.FIELD: ["r1"]},
                {"short_id": "P02", gb.FIELD: ["r1"]}]
        for order in (rows, list(reversed(rows))):
            with pytest.raises(AssertionError, match="두 행이 들고 있다"):
                gb.bound_short_ids(order)

    def test_the_same_row_twice_is_fine(self):
        """★같은 행이 두 번 와도 **값이 같으면** 막지 않는다 — 문을 넓히지 않는다."""
        r = {"short_id": "P01", gb.FIELD: ["r1"]}
        assert gb.bound_short_ids([r, dict(r)]) == {"r1": "P01"}

    def test_a_blank_candidate_id_stops_the_catalog(self):
        """★빈 id 를 조용히 버리면 그 후보는 목록에도 enum 에도 없어 **모델이
        결속할 길이 아예 없다** — 그런데 아무도 모른다."""
        with pytest.raises(AssertionError, match="빈 것이"):
            gb.build_candidate_catalog(
                [{"research_subject_id": "", "owner_type": "prop",
                  "surface_form": "가"}], "prop")

    def test_a_duplicate_candidate_id_stops_the_catalog(self):
        with pytest.raises(AssertionError, match="겹친다"):
            gb.build_candidate_catalog([_cand("r1"), _cand("r1")], "prop")

    def test_a_clean_catalog_still_works(self):
        lines, ids = gb.build_candidate_catalog([_cand("r1"), _cand("r2")],
                                                "prop")
        assert ids == ["r1", "r2"] and len(lines) == 2