"""합성 원고로 C(c) 계약을 **끝에서 끝까지** 태운다. ★무료 · 모델 0.

지금까지의 계약 시험은 손으로 지은 행 두어 개를 넣었다. 그것은 규칙을
잠그지만 **「원고 하나가 실제로 이 사슬을 통과하는가」**는 안 본다.

여기서는 `fixtures/synthetic_episode` 의 원고를 구간으로 나누고, 각 구간이
낼 법한 행을 **원고의 정본 span 으로** 만들어 `reduce_rows` →
`should_register` → `assign_final_ids` 를 한 번에 태운다.

★**모델을 안 부른다.** 「구간이 이런 행을 낸다면」을 코드가 짓는 것이고,
모델이 실제로 그렇게 낼지는 **이 시험이 재는 것이 아니다** — 그건 유료다.
여기서 재는 것은 **계약이 그 행들을 옳게 줄이는가** 뿐이다.
"""
import pytest

from app.modules.pipeline import grounding_chunk_merge as cm
from tests.grounding.fixtures import synthetic_episode as ep


@pytest.fixture(scope="module", autouse=True)
def _planted():
    """★심었다고 적은 것이 원고에 **실제로 있는지** 먼저 본다.

    안 그러면 아래 시험들이 아무것도 안 재면서 초록이 된다.
    """
    ep.assert_planted()


#: ★한 구간이 여러 행을 낸다 — `row_index` 가 달라야 `local_id` 가 안 겹친다.
#:  앞 판은 전부 `0` 이라 다섯 갈래 시험이 「겹친다」로 섰다. 시험이 잡았다.
_seq = {}


def _row(chunk_scene, owner, word, occurrence=1, shots=None, **over):
    """그 구간이 낼 법한 행 하나. ★span 은 **원고에서 잰다**.

    ★반복 축은 **샷 단위**다. `shots` 를 안 주면 그 씬의 샷 하나에 결속한다 —
    같은 씬 안에서 여러 번 불러도 **1회**다.
    """
    key = f"scene-{chunk_scene}"
    _seq[key] = _seq.get(key, -1) + 1
    d = {"local_id": cm.local_id(key, _seq[key]),
         "owner_type": owner, "surface_form": word,
         "occurrences": [{"source_span": ep.span_of(chunk_scene, word,
                                                    occurrence),
                          "source_quote": word}],
         "shot_binding_status": "bound_complete",
         "shot_appearance_ids": (list(shots) if shots is not None
                                 else [f"s{chunk_scene}#1"]),
         "hard_to_generate": False, "viewers_would_notice": False}
    d.update(over)
    return d


class TestTheManuscriptItself:
    def test_the_chunks_are_the_production_convention(self):
        """★구간 경계는 `BUNDLE_TARGET` 이 정한다 — 새 수를 안 지어낸다."""
        from app.modules.pipeline.entity_lister import BUNDLE_TARGET

        assert len(ep.bundles()) >= 2, "★한 구간이면 구간 간 반례가 안 산다"
        for part in ep.bundles():
            total = sum(s["length"] for s in ep.segments()
                        if s["scene_index"] in part)
            assert total <= BUNDLE_TARGET or len(part) == 1

    def test_the_shots_match_the_scenes(self):
        assert [s["scene_index"] for s in ep.shot_scenes()] == \
            [s["scene_index"] for s in ep.segments()]
        assert all(sc["shots"] for sc in ep.shot_scenes())

    def test_every_span_lands_on_its_word(self):
        """★좌표가 원문의 그 자리인가 — 아니면 이 fixture 가 거짓말이다."""
        text = ep.segment_texts()
        for sc, word in ((1, "가방"), (2, "정류장 표지"), (3, "옛 요금표"),
                         (4, "가방")):
            sp = ep.span_of(sc, word)
            assert text[sp["segment_id"]][sp["start"]:sp["end"]] == word


class TestTheChainOnARealManuscript:
    """★★원고 하나가 사슬을 통과하는가."""

    def test_the_same_thing_in_two_chunks_becomes_one_registered_row(self):
        """① 1장·4장의 `가방` — 합쳐야 2회가 되어 등록된다."""
        rows = [_row(1, "prop", "가방"), _row(4, "prop", "가방")]
        assert cm._should_register([rows[0]])[0] is False, "★혼자서는 1회다"
        got = cm._reduce_rows([dict(r) for r in rows],
                             [{"remove_local_id": rows[1]["local_id"],
                               "keep_local_id": rows[0]["local_id"],
                               "relation": cm.REL_SAME}])
        assert got["counts"]["out"] == 1
        assert cm._should_register(got["rows"]) == (True, "shot_appearances=2")

    def test_the_same_word_for_two_different_things_is_not_merged(self):
        """② 1장의 `표`(승차권)와 4장의 `표`(게시판) — **안 합친다**."""
        rows = [_row(1, "prop", "표"), _row(4, "prop", "표")]
        got = cm._reduce_rows([dict(r) for r in rows], [])
        assert got["counts"]["out"] == 2, "★이름이 같다고 합쳤다"
        for r in got["rows"]:
            assert cm._should_register([r])[0] is False

    def test_twice_in_one_shot_is_still_one_appearance(self):
        """★★★③ 2장의 `정류장 표지` 두 번 — **자리는 둘, 등장은 한 씬**이다.

        ★유료 주행이 이것을 뒤집었다 (2026-08-31). 앞 판은 「한 구간 안 2회 =
        2회」로 잠갔는데, production 반복축(`entity_filter._appearance_count`)은
        **등장한 자리의 수**를 센다 — 부른 횟수가 아니다. 같은 씬에서 여러 말로
        불러도 등장은 한 번이다.
        """
        row = _row(2, "prop", "정류장 표지")
        row["occurrences"].append(
            {"source_span": ep.span_of(2, "정류장 표지", 2),
             "source_quote": "정류장 표지"})
        assert cm.unique_anchor_count([row]) == 2, "★자리는 둘이 맞다"
        assert cm.appearance_count([row]) == (1, False), "★등장은 한 샷이다"
        assert cm._should_register([row]) == (False, "shot_appearances=1")

    def test_the_twice_seen_sign_needs_the_exception_axis(self):
        """★positive control — 그래서 두 축이 서면 **예외 축으로** 산다."""
        row = _row(2, "prop", "정류장 표지", hard_to_generate=True,
                   viewers_would_notice=True)
        assert cm._should_register([row]) == (True, "grounding_exception")

    def test_the_part_is_not_folded_into_the_whole(self):
        """④ 벽에 붙인 `긴 걸상` 은 `대합실` 의 **부분**이다 — 합치면 사라진다.

        ★앞 판 반례는 「대합실 ⊂ 정류장」이었는데, 방은 **장면을 담는 공간**
        이라 관계 계약상 `location` 으로도 읽힌다 — 모호했다 (Codex).
        """
        part = _row(3, "location_part", "긴 걸상")
        whole = _row(3, "location", "대합실")
        got = cm._reduce_rows([dict(part), dict(whole)],
                             [{"remove_local_id": part["local_id"],
                               "keep_local_id": whole["local_id"],
                               "relation": cm.REL_PART_OF}])
        assert got["counts"]["out"] == 2
        assert got["part_of"] == [{"part": part["local_id"],
                                   "whole": whole["local_id"]}]

    def test_the_once_only_hard_thing_is_registered_by_the_exception(self):
        """⑥ 3장의 `옛 요금표` — 한 번뿐이라 **예외 축**으로만 산다."""
        plain = _row(3, "prop", "옛 요금표")
        assert cm._should_register([plain]) == (False, "shot_appearances=1")
        flagged = _row(3, "prop", "옛 요금표",
                       hard_to_generate=True, viewers_would_notice=True)
        assert cm._should_register([flagged]) == (True, "grounding_exception")

    def test_all_five_owners_survive_and_get_ids(self):
        """⑤ 다섯 갈래가 다 남고 각자 번호를 받는다."""
        rows = [_row(1, "character", "운전사"),
                _row(3, "location", "대합실"),
                _row(1, "prop", "가방"),
                _row(3, "location_part", "긴 걸상"),
                _row(3, "outlook", "제복 상의")]
        got = cm._reduce_rows([dict(r) for r in rows], [])
        assert got["counts"]["out"] == 5
        ids = cm._assign_final_ids(got["rows"])
        assert ids["contested"] == []
        assert sorted(ids["ids"].values()) == ["C01", "L01", "LP01", "O01",
                                               "P01"]

    def test_the_manuscript_verifies_every_quote(self):
        """★★`assert_rows(segments=…)` 가 **원문 대조**를 실제로 한다."""
        rows = [_row(1, "prop", "가방")]
        cm._assert_rows(rows, segments=ep.segment_texts())
        rows[0]["occurrences"][0]["source_quote"] = "없는말"
        with pytest.raises(AssertionError, match="그 자리에 없다"):
            cm._assert_rows(rows, segments=ep.segment_texts())


class TestThePublicEndpointCannotSkipVerification:
    """★★★**공개 경로가 원문 검증을 우회하던 것** (Codex).

        `reduce_rows` 도 `assign_final_ids` 도 `segments` 없이 불렸고
        `should_register` 는 검증을 아예 안 했다 —
        `source_quote="WRONG"` 인 행이 그냥 지나갔다.

    이제 공개 경로는 `reduce_episode` **하나**이고 `segments` 가
    **required keyword-only** 다. 원문 없이 부를 길이 없다.
    """

    _n = {}

    def _row(self, scene, word, shots=None, **over):
        """★한 장면이 여러 행을 낸다 — `row_index` 를 달리 준다."""
        key = f"scene-{scene}"
        self._n[key] = self._n.get(key, -1) + 1
        d = {"local_id": cm.local_id(key, self._n[key]),
             "owner_type": "prop", "surface_form": word,
             "occurrences": [{"source_span": ep.span_of(scene, word),
                              "source_quote": word}],
             "shot_binding_status": "bound_complete",
             "shot_appearance_ids": (list(shots) if shots is not None
                                     else [f"s{scene}#1"]),
             "hard_to_generate": False, "viewers_would_notice": False}
        d.update(over)
        return d

    def test_the_happy_path_runs_end_to_end(self):
        got = cm.reduce_episode([self._row(1, "가방")], [],
                                segments=ep.segment_texts())
        assert got["counts"]["out"] == 1
        one = next(iter(got["registered"].values()))
        assert one["registered"] is False        # 1장에 한 번뿐이다
        assert one["reason"] == "shot_appearances=1"
        # ★등록이 안 됐으니 **번호도 없다** — 최종 ID 는 등록된 것의 신원이다.
        assert got["final_ids"] == {} and one["final_id"] is None

    def test_a_quote_that_is_not_there_stops_before_anything(self):
        bad = self._row(1, "가방")
        bad["occurrences"][0]["source_quote"] = "없는말"
        with pytest.raises(AssertionError, match="그 자리에 없다"):
            cm.reduce_episode([bad], [], segments=ep.segment_texts())

    def test_an_empty_quote_cannot_skip_the_check(self):
        """★빈 인용은 「없는 것」이 아니라 **검증을 건너뛰는 문**이었다."""
        bad = self._row(1, "가방")
        bad["occurrences"][0]["source_quote"] = ""
        with pytest.raises(AssertionError, match="source_quote"):
            cm.reduce_episode([bad], [], segments=ep.segment_texts())

    def test_it_refuses_to_run_without_the_manuscript(self):
        with pytest.raises(AssertionError, match="segments"):
            cm.reduce_episode([self._row(1, "가방")], [], segments={})

    def test_two_different_things_do_not_pool_their_occurrences(self):
        """★★★**등록 판정이 에피소드 통째로 섞이던 것** (Codex).

        `_should_register` 의 입력 의미는 「**같은 실물**의 여러 관측 행」이다.
        줄인 뒤의 행 전부를 한 번에 넣었더니, 각각 **한 번뿐인** 두 물건이
        「2회」가 되어 **둘 다 등록**됐다.
        """
        got = cm.reduce_episode(
            [self._row(1, "가방"), self._row(1, "표")], [],
            segments=ep.segment_texts())
        assert got["counts"]["out"] == 2
        for v in got["registered"].values():
            assert v["registered"] is False, "★다른 실물의 출현이 합쳐졌다"
            assert v["reason"] == "shot_appearances=1"

    def test_one_exception_does_not_register_its_neighbour(self):
        """★예외 판정도 섞이면 안 된다 — 옆 행까지 같이 열린다."""
        flagged = self._row(1, "가방", hard_to_generate=True,
                            viewers_would_notice=True)
        got = cm.reduce_episode([flagged, self._row(1, "표")], [],
                                segments=ep.segment_texts())
        by = got["registered"]
        assert by[flagged["local_id"]]["reason"] == "grounding_exception"
        other = next(k for k in by if k != flagged["local_id"])
        assert by[other]["registered"] is False, "★옆 행까지 열렸다"

    def test_every_kept_row_gets_its_own_verdict(self):
        """★남은 행마다 한 줄 — 빠지면 그 행은 아무도 안 본다."""
        rows = [self._row(1, "가방"), self._row(1, "표"),
                self._row(3, "옛 요금표")]
        got = cm.reduce_episode(rows, [], segments=ep.segment_texts())
        assert set(got["registered"]) == {r["local_id"] for r in rows}

    def test_the_module_exposes_only_the_verified_path(self):
        """★`_` 없는 이름으로 줄이기·등록·발급을 부를 길이 없어야 한다."""
        for name in ("reduce_rows", "should_register", "assign_final_ids",
                     "assert_rows"):
            assert not hasattr(cm, name), f"★{name} 이 공개돼 있다"

    def test_two_chunks_merge_and_register_through_the_endpoint(self):
        """★① 반례를 **끝점으로** — 1장·4장의 `가방`."""
        a, b = self._row(1, "가방"), self._row(4, "가방")
        got = cm.reduce_episode(
            [a, b], [{"remove_local_id": b["local_id"],
                      "keep_local_id": a["local_id"],
                      "relation": cm.REL_SAME}],
            segments=ep.segment_texts())
        assert got["counts"]["out"] == 1
        kept = got["registered"][a["local_id"]]
        assert kept["registered"] is True
        assert kept["reason"] == "shot_appearances=2"

    @pytest.mark.parametrize("order", [0, 1])
    def test_a_split_ledger_conflict_stops_the_same_way_either_order(self, order):
        """★★★AB 든 BA 든 **같은 이유로** 선다 — 순서가 답을 정하면 안 된다."""
        pair = [{"local_id": "o1", "hard_to_generate": False,
                 "viewers_would_notice": False},
                {"local_id": "o1", "hard_to_generate": True,
                 "viewers_would_notice": True}]
        row = self._row(1, "가방",
                        **{cm.EXCEPTION_LEDGER: pair if order == 0
                           else list(reversed(pair))})
        with pytest.raises(AssertionError, match="판정이 둘이다"):
            cm.reduce_episode([row], [], segments=ep.segment_texts())

    def test_a_string_bool_in_the_ledger_stops(self):
        """★`bool("false")` 는 참이다 — 장부에서도 막는다."""
        row = self._row(1, "가방", **{cm.EXCEPTION_LEDGER: [
            {"local_id": "o", "hard_to_generate": "false",
             "viewers_would_notice": "false"}]})
        with pytest.raises(AssertionError, match="bool 이 아니다"):
            cm.reduce_episode([row], [], segments=ep.segment_texts())

    @pytest.mark.parametrize("bad", ["0", True, 1.0, None])
    def test_a_span_coordinate_that_is_not_a_plain_int_stops(self, bad):
        """★`int(x)` 강제를 없앴다 — `"3"` 도 `True` 도 좌표가 아니다."""
        row = self._row(1, "가방")
        row["occurrences"][0]["source_span"]["start"] = bad
        with pytest.raises(AssertionError):
            cm.reduce_episode([row], [], segments=ep.segment_texts())


class TestTheRegistrationLedgerIsPerReferent:
    """★Codex 가 지정한 세 반례를 **공개 끝점**으로 잠근다.

        (a) 무관한 1회 + 1회        → 둘 다 false
        (b) 2회 대상 + 1회 대상     → 앞엣것만 true
        (c) hard&notice + 그냥      → 앞엣것만 true
    """

    _n = {}

    def _row(self, scene, word, occurrence=1, shots=None, **over):
        key = f"scene-{scene}"
        self._n[key] = self._n.get(key, -1) + 1
        d = {"local_id": cm.local_id(key, self._n[key]),
             "owner_type": "prop", "surface_form": word,
             "occurrences": [{"source_span": ep.span_of(scene, word,
                                                        occurrence),
                              "source_quote": word}],
             "shot_binding_status": "bound_complete",
             "shot_appearance_ids": (list(shots) if shots is not None
                                     else [f"s{scene}#1"]),
             # ★기본은 **두 축을 밝힌** 행이다. 안 밝히면 미확정이라
             #  「그냥 미등록」을 뜻하는 시험이 성립하지 않는다.
             "hard_to_generate": False, "viewers_would_notice": False}
        d.update(over)
        return d

    def test_a_unrelated_singles_are_both_false(self):
        got = cm.reduce_episode([self._row(1, "가방"), self._row(1, "표")],
                                [], segments=ep.segment_texts())
        assert all(v["registered"] is False
                   for v in got["registered"].values())
        assert all(v["disposition"] == "not_registered"
                   for v in got["registered"].values())

    def test_b_only_the_twice_seen_one_registers(self):
        """★★두 **씬**에 나온 것만 반복 축으로 산다 — 한 씬 안 두 자리는 1회다."""
        twice = self._row(1, "가방", shots=["s1#1", "s4#1"])   # ★두 샷
        twice["occurrences"].append(
            {"source_span": ep.span_of(4, "가방", 1), "source_quote": "가방"})
        once = self._row(3, "옛 요금표")
        got = cm.reduce_episode([twice, once], [],
                                segments=ep.segment_texts())
        by = got["registered"]
        assert by[twice["local_id"]]["registered"] is True
        assert by[twice["local_id"]]["reason"] == "shot_appearances=2"
        assert by[once["local_id"]]["registered"] is False

    def test_c_only_the_flagged_one_registers(self):
        flagged = self._row(3, "옛 요금표", hard_to_generate=True,
                            viewers_would_notice=True)
        plain = self._row(1, "가방")
        got = cm.reduce_episode([flagged, plain], [],
                                segments=ep.segment_texts())
        by = got["registered"]
        assert by[flagged["local_id"]]["reason"] == "grounding_exception"
        assert by[plain["local_id"]]["registered"] is False

    def test_a_dropped_candidate_gets_no_final_id(self):
        """★★★**뒤집힌 시험** — 앞 판은 「모든 줄이 final_id 를 갖는다」를
        잠갔고, 그건 **잘못된 동작을 계약으로 못박은 것**이었다 (Codex).

        최종 ID 는 「등록된 엔티티의 신원」이지 「이 판에 나온 행의 순번」이
        아니다. 탈락 후보는 durable `local_id` 로 감사하면 된다.
        """
        rows = [self._row(1, "가방"), self._row(1, "표")]
        got = cm.reduce_episode(rows, [], segments=ep.segment_texts())
        assert got["final_ids"] == {}, "★탈락 행에 번호를 줬다"
        for v in got["registered"].values():
            assert v["registered"] is False and v["final_id"] is None

    def test_a_registered_row_keeps_its_id_whatever_drops_around_it(self):
        """★★탈락 후보가 있고 없고가 **등록된 것의 번호를 바꾸면 안 된다**.

        Codex 재현: A(1회 탈락) + B(2회 등록) → A=P01·B=P02 였고,
        B 만 넣으면 B=P01 이었다.
        """
        twice = self._row(1, "가방", shots=["s1#1", "s4#1"])   # ★두 샷
        twice["occurrences"].append(
            {"source_span": ep.span_of(4, "가방", 1), "source_quote": "가방"})
        alone = cm.reduce_episode([twice], [], segments=ep.segment_texts())

        dropped = self._row(1, "표")            # 한 씬 → 탈락
        with_junk = cm.reduce_episode([dropped, twice], [],
                                      segments=ep.segment_texts())
        assert with_junk["final_ids"][twice["local_id"]] == \
            alone["final_ids"][twice["local_id"]], \
            "★탈락한 후보가 등록된 것의 번호를 흔들었다"
        assert dropped["local_id"] not in with_junk["final_ids"]

    def test_the_ledger_never_shows_an_id_the_map_does_not_have(self):
        """★장부의 `final_id` 와 `final_ids` 가 **같은 것**이어야 한다.

        ★`get()` 이 `None` 을 내주는 바람에 「장부에 탈락 번호를 넣는」
        되돌리기를 시험이 못 잡았다. 두 자료를 **맞대어** 본다.
        """
        twice = self._row(2, "정류장 표지")
        twice["occurrences"].append(
            {"source_span": ep.span_of(2, "정류장 표지", 2),
             "source_quote": "정류장 표지"})
        got = cm.reduce_episode([twice, self._row(1, "가방")], [],
                                segments=ep.segment_texts())
        from_ledger = {lid: v["final_id"]
                       for lid, v in got["registered"].items()
                       if v["final_id"] is not None}
        assert from_ledger == got["final_ids"]
        # ★그리고 **등록된 것만** 번호를 갖는다
        assert all(got["registered"][lid]["registered"] is True
                   for lid in got["final_ids"])

    def test_two_registered_rows_get_different_ids(self):
        """★반대쪽 — 등록이 둘이면 서로 다른 번호여야 한다."""
        a = self._row(1, "가방", shots=["s1#1", "s4#1"])   # ★두 샷 → 반복 축
        a["occurrences"].append(
            {"source_span": ep.span_of(4, "가방", 1), "source_quote": "가방"})
        b = self._row(3, "옛 요금표", hard_to_generate=True,
                      viewers_would_notice=True)   # 한 씬 → 예외 축
        got = cm.reduce_episode([a, b], [], segments=ep.segment_texts())
        ids = got["final_ids"]
        assert len(set(ids.values())) == 2 and len(ids) == 2

    def test_an_id_contested_row_is_not_quietly_registered(self):
        """★★ID 를 못 준 행이 「등록됨」으로 흘러가면, 뒤에서 무엇을 가리키는지
        모르는 행이 생긴다. **미확정**으로 세운다."""
        a = self._row(1, "가방")
        b = self._row(1, "가방")          # ★근거가 **똑같다**
        b["occurrences"] = [dict(o) for o in a["occurrences"]]
        got = cm.reduce_episode([a, b], [], segments=ep.segment_texts())
        assert got["id_contested"], "★근거가 같은데 안 다퉜다"
        for lid in got["id_contested"][0]["local_ids"]:
            v = got["registered"][lid]
            assert v["registered"] is False
            assert v["disposition"] == "id_contested"
            assert v["final_id"] is None


class TestTheStoredRowDoesNotDependOnArrivalOrder:
    """★★★**병렬 구간이 도착하는 순서**가 저장될 내용을 바꾸면 안 된다.

    구간을 병렬로 돌리는 것이 (c) 의 요점인데, 합친 행의 `occurrences` 가
    첫 등장 순이면 A→B 와 B→A 가 다른 bytes 를 낸다. 체크포인트가 그때그때
    다르면 하류 지문이 흔들려 **멀쩡한 것이 재생성**된다.

    ★내가 Codex 보다 먼저 훑어서 찾은 것이다 — 오늘 반복된 다섯 부류
    (truthiness · 안 쓰는 값 · 순서 의존 · 조용한 버림 · 두 벌)로 봤다.
    """

    _n = {}

    def _row(self, scene, word, occurrence=1):
        key = f"scene-{scene}"
        self._n[key] = self._n.get(key, -1) + 1
        return {"local_id": cm.local_id(key, self._n[key]),
                "owner_type": "prop", "surface_form": word,
                "occurrences": [{"source_span": ep.span_of(scene, word,
                                                           occurrence),
                                 "source_quote": word}]}

    def test_merging_gives_the_same_row_either_arrival_order(self):
        a, b = self._row(1, "가방"), self._row(4, "가방")
        first = cm.reduce_episode(
            [dict(a), dict(b)],
            [{"remove_local_id": b["local_id"], "keep_local_id": a["local_id"],
              "relation": cm.REL_SAME}], segments=ep.segment_texts())
        second = cm.reduce_episode(
            [dict(b), dict(a)],
            [{"remove_local_id": a["local_id"], "keep_local_id": b["local_id"],
              "relation": cm.REL_SAME}], segments=ep.segment_texts())
        spans = lambda g: [cm.canonical_span(o)
                           for o in g["rows"][0]["occurrences"]]
        assert spans(first) == spans(second), "★도착 순서가 저장 내용을 바꿨다"

    def test_the_occurrences_are_in_span_order(self):
        a, b = self._row(1, "가방"), self._row(4, "가방")
        got = cm.reduce_episode(
            [dict(b), dict(a)],
            [{"remove_local_id": a["local_id"], "keep_local_id": b["local_id"],
              "relation": cm.REL_SAME}], segments=ep.segment_texts())
        got_spans = [cm.canonical_span(o)
                     for o in got["rows"][0]["occurrences"]]
        assert got_spans == sorted(got_spans)

    def test_the_whole_result_is_byte_identical_either_arrival_order(self):
        """★★★**행 배열 자체**가 도착 순서를 타던 것 (Codex).

        한 행 **안**의 출현만 정렬해선 모자랐다 — `rows` 배열이 여전히
        입력 순서라 `json.dumps(sort_keys=True)` bytes 가 갈렸다.
        **ID 만 견주면 이 결함을 또 놓친다** — 직렬화를 통째로 본다.
        """
        import json

        a = self._row(1, "가방")
        b = self._row(1, "표")
        first = cm.reduce_episode([dict(a), dict(b)], [],
                                  segments=ep.segment_texts())
        second = cm.reduce_episode([dict(b), dict(a)], [],
                                   segments=ep.segment_texts())
        assert json.dumps(first, sort_keys=True, ensure_ascii=False) == \
            json.dumps(second, sort_keys=True, ensure_ascii=False), \
            "★도착 순서가 저장될 bytes 를 바꿨다"

    def test_the_rows_are_in_canonical_order(self):
        rows = [self._row(4, "가방"), self._row(1, "표"), self._row(3, "옛 요금표")]
        got = cm.reduce_episode([dict(r) for r in rows], [],
                                segments=ep.segment_texts())
        keys = [(cm.evidence_key(r), str(r["local_id"])) for r in got["rows"]]
        assert keys == sorted(keys), "★행 배열이 도착 순서다"


class TestOrderIndependenceAsAProperty:
    """★★★순서 의존을 **하나씩** 잡다가 세 번 났다 — 성질로 한 번에 건다.

        1차  합친 행의 `occurrences` 가 첫 등장 순
        2차  `rows` 배열이 입력 순
        3차  `bound_short_ids` 가 첫 값만

    구간을 **병렬**로 돌리는 것이 이 구조의 요점이므로, 「도착 순서가 산출을
    안 바꾼다」는 한 자리의 규칙이 아니라 **끝점의 성질**이어야 한다.

    ★그래서 입력 **순열을 전부** 돌려 직렬화 bytes 를 견준다. 다음에 어느
    자리에 순서 의존이 새로 생겨도 여기서 잡힌다.
    """

    _n = {}

    def _row(self, scene, word, occurrence=1, shots=None, **over):
        key = f"scene-{scene}"
        self._n[key] = self._n.get(key, -1) + 1
        d = {"local_id": cm.local_id(key, self._n[key]),
             "owner_type": "prop", "surface_form": word,
             "occurrences": [{"source_span": ep.span_of(scene, word,
                                                        occurrence),
                              "source_quote": word}],
             "shot_binding_status": "bound_complete",
             "shot_appearance_ids": (list(shots) if shots is not None
                                     else [f"s{scene}#1"]),
             # ★기본은 **두 축을 밝힌** 행이다. 안 밝히면 미확정이라
             #  「그냥 미등록」을 뜻하는 시험이 성립하지 않는다.
             "hard_to_generate": False, "viewers_would_notice": False}
        d.update(over)
        return d

    @staticmethod
    def _bytes(result):
        import json

        return json.dumps(result, sort_keys=True, ensure_ascii=False)

    def test_every_permutation_of_plain_rows_gives_the_same_bytes(self):
        import itertools

        rows = [self._row(1, "가방"), self._row(1, "표"),
                self._row(3, "옛 요금표")]
        seen = {self._bytes(cm.reduce_episode(
            [dict(r) for r in perm], [], segments=ep.segment_texts()))
            for perm in itertools.permutations(rows)}
        assert len(seen) == 1, f"★순열마다 다른 산출이 {len(seen)}가지"

    def test_every_permutation_with_a_merge_gives_the_same_bytes(self):
        """★합치기가 끼어도 마찬가지다 — 판정만 같은 것이 아니라 **bytes** 가."""
        import itertools

        a, b = self._row(1, "가방"), self._row(4, "가방")
        c = self._row(3, "옛 요금표", hard_to_generate=True,
                      viewers_would_notice=True)
        dec = [{"remove_local_id": b["local_id"],
                "keep_local_id": a["local_id"], "relation": cm.REL_SAME}]
        seen = {self._bytes(cm.reduce_episode(
            [dict(r) for r in perm], dec, segments=ep.segment_texts()))
            for perm in itertools.permutations([a, b, c])}
        assert len(seen) == 1, f"★순열마다 다른 산출이 {len(seen)}가지"

    def test_the_decision_order_does_not_matter_either(self):
        """★★**판정 목록의 순서**도 산출을 바꾸면 안 된다.

        ★앞 판은 자기 자신으로 합치는 판정을 넣어서(거부됨) 이 자리를 못
        태웠다. **둘이 한 keeper 로 합쳐지는** 판을 써야 `merged` 목록의
        순서가 실제로 갈린다 — 그것이 합친 행의 `occurrences` 를 흔든다.
        """
        import itertools

        keep = self._row(1, "가방")
        g1, g2 = self._row(4, "가방"), self._row(2, "정류장 표지")
        decs = [{"remove_local_id": g["local_id"],
                 "keep_local_id": keep["local_id"], "relation": cm.REL_SAME}
                for g in (g1, g2)]
        rows = [keep, g1, g2]
        seen = {self._bytes(cm.reduce_episode(
            [dict(r) for r in rows], list(perm), segments=ep.segment_texts()))
            for perm in itertools.permutations(decs)}
        assert len(seen) == 1, f"★판정 순서마다 다른 산출이 {len(seen)}가지"


class TestTwoMorePropertiesOfTheEndpoint:
    """★순서 말고 **두 성질** 더 — 지금은 서지만, 깨지면 여기서 잡힌다.

        무관 추가 불변  무관한 행을 더해도 기존 행의 판정·번호가 안 바뀐다
        멱등            줄인 결과를 다시 넣어도 같다

    ★앞서 「탈락 후보가 등록된 것의 번호를 흔든다」가 바로 첫째를 깬 것이었다.
    한 번 겪었으면 **성질로** 걸어 둔다 — 다음 자리에서 또 나기 때문이다.
    """

    _n = {}

    def _row(self, scene, word, occurrence=1, shots=None, **over):
        key = f"scene-{scene}"
        self._n[key] = self._n.get(key, -1) + 1
        d = {"local_id": cm.local_id(key, self._n[key]),
             "owner_type": "prop", "surface_form": word,
             "occurrences": [{"source_span": ep.span_of(scene, word,
                                                        occurrence),
                              "source_quote": word}],
             "shot_binding_status": "bound_complete",
             "shot_appearance_ids": (list(shots) if shots is not None
                                     else [f"s{scene}#1"]),
             # ★기본은 **두 축을 밝힌** 행이다. 안 밝히면 미확정이라
             #  「그냥 미등록」을 뜻하는 시험이 성립하지 않는다.
             "hard_to_generate": False, "viewers_would_notice": False}
        d.update(over)
        return d

    def _twice(self):
        r = self._row(2, "정류장 표지")
        r["occurrences"].append(
            {"source_span": ep.span_of(2, "정류장 표지", 2),
             "source_quote": "정류장 표지"})
        return r

    def test_adding_an_unrelated_row_changes_nothing_for_the_others(self):
        base = self._twice()
        alone = cm.reduce_episode([dict(base)], [],
                                  segments=ep.segment_texts())
        lid = base["local_id"]
        for extra in (self._row(1, "가방"),
                      self._row(3, "옛 요금표", hard_to_generate=True,
                                viewers_would_notice=True)):
            with_extra = cm.reduce_episode([dict(base), dict(extra)], [],
                                           segments=ep.segment_texts())
            assert with_extra["registered"][lid] == alone["registered"][lid], \
                "★무관한 행이 기존 판정을 바꿨다"

    def test_feeding_the_result_back_in_is_stable(self):
        """★멱등 — 재개나 재병합에서 같은 것을 두 번 돌려도 안 흔들린다."""
        import json

        first = cm.reduce_episode([dict(self._twice())], [],
                                  segments=ep.segment_texts())
        again = cm.reduce_episode([dict(r) for r in first["rows"]], [],
                                  segments=ep.segment_texts())
        dump = lambda x: json.dumps(x, sort_keys=True, ensure_ascii=False)
        assert dump(first["rows"]) == dump(again["rows"])
        assert first["registered"] == again["registered"]
        assert first["final_ids"] == again["final_ids"]


class TestPartOfIsAlsoOrderIndependent:
    """★★★`part_of` **판정 순열** — 앞 성질 시험이 `same_referent` 만 태웠다.

    「입력 순열 전체」로는 **판정 순열**을 못 대신한다. 관계 장부가 판정
    도착 순서대로 쌓이면, 같은 판인데 CP bytes 가 갈린다.
    """

    _n = {}

    def _row(self, scene, word, owner="prop"):
        key = f"scene-{scene}"
        self._n[key] = self._n.get(key, -1) + 1
        return {"local_id": cm.local_id(key, self._n[key]),
                "owner_type": owner, "surface_form": word,
                "occurrences": [{"source_span": ep.span_of(scene, word),
                                 "source_quote": word}],
                "shot_binding_status": "bound_complete",
                "shot_appearance_ids": [f"s{scene}#1"],
                "hard_to_generate": False, "viewers_would_notice": False}

    def _setup(self):
        whole = self._row(3, "대합실", owner="location")
        p0 = self._row(3, "긴 걸상", owner="location_part")
        p1 = self._row(2, "정류장 표지", owner="location_part")
        decs = [{"remove_local_id": p["local_id"],
                 "keep_local_id": whole["local_id"],
                 "relation": cm.REL_PART_OF} for p in (p0, p1)]
        return [whole, p0, p1], decs

    def test_every_decision_permutation_gives_the_same_bytes(self):
        import itertools
        import json

        rows, decs = self._setup()
        seen = {json.dumps(cm.reduce_episode(
            [dict(r) for r in rows], list(perm),
            segments=ep.segment_texts()), sort_keys=True, ensure_ascii=False)
            for perm in itertools.permutations(decs)}
        assert len(seen) == 1, f"★판정 순서마다 다른 산출이 {len(seen)}가지"

    def test_the_parts_are_in_canonical_order(self):
        rows, decs = self._setup()
        got = cm.reduce_episode([dict(r) for r in rows],
                                list(reversed(decs)),
                                segments=ep.segment_texts())
        keys = [(x["part"], x["whole"]) for x in got["part_of"]]
        assert keys == sorted(keys), "★관계 장부가 판정 도착 순서다"

    def test_the_parts_still_survive_as_their_own_rows(self):
        """★positive control — 정렬한다고 **부분이 사라지면** 안 된다."""
        rows, decs = self._setup()
        got = cm.reduce_episode([dict(r) for r in rows], decs,
                                segments=ep.segment_texts())
        assert got["counts"]["out"] == 3, "★부분을 합쳐 없앴다"
        assert got["counts"]["part_of"] == 2
        assert len(got["part_of"]) == 2


@pytest.mark.parametrize("relation", cm.RELATIONS)
class TestEveryRelationKindGetsItsOwnPermutation:
    """★★★**관계 종류마다** 갈래가 따로 있다.

    앞 판에서 `same_referent` 만 태워 `part_of` 가 판정 도착 순서를 타는 것을
    놓쳤다. 목록을 손으로 적으면 **종류가 늘 때 같은 구멍이 다시 난다** —
    그래서 `cm.RELATIONS` **그 자체**로 돌린다. 새 관계를 더하면 이 시험이
    자동으로 그것도 태운다.
    """

    _n = {}

    def _row(self, scene, word, owner):
        key = f"scene-{scene}"
        self._n[key] = self._n.get(key, -1) + 1
        return {"local_id": cm.local_id(key, self._n[key]),
                "owner_type": owner, "surface_form": word,
                "occurrences": [{"source_span": ep.span_of(scene, word),
                                 "source_quote": word}]}

    def _case(self, relation):
        """관계 종류에 맞는 소유 갈래로 판을 짠다."""
        if relation == cm.REL_PART_OF:
            keep = self._row(3, "대합실", "location")
            gone = [self._row(3, "긴 걸상", "location_part"),
                    self._row(2, "정류장 표지", "location_part")]
        else:
            keep = self._row(1, "가방", "prop")
            gone = [self._row(4, "가방", "prop"), self._row(2, "표", "prop")]
        decs = [{"remove_local_id": g["local_id"],
                 "keep_local_id": keep["local_id"],
                 "relation": relation} for g in gone]
        return [keep] + gone, decs

    def test_all_decision_permutations_serialize_to_one(self, relation):
        import itertools
        import json

        rows, decs = self._case(relation)
        seen = {json.dumps(cm.reduce_episode(
            [dict(r) for r in rows], list(perm),
            segments=ep.segment_texts()), sort_keys=True, ensure_ascii=False)
            for perm in itertools.permutations(decs)}
        assert len(seen) == 1, (
            f"★{relation} 판정 순서마다 다른 산출이 {len(seen)}가지")

    def test_the_relation_actually_took_effect(self, relation):
        """★positive control — 판정이 **아무 일도 안 했으면** 위 시험은 공짜다."""
        rows, decs = self._case(relation)
        plain = cm.reduce_episode([dict(r) for r in rows], [],
                                  segments=ep.segment_texts())
        got = cm.reduce_episode([dict(r) for r in rows], decs,
                                segments=ep.segment_texts())
        assert got != plain, f"★{relation} 판정이 산출을 안 바꿨다"
        # ★`counts` 칸 이름은 관계 이름과 **다르다** — `same_referent` 는
        #  「몇 행이 접혔나(`merged`)」로 센다. 시험 편의로 프로덕션 칸을
        #  바꾸지 않고, 여기서 잇는다. 새 관계를 더하면 여기서 KeyError 로
        #  막히므로 「세는 칸을 안 만들었다」가 조용히 지나가지 않는다.
        counted = {cm.REL_SAME: "merged", cm.REL_PART_OF: "part_of"}[relation]
        assert got["counts"][counted], f"★{relation} 이 {counted} 로 안 세어졌다"