"""C(c) — 병렬 구간을 줄이는 계약. ★Codex 가 지정한 **여섯 반례**. 같은 owner 같은 실물 · 같은 이름 다른 실물 · 한 chunk 에 2회 · part_of · 다섯 owner · provenance 누락 ★이 계약은 **아직 아무 데도 안 붙는다**. 구조가 승인되기 전에는 배선하지 않는다. 여기 있는 것은 「붙이면 이렇게 동작한다」를 미리 잠근 것이다. ★fixture 이름은 아무 뜻이 없다 — 계약이 이름을 **안 본다**. """ import pytest from app.modules.pipeline import grounding_chunk_merge as cm from app.modules.pipeline.grounding_binding import FIELD def _occ(seg, start, end=None, quote=None): """★신원은 **정본 span** 이다 — 모델이 쓴 인용이 아니다.""" end = start + 5 if end is None else end return {"source_span": {"segment_id": seg, "start": start, "end": end}, "source_quote": quote if quote is not None else f"q-{seg}{start}"} def _span_of(name): """★자리 이름 → **안정 span**. 같은 이름이면 언제나 같은 자리다. 시험이 「같은 자리」·「다른 자리」를 뜻대로 쓰려면 이름과 좌표가 1:1 이어야 한다 — 순번으로 매기면 목록 순서에 따라 뜻이 바뀐다. ★★자리 이름마다 **다른 씬**에 둔다 (2026-08-31). 등록 축이 세는 것은 **등장한 씬의 수**라, 한 씬에 몰아 두면 「두 자리」가 「1회 등장」이 된다. 같은 씬 안의 서로 다른 자리는 `_same_scene_span()` 으로 만든다. """ n = abs(hash(("span", name))) % 1000 return _occ(f"seg-{name}", n * 10) def _same_scene_span(name, scene="seg1"): """**같은 씬 안**의 다른 자리. ★등장 축으로는 1회다.""" n = abs(hash(("span", name))) % 1000 return _occ(scene, n * 10) def _row(chunk, idx, owner="prop", anchors=(), shots=None, **over): """★`anchors` 는 **자리 이름**이다 — 같은 이름이면 같은 span. ★★반복 축은 이제 **샷 단위**다. `shots` 를 안 주면 자리마다 샷 하나를 만들어 `bound_complete` 로 둔다(자리 둘 = 샷 둘). 샷 결속을 **아예 안 주려면** `shots=None` 대신 `shot_binding_status` 를 직접 덮어쓴다 — 그러면 **미확정**이 된다(그게 새 계약이다). """ names = list(dict.fromkeys(anchors)) d = {"local_id": cm.local_id(chunk, idx), "chunk_id": chunk, "owner_type": owner, "surface_form": f"{chunk}-{idx}", "occurrences": [_span_of(a) for a in names], "shot_binding_status": "bound_complete", "shot_appearance_ids": (list(shots) if shots is not None else [f"s{a}#1" for a in names]), "hard_to_generate": False, "viewers_would_notice": False} if over.pop("no_flags", False): # ★두 축을 **안 밝힌** 행 — 미확정을 뜻한다. 「합치면서 없던 판정을 # 심지 않는다」 같은 시험이 이것을 쓴다. d.pop("hard_to_generate"); d.pop("viewers_would_notice") d.update(over) return d def _dec(src, dst, rel=cm.REL_SAME): return {"remove_local_id": src, "keep_local_id": dst, "relation": rel} class TestTheInputIsCheckedBeforeReducing: """★계약 3 — 앞 판은 빈 id 를 조용히 버리고 겹친 것은 **덮어썼다**.""" def test_a_blank_local_id_stops(self): with pytest.raises(AssertionError, match="local_id"): cm._reduce_rows([{"local_id": "", "owner_type": "prop", "occurrences": []}], []) def test_a_duplicate_local_id_stops(self): r = _row("c1", 0, anchors=("S1",)) with pytest.raises(AssertionError, match="겹친다"): cm._reduce_rows([r, dict(r)], []) def test_an_unknown_owner_stops(self): with pytest.raises(AssertionError, match="모르는 owner"): cm._reduce_rows([_row("c1", 0, "없는갈래")], []) def test_a_missing_occurrences_field_stops(self): bad = _row("c1", 0) del bad["occurrences"] with pytest.raises(AssertionError, match="occurrences"): cm._reduce_rows([bad], []) def test_a_missing_span_stops(self): """★모델이 쓴 인용은 **신원이 못 된다** — 정본 좌표가 있어야 한다.""" bad = _row("c1", 0) bad["occurrences"] = [{"source_quote": "가"}] with pytest.raises(AssertionError, match="segment_id"): cm._reduce_rows([bad], []) @pytest.mark.parametrize("sp", [ {"segment_id": "s1", "start": -1, "end": 5}, {"segment_id": "s1", "start": "0", "end": 5}, {"segment_id": "s1", "start": True, "end": 5}, {"segment_id": "s1", "start": 5, "end": 5}, {"segment_id": "s1", "start": 9, "end": 5}, ]) def test_a_broken_span_stops(self, sp): bad = _row("c1", 0) bad["occurrences"] = [{"source_span": sp, "source_quote": "가"}] with pytest.raises(AssertionError): cm._reduce_rows([bad], []) def test_a_quote_that_is_not_at_that_place_stops(self): """★★원문을 주면 **거기 실제로 있는지**까지 본다 — 지어낸 것이다.""" row = _row("c1", 0) row["occurrences"] = [{"source_span": {"segment_id": "s1", "start": 0, "end": 5}, "source_quote": "딴말"}] segs = {"s1": "0123456789"} with pytest.raises(AssertionError, match="그 자리에 없다"): cm._assert_rows([row], segments=segs) row["occurrences"][0]["source_quote"] = "01234" cm._assert_rows([row], segments=segs) # ★맞으면 통과한다 @pytest.mark.parametrize("bad", ["false", "true", 1, 0, "", None]) def test_a_flag_that_is_not_exactly_bool_stops(self, bad): """★`bool("false")` 는 **참**이다. 이 부류는 이 저장소에서 반복해 났다.""" row = _row("c1", 0, anchors=("S1",), hard_to_generate=bad) if bad is None: cm._reduce_rows([row], []) # None 은 「없음」이라 통과한다 return with pytest.raises(ValueError, match="bool"): cm._reduce_rows([row], []) def test_the_owners_come_from_the_a0_schema(self): """★두 벌로 적으면 A0 가 갈래를 늘려도 여기는 모른다.""" from app.modules.pipeline import grounding_a0 as a0 sch = a0.load_pack()["stems"][a0.SCHEMA_STEM]["content"] import json as _j if isinstance(sch, str): sch = _j.loads(sch) want = (sch.get("schema", sch)["properties"]["candidates"]["items"] ["properties"]["owner_type"]["enum"]) assert list(cm.OWNERS) == list(want) def test_the_threshold_has_one_source(self): """★★시험이 프로덕션 **소스를 정규식으로 읽던** 것도 되돌렸다. 검사를 시험으로 옮긴 것만으로는 **SOT 가 하나가 안 된다** (Codex). 값은 `grounding_entity_contract` 한 곳에 있고, 여기는 그것을 **가리킬 뿐**이다. """ import inspect from app.modules.pipeline import grounding_entity_contract as ec assert cm.MIN_OCCURRENCES == ec.ENTITY_MIN_OCCURRENCES assert cm._PREFIX is ec.OWNER_PREFIX # ★★값 비교로는 **못 잡는다** — 파이썬이 작은 정수를 공유해서 # `2 is 2` 가 참이다. 되돌려 보니 0 failed 였다. 그래서 **이 모듈이 # 숫자를 다시 적지 않는지**를 본다. src = inspect.getsource(cm) i = src.index("MIN_OCCURRENCES =") line = src[i:src.index("\n", i)] assert "ENTITY_MIN_OCCURRENCES" in line, \ f"★문턱을 여기서 다시 적었다: {line!r}" class TestCyclesAndConflicts: """★계약 4 — 도착 순서로 keeper 가 갈리면 병렬에서 매번 다른 답이 나온다.""" def test_a_two_row_cycle_is_refused_whole(self): rows = [_row("c1", 0, anchors=("S1",)), _row("c2", 0, anchors=("S2",))] got = cm._reduce_rows(rows, [_dec("c1#0", "c2#0"), _dec("c2#0", "c1#0")]) assert got["counts"]["out"] == 2, "★고리에서 하나를 임의로 남겼다" assert set(got["refused"]) == {"c1#0", "c2#0"} def test_a_three_row_cycle_is_refused_whole(self): rows = [_row(f"c{i}", 0, anchors=(f"S{i}",)) for i in (1, 2, 3)] got = cm._reduce_rows(rows, [_dec("c1#0", "c2#0"), _dec("c2#0", "c3#0"), _dec("c3#0", "c1#0")]) assert got["counts"]["out"] == 3 def test_a_chain_without_a_cycle_still_works(self): """★고리가 아니면 막지 않는다 — 문을 필요 이상으로 넓히지 않는다.""" rows = [_row(f"c{i}", 0, anchors=(f"S{i}",)) for i in (1, 2, 3)] got = cm._reduce_rows(rows, [_dec("c1#0", "c2#0")]) assert got["counts"]["out"] == 2 def test_both_relations_on_one_row_is_refused(self): rows = [_row("c1", 0, "location", anchors=("S1",)), _row("c1", 1, "location_part", anchors=("S2",))] got = cm._reduce_rows(rows, [_dec("c1#1", "c1#0"), _dec("c1#1", "c1#0", cm.REL_PART_OF)]) assert got["counts"]["out"] == 2 assert "part_of" in got["refused"]["c1#1"] class TestExceptionNeedsBothOnOneRow: """★★계약 1 — 갈래로 `any` 를 따로 모으면 아무도 안 한 판정이 선다.""" def test_hard_here_and_notice_there_is_not_an_exception(self): rows = [_row("c1", 0, anchors=("S1",), hard_to_generate=True), _row("c5", 2, anchors=("S1",), viewers_would_notice=True)] ok, why = cm._should_register(rows) assert ok is not True, "★서로 다른 행의 두 flag 로 예외가 섰다" # ★★미확정은 **미등록이 아니다** — 반복 축도 못 정했으면 `None` 이다 assert ok is None assert why == "exception_unresolved" def test_both_on_one_row_is_an_exception(self): rows = [_row("c1", 0, anchors=("S1",), hard_to_generate=True, viewers_would_notice=True), _row("c5", 2, anchors=("S1",))] assert cm._should_register(rows) == (True, "grounding_exception") def test_a_split_verdict_is_unresolved_not_a_plain_no(self): """★「예외 아님」과 「구간마다 갈렸다」는 다르다.""" rows = [_row("c1", 0, anchors=("S1",), hard_to_generate=True)] assert cm._should_register(rows)[1] == "exception_unresolved" def test_the_merge_does_not_undo_the_same_row_rule(self): """★★★**합치면 규칙이 되돌아가던 것** (Codex 재현). `reduce_rows` 가 두 bool 을 각각 OR 해서 한 행의 (T, T) 로 만들었다 — 합치기 전엔 미확정인데 합친 뒤엔 예외가 섰다. 앞서 고친 바로 그 규칙이 **reducer 뒤에서** 되살아난 것이다. bool 을 합치지 말고 **판정을 나른다.** """ split = [_row("c1", 0, anchors=("S1",), hard_to_generate=True), _row("c5", 2, anchors=("S1",), viewers_would_notice=True)] before = cm._should_register(split) got = cm._reduce_rows([dict(r) for r in split], [_dec("c5#2", "c1#0")]) after = cm._should_register(got["rows"]) assert before == after == (None, "exception_unresolved") def test_a_real_exception_survives_the_merge(self): """★반대쪽도 본다 — 진짜 예외를 합치면서 잃으면 그것도 결함이다.""" both = [_row("c1", 0, anchors=("S1",), hard_to_generate=True, viewers_would_notice=True), _row("c5", 2, anchors=("S1",))] got = cm._reduce_rows([dict(r) for r in both], [_dec("c5#2", "c1#0")]) assert cm._should_register(got["rows"]) == (True, "grounding_exception") def test_the_original_row_verdicts_are_not_overwritten(self): """★어느 구간이 뭐라 했는지가 사라지면 되짚을 수 없다.""" rows = [_row("c1", 0, anchors=("S1",), no_flags=True, hard_to_generate=True), _row("c5", 2, anchors=("S2",), no_flags=True, viewers_would_notice=True)] got = cm._reduce_rows([dict(r) for r in rows], [_dec("c5#2", "c1#0")]) keep = got["rows"][0] assert keep["hard_to_generate"] is True assert "viewers_would_notice" not in keep, \ "★합치면서 없던 판정을 심었다" assert keep[cm.EXCEPTION_STATE] == cm.EXC_UNRESOLVED def test_the_removed_rows_verdict_survives_in_the_ledger(self): """★★★**지운 행이 뭐라 했는지가 결과에 남는가** (Codex). 앞 판은 「둘 다 참인 행」만 담아서, 지워진 행이 `notice=True` 였다는 사실이 **결과 어디에도 없었다**. 상태 문자열 하나로는 못 되짚는다. """ rows = [_row("c1", 0, anchors=("S1",), hard_to_generate=True), _row("c2", 0, anchors=("S1",), viewers_would_notice=True)] got = cm._reduce_rows([dict(r) for r in rows], [_dec("c2#0", "c1#0")]) led = got["rows"][0][cm.EXCEPTION_LEDGER] assert {e["local_id"] for e in led} == {"c1#0", "c2#0"} gone = next(e for e in led if e["local_id"] == "c2#0") assert gone["viewers_would_notice"] is True, \ "★지운 행의 판정이 사라졌다" assert got["rows"][0][cm.EXCEPTION_STATE] == cm.EXC_UNRESOLVED def test_the_ledger_is_flattened_and_deduped_on_re_merge(self): """★두 번 합쳐도 장부가 겹치거나 잘리지 않는다.""" rows = [_row("c1", 0, anchors=("S1",), hard_to_generate=True), _row("c2", 0, anchors=("S1",), viewers_would_notice=True), _row("c3", 0, anchors=("S1",))] first = cm._reduce_rows([dict(r) for r in rows], [_dec("c2#0", "c1#0")]) again = cm._reduce_rows([dict(r) for r in first["rows"]], [_dec("c3#0", "c1#0")]) led = again["rows"][0][cm.EXCEPTION_LEDGER] assert sorted(e["local_id"] for e in led) == ["c1#0", "c2#0", "c3#0"] assert len(led) == 3, "★겹쳤다" def test_no_flags_at_all_is_a_plain_no(self): rows = [_row("c1", 0, anchors=("S1",))] assert cm._should_register(rows) == (False, "shot_appearances=1") class TestLocalIdIsCodeIssued: """★계약 3 — 모델이 안정 ID 를 써내는 안은 안 된다.""" def test_it_is_deterministic_and_globally_unique(self): assert cm.local_id("c1", 0) == "c1#0" assert cm.local_id("c1", 0) != cm.local_id("c2", 0) @pytest.mark.parametrize("bad", ["", " "]) def test_a_blank_chunk_stops(self, bad): with pytest.raises(ValueError, match="chunk_id"): cm.local_id(bad, 0) @pytest.mark.parametrize("bad", [True, "0", -1, 1.0]) def test_the_row_index_must_be_a_plain_non_negative_int(self, bad): with pytest.raises(ValueError, match="row_index"): cm.local_id("c1", bad) class TestTwiceInOneChunkCounts: """★★계약 2 — `appears_here: bool` 이면 「2회 이상」이 그 자리에서 깨진다.""" def test_two_shots_count_as_two(self): rows = [_row("c1", 0, anchors=("S1", "S2"))] assert cm.appearance_count(rows) == (2, False) ok, why = cm._should_register(rows) assert ok and why == "shot_appearances=2" def test_two_places_in_one_scene_count_as_one(self): """★★★유료 주행 1회가 드러낸 것 (2026-08-31). 모델은 같은 것을 여러 말로 부른다 — 한 씬 안에서 요금표를 「옛 요금표」·「가로로 길게 켠 나무판」·「판」 셋으로 불렀다. 자리 수로 세면 **3회**가 되어 한 번 나온 것이 반복 축으로 살고, 그러면 이 실험의 핵심인 **고증 예외 축을 영영 못 잰다.** production 은 `entity_filter._appearance_count` 로 **등장한 자리의 수**를 센다 — 부른 횟수가 아니다. """ row = {"local_id": cm.local_id("c1", 0), "owner_type": "prop", "surface_form": "x", "occurrences": [_same_scene_span("A"), _same_scene_span("B"), _same_scene_span("C")], "shot_binding_status": "bound_complete", "shot_appearance_ids": ["s1#1"]} # ★한 샷에서 세 번 불렀다 assert cm.unique_anchor_count([row]) == 3, "★자리는 셋이 맞다" assert cm.appearance_count([row]) == (1, False), "★등장은 한 샷이다" assert cm.scene_appearance_count([row]) == 1 ok, why = cm._should_register([row]) assert ok is False and why == "shot_appearances=1" def test_the_same_thing_named_three_ways_still_needs_the_exception(self): """★positive control — 그래서 두 축이 서면 **예외 축으로** 산다.""" row = {"local_id": cm.local_id("c1", 0), "owner_type": "prop", "surface_form": "x", "hard_to_generate": True, "viewers_would_notice": True, "occurrences": [_same_scene_span("A"), _same_scene_span("B"), _same_scene_span("C")], "shot_binding_status": "bound_complete", "shot_appearance_ids": ["s1#1"]} ok, why = cm._should_register([row]) assert ok and why == "grounding_exception" def test_the_same_anchor_twice_counts_once(self): rows = [_row("c1", 0, anchors=("S1", "S1"))] assert cm.unique_anchor_count(rows) == 1 def test_one_occurrence_alone_is_not_registered(self): ok, _w = cm._should_register([_row("c1", 0, anchors=("S1",))]) assert ok is False def test_one_occurrence_but_hard_and_noticed_is_registered(self): """★고증 예외 축 — **한 번만 나와도** 등록한다.""" ok, why = cm._should_register([_row( "c1", 0, anchors=("S1",), hard_to_generate=True, viewers_would_notice=True)]) assert ok and why == "grounding_exception" def test_hard_without_notice_is_not_the_exception(self): """★두 조건이 **AND** 다 — 하나만으로 열면 예외가 아니라 통과문이 된다. ★notice 를 **안 밝힌** 것은 「아니다」가 아니라 **미확정**이다 (2026-08-31). 어느 쪽이든 **등록은 아니다**. """ ok, _w = cm._should_register([_row( "c1", 0, anchors=("S1",), hard_to_generate=True)]) assert ok is not True def test_one_flag_on_is_unresolved_not_a_plain_no(self): """★한쪽만 켜진 것은 **「아님」이 아니라 미확정**이다 (기존 계약).""" ok, why = cm._should_register([_row( "c1", 0, anchors=("S1",), hard_to_generate=True, viewers_would_notice=False)]) assert ok is None and why == "exception_unresolved" def test_both_flags_off_is_a_plain_no(self): """★positive control — 둘 다 **거짓**이라고 밝히면 미등록이다.""" ok, why = cm._should_register([_row( "c1", 0, anchors=("S1",), hard_to_generate=False, viewers_would_notice=False)]) assert ok is False and why == "shot_appearances=1" def test_the_two_axes_are_a_union_across_chunks(self): rows = [_row("c1", 0, anchors=("S1",)), _row("c9", 3, anchors=("S9",))] ok, why = cm._should_register(rows) assert ok and why == "shot_appearances=2" class TestSameOwnerCrossChunk: """★계약 1 — 병렬 구간의 핵심은 **같은 owner 안의 중복**이다.""" def test_two_chunks_holding_one_thing_become_one_row(self): rows = [_row("c1", 0, anchors=("S1",)), _row("c5", 2, anchors=("S5",))] got = cm._reduce_rows(rows, [_dec("c5#2", "c1#0")]) assert got["counts"]["out"] == 1 keep = got["rows"][0] assert keep["local_id"] == "c1#0" # ★출현 신원은 **정본 span** 이다 — 두 자리가 다 남아야 하고, # **span 순**이다(도착 순서가 저장 내용을 바꾸면 안 된다). got = [cm.canonical_span(o) for o in keep["occurrences"]] assert sorted(got) == sorted( [cm.canonical_span(_span_of("S1")), cm.canonical_span(_span_of("S5"))]) assert got == sorted(got), "★도착 순서가 남았다" assert keep["merged_from"] == ["c5#2"] def test_the_union_makes_it_registerable(self): """★합치기 **전에는** 둘 다 1회라 등록이 안 된다 — 합쳐야 2회다.""" rows = [_row("c1", 0, anchors=("S1",)), _row("c5", 2, anchors=("S5",))] assert cm.unique_anchor_count([rows[0]]) == 1 assert cm._should_register([rows[0]])[0] is False got = cm._reduce_rows(rows, [_dec("c5#2", "c1#0")]) assert cm._should_register(got["rows"])[0] is True def test_the_exception_verdict_survives_the_merge(self): """★★**bool 이 아니라 판정**이 따라온다 — bool 두 개를 OR 하면 갈린 판정이 예외로 둔갑한다(아래 반례).""" # ★자리를 **같게** 둔다 — 안 그러면 합쳐서 출현이 2가 되고 # 그쪽 축이 먼저 서서 예외 축을 못 본다. rows = [_row("c1", 0, anchors=("S1",)), _row("c5", 2, anchors=("S1",), hard_to_generate=True, viewers_would_notice=True)] got = cm._reduce_rows([dict(r) for r in rows], [_dec("c5#2", "c1#0")]) assert cm.unique_anchor_count(got["rows"]) == 1 assert got["rows"][0][cm.EXCEPTION_STATE] == cm.EXC_YES assert cm._should_register(got["rows"])[1] == "grounding_exception" class TestSameNameDifferentThing: """★이름이 같아도 모델이 안 합치라면 **안 합친다**. 계약이 이름을 안 본다.""" def test_without_a_decision_nothing_is_merged(self): rows = [_row("c1", 0, anchors=("S1",), surface_form="같은 이름"), _row("c5", 2, anchors=("S5",), surface_form="같은 이름")] got = cm._reduce_rows(rows, []) assert got["counts"]["out"] == 2, "★이름이 같다고 합쳤다" def test_changing_only_the_name_changes_nothing(self): """★★**뒤집힌 시험** — 앞 판은 `assert ... or True` 라 **언제나 통과**했다 (Codex BLOCK-6). 이제 프로덕션 함수를 두 번 태워 **이름만 바꾼 입력**이 같은 결과를 내는지 본다. """ base = [_row("c1", 0, anchors=("S1",)), _row("c5", 2, anchors=("S5",))] a = cm._reduce_rows([dict(r) for r in base], []) renamed = [dict(r, surface_form="완전히 다른 말") for r in base] b = cm._reduce_rows(renamed, []) assert a["counts"] == b["counts"] assert [r["local_id"] for r in a["rows"]] == \ [r["local_id"] for r in b["rows"]] def test_making_two_names_identical_does_not_merge_them(self): """★이름을 **같게** 만들어도 안 합친다 — 계약이 이름을 안 본다.""" same = [_row("c1", 0, anchors=("S1",), surface_form="같"), _row("c5", 2, anchors=("S5",), surface_form="같")] assert cm._reduce_rows(same, [])["counts"]["out"] == 2 class TestPartOfIsNotAMerge: """★★계약 5 — 부분을 전체에 합치면 **그 부분이 사라진다**.""" def test_a_part_stays_its_own_row(self): rows = [_row("c1", 0, "location", anchors=("S1",)), _row("c1", 1, "location_part", anchors=("S1",))] got = cm._reduce_rows(rows, [_dec("c1#1", "c1#0", cm.REL_PART_OF)]) assert got["counts"]["out"] == 2, "★부분을 합쳐 없앴다" assert got["part_of"] == [{"part": "c1#1", "whole": "c1#0"}] def test_a_missing_relation_is_not_guessed(self): """★빠진 `relation` 을 `same_referent` 로 짐작하면 부분이 지워진다.""" rows = [_row("c1", 0, "location"), _row("c1", 1, "location_part")] got = cm._reduce_rows(rows, [{"remove_local_id": "c1#1", "keep_local_id": "c1#0"}]) assert got["counts"]["out"] == 2 assert "relation" in got["refused"]["c1#1"] class TestAllFiveOwners: """★계약 1 — 다섯 갈래를 전부 받는다.""" def test_every_owner_survives_a_no_op_reduce(self): """★행 배열이 **정본 순서**로 나오므로 입력 순서와 다를 수 있다 — 중요한 것은 **하나도 안 사라지는 것**이다.""" rows = [_row("c1", i, o, anchors=(f"S{i}",)) for i, o in enumerate(cm.OWNERS)] got = cm._reduce_rows(rows, []) assert sorted(r["owner_type"] for r in got["rows"]) == \ sorted(cm.OWNERS) keys = [(cm.evidence_key(r), str(r["local_id"])) for r in got["rows"]] assert keys == sorted(keys), "★행 배열이 도착 순서다" def test_a_cross_owner_same_referent_is_refused(self): """★owner 가 다른데 「같은 실물」이라면 못 정한 것이다.""" rows = [_row("c1", 0, "prop"), _row("c1", 1, "character")] got = cm._reduce_rows(rows, [_dec("c1#1", "c1#0")]) assert got["counts"]["out"] == 2 assert "owner" in got["refused"]["c1#1"] class TestEveryRemoveNeedsOneValidKeeper: """★★계약 4 — 「근거를 든 행만 보호」로는 구간 행이 실제로 사라진다.""" def test_an_unknown_keeper_refuses(self): rows = [_row("c1", 0, anchors=("S1",))] got = cm._reduce_rows(rows, [_dec("c1#0", "c9#9")]) assert got["counts"]["out"] == 1 assert "모르는 keep" in got["refused"]["c1#0"] def test_merging_into_itself_refuses(self): rows = [_row("c1", 0)] got = cm._reduce_rows(rows, [_dec("c1#0", "c1#0")]) assert got["counts"]["out"] == 1 def test_two_keepers_refuse(self): rows = [_row("c1", 0), _row("c2", 0), _row("c3", 0)] got = cm._reduce_rows(rows, [_dec("c1#0", "c2#0"), _dec("c1#0", "c3#0")]) assert got["counts"]["out"] == 3 assert "keeper 가" in got["refused"]["c1#0"] def test_a_keeper_that_is_also_removed_refuses(self): rows = [_row("c1", 0), _row("c2", 0), _row("c3", 0)] got = cm._reduce_rows(rows, [_dec("c1#0", "c2#0"), _dec("c2#0", "c3#0")]) assert "c1#0" in got["refused"] assert got["counts"]["out"] >= 2 def test_a_row_without_provenance_is_still_protected(self): """★★여기가 기존 `entity_merge` 와 갈리는 자리다 — **모든** remove 를 본다. 근거 칸이 없다고 그냥 지우면 구간 행이 사라진다.""" rows = [_row("c1", 0), _row("c2", 0)] assert FIELD not in rows[0] got = cm._reduce_rows(rows, [_dec("c1#0", "c9#9")]) assert got["counts"]["out"] == 2 class TestTheFinalIdIsDeterministic: """★★★**모델이 행 순서만 바꿔도** 번호가 맞바뀌던 것 (Codex 재현). run1 근거A → P01 · 근거B → P02 run2 (행 순서만 반대) 근거B → P01 · 근거A → P02 앞 판의 정렬 열쇠 마지막이 `chunk#row_index` 였다 — 그건 **모델이 낸 순서**다. 이제 열쇠는 `owner + 모든 (anchor, quote)` 로, **원문에서 확인되는 것**뿐이다. """ @staticmethod def _ev(lid, q, start=70, owner="prop"): """★신원은 **정본 span**. `q` 는 모델이 쓴 인용이라 audit 일 뿐이다.""" return {"local_id": lid, "owner_type": owner, "occurrences": [{"source_span": {"segment_id": "s1", "start": start, "end": start + 5}, "source_quote": q}]} def test_changing_only_the_quote_does_not_swap_the_numbers(self): """★★★**인용 길이만 바꿔도 번호가 맞바뀌던 것** (Codex 재현). A 의 인용을 `A-short` → `Z-longer-valid-span` 으로만 바꾸자 A 가 P01 → P02, B 가 P02 → P01 이 됐다. 모델이 쓴 문자열을 정렬 열쇠에 넣었기 때문이다. 이제 열쇠는 **정본 span** 뿐이다. """ run1 = cm._assign_final_ids([self._ev("c1#0", "A-short", 0), self._ev("c1#1", "B-quote", 100)])["ids"] run2 = cm._assign_final_ids([self._ev("c1#0", "Z-longer-valid-span", 0), self._ev("c1#1", "B-quote", 100)])["ids"] assert run1 == run2, "★인용만 바꿨는데 번호가 바뀌었다" def test_reordering_the_model_rows_does_not_swap_the_numbers(self): run1 = cm._assign_final_ids([self._ev("c1#0", "첫째", 0), self._ev("c1#1", "둘째", 100)])["ids"] # ★같은 근거에 **다른 local_id** 가 붙은 판 — 모델이 순서만 바꾼 것 run2 = cm._assign_final_ids([self._ev("c1#1", "첫째", 0), self._ev("c1#0", "둘째", 100)])["ids"] assert run1["c1#0"] == run2["c1#1"], "★근거가 같은데 번호가 바뀌었다" assert run1["c1#1"] == run2["c1#0"] def test_identical_evidence_is_contested_not_numbered(self): """★근거가 **똑같으면** 구조적으로 못 가른다 — 안정 ID 를 주장 안 한다.""" got = cm._assign_final_ids([self._ev("c1#0", "같", 50), self._ev("c1#1", "같", 50)]) assert got["ids"] == {} assert got["contested"][0]["local_ids"] == ["c1#0", "c1#1"] def test_the_key_does_not_contain_the_local_id(self): a = cm.evidence_key(self._ev("c1#0", "가", 20)) b = cm.evidence_key(self._ev("c9#7", "나", 20)) assert a == b, "★열쇠에 모델이 정한 순서가 들어 있다" def test_the_order_of_arrival_does_not_change_the_ids(self): rows = [self._ev("c9#1", "가", 90), self._ev("c1#0", "나", 10), self._ev("c5#2", "다", 50, owner="location")] assert cm._assign_final_ids(rows) == \ cm._assign_final_ids(list(reversed(rows))) def test_it_numbers_by_owner_from_one(self): rows = [self._ev("c9#1", "가", 90), self._ev("c1#0", "나", 10), self._ev("c5#2", "다", 50, owner="location")] assert sorted(cm._assign_final_ids(rows)["ids"].values()) == \ ["L01", "P01", "P02"] def test_it_validates_its_own_input(self): """★★`assign_final_ids` 는 **직접 불릴 수 있다** — `reduce_rows` 에만 검증을 두면 겹친 id 가 조용히 덮어써 행 하나가 사라진다 (Codex).""" r = self._ev("c1#0", "가", 30) with pytest.raises(AssertionError, match="겹친다"): cm._assign_final_ids([r, dict(r)]) with pytest.raises(AssertionError, match="local_id"): cm._assign_final_ids([{"local_id": "", "owner_type": "prop", "occurrences": []}]) def test_all_five_owners_get_a_prefix(self): rows = [_row("c1", i, o, anchors=(f"S{i}",)) for i, o in enumerate(cm.OWNERS)] assert len(set(cm._assign_final_ids(rows)["ids"].values())) == 5 def test_the_prefix_table_and_the_a0_owners_are_one_set(self): """★A0 가 갈래를 늘리면 `assert_rows` 는 받는데 발급이 **터진다**.""" from app.modules.pipeline import grounding_entity_contract as ec ec.assert_prefix_parity() assert set(ec.OWNER_PREFIX) == set(cm.OWNERS) class TestItIsNotWiredYet: """★구조가 승인되기 전에는 **주행에 안 붙는다**. ★한 판 앞서 이 시험은 「`app/` 안에서 이 모듈을 import 하는 줄이 0」이었다. 그런데 C(c) 는 **여러 모듈**로 이뤄지고(payload 조립·좌표 매김·줄이기), 그것들끼리는 서로 불러야 한다. 그 조건 그대로면 **C(c) 를 짓는 것 자체**가 막힌다 — 잠근 자리가 틀린 것이다. 막아야 하는 것은 「모듈이 서로 부르는 것」이 아니라 **「주행이 그것을 타는 것」**이다. 그래서 조건을 **스텝 경계**로 옮긴다. """ #: C(c) 를 이루는 모듈들. ★여기 있는 것끼리는 불러도 된다. #: #: ★★producer 스텝(`grounding_chunk_step`)이 들어왔다 (2026-08-31, Codex #: 승인 범위). 그러면서 이 시험이 지키는 뜻이 **바뀌었다** — #: #: 앞 「C(c) 코드를 부르는 자리가 **없다**」 #: 지금 「C(c) 에 **닿는 길**이 없다」 #: #: 스텝 파일은 있지만 manifest·`STEP_CLASSES`·다른 소비자 어디에서도 #: 그 스텝에 닿지 못한다. **그 조건은 `test_chunk_producer_step.py` 가 #: 따로 잰다** — 여기만 보고 「안 켜졌다」로 읽으면 안 된다. CC_MODULES = ("grounding_chunk_merge", "grounding_chunk", "grounding_chunk_adapter", "grounding_chunk_plan", "grounding_chunk_step") def test_the_producer_step_is_the_only_step_file_in_the_family(self): """★C(c) 를 부르는 **스텝**은 하나뿐이다 — 늘면 여기서 보인다.""" from pathlib import Path as _P root = _P(__file__).resolve().parents[2] / "app" / "core" / "steps" got = sorted(f.stem for f in root.rglob("*.py") if f.stem in self.CC_MODULES) assert got == ["grounding_chunk_step"], f"★스텝이 늘었다: {got}" @staticmethod def _imports(path) -> set: """★**AST 로 실제 import 만** 본다. 앞 판은 글자로 찾아서 **주석 한 줄**(「`reduce_episode(...)` 산출」)을 배선으로 읽었다. 그러면 「왜 이렇게 고쳤는지」를 지워야 시험이 통과한다 — 시험이 설명을 못 쓰게 만드는 것은 잠근 자리가 틀린 것이다. """ import ast out = set() for n in ast.walk(ast.parse(path.read_text(encoding="utf-8"))): if isinstance(n, ast.Import): out |= {a.name for a in n.names} elif isinstance(n, ast.ImportFrom): out.add(n.module or "") out |= {f"{n.module or chr(39)*0}.{a.name}" for a in n.names} return out def _offenders(self, root: str) -> list: from pathlib import Path as _P hits = [] for f in sorted(_P(root).rglob("*.py")): if f.stem in self.CC_MODULES: continue # ★C(c) 끼리는 불러도 된다 got = self._imports(f) for mod in self.CC_MODULES: if any(part.split(".")[-1] == mod for part in got): hits.append(f"{f}: {mod}") return hits #: ★2026-09-01 D 활성화 — 스텝 등록부 **한 곳**만 C(c) 를 부른다. #: 그 자리가 곧 활성화 경계다. 늘면 경계가 두 벌이 된다. ALLOWED_WIRING = ("app/core/steps/__init__.py: grounding_chunk_step",) def test_only_the_registry_reaches_it(self): """★★**뒤집은 시험** — 앞에는 「스텝이 안 탄다」였다. 활성화했으므로 이제 묻는 것은 **「어디서 타나」**다. 등록부 하나여야 한다 — 여러 자리에서 부르면 끄고 켜는 자리가 흩어진다. """ hits = self._offenders("app/core") assert tuple(hits) == self.ALLOWED_WIRING, f"★배선 자리가 다르다: {hits}" def test_outside_that_one_place_nothing_imports_it(self): hits = self._offenders("app") assert tuple(hits) == self.ALLOWED_WIRING, f"★배선 자리가 늘었다: {hits}" def test_the_guard_would_catch_a_real_import(self, tmp_path): """★★positive control — 글자로 보던 판은 **주석도** 잡았다. AST 로 바꾼 뒤에도 **진짜 import 는 잡는지** 봐야 한다.""" d = tmp_path / "app" d.mkdir() (d / "innocent.py").write_text( "# grounding_chunk_merge 를 설명만 한다\n" "def f():\n \"\"\"reduce_episode 산출\"\"\"\n", encoding="utf-8") assert self._offenders(str(d)) == [], "★주석을 배선으로 읽었다" (d / "guilty.py").write_text( "from app.modules.pipeline import grounding_chunk_merge\n", encoding="utf-8") assert self._offenders(str(d)), "★진짜 import 를 놓쳤다" def test_the_step_registry_knows_exactly_one_chunk_step(self): """★뒤집었다 — 등록은 됐되 **하나뿐**이어야 한다.""" from app.core.steps import STEP_CLASSES assert [k for k in STEP_CLASSES if "chunk" in k] == ["grounding_chunk"] class TestTheContractIsActuallyConsumed: """★★★**상수를 만들어 놓고 아무도 안 썼다** (Codex). 「SOT 한 벌」이라고 보고했는데 `EntityFilterStep` 은 여전히 `max_scenes=2` 를 직접 넘기고 있었다. 소스 문자열 검사로는 그것을 못 잡는다 — **실제로 넘어간 값**을 잡는다. """ 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 _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") def test_the_filter_receives_the_shared_threshold(self, env, monkeypatch): """★끝점 — `filter_low_frequency_entities` 에 **실제로 넘어간** `max_scenes` 가 공용 상수와 같은가.""" from app.core.steps import STEP_CLASSES from app.modules.pipeline import entity_filter as ef from app.modules.pipeline import grounding_entity_contract as ec seen = {} def _fake(**kw): seen.update(kw) return {"filtered": {}} monkeypatch.setattr(ef, "filter_low_frequency_entities", _fake) self._cp(env, "entity_merge", {"characters": [], "locations": [], "props": []}) self._cp(env, "entity_relation", {"relations": []}) self._cp(env, "scene_save", {"segments": [{"text": "가", "length": 1, "scene_index": 1}]}) self._cp(env, "text_cleanup", {"cleaned_text": "가"}) # ★★**값을 바꿔 보고 끝점이 따라오는지** 본다. 같은 값끼리 견주면 # `max_scenes=2` 로 되돌려도 시험이 초록이다 — 실제로 그랬다. monkeypatch.setattr(ec, "ENTITY_MIN_OCCURRENCES", 7) runner = STEP_CLASSES["entity_filter"]( step_id="entity_filter", project_id="p", episode_id="e", db=None, project_config={}) runner._execute() assert seen.get("max_scenes") == 7, ( f"★필터가 받은 값 {seen.get('max_scenes')!r} — 계약을 안 쓰고 " "제 숫자를 쓴다(두 벌)") def test_the_short_id_prefixes_come_from_the_contract(self, project_db, monkeypatch): """★★끝점 — 계약의 접두를 바꾸면 **실제로 붙는 `short_id`** 가 따라오나. `_prefix_for` 만 견주면 호출부가 `"X"` 를 박아도 초록이다 — 그랬다. """ from app.core.steps.entity_steps import _assign_short_ids from app.modules.pipeline import grounding_entity_contract as ec monkeypatch.setitem(ec.OWNER_PREFIX, "character", "ZZ") class _Runner: """`_assign_short_ids` 가 보는 것은 `db` 와 `project_id` 뿐이다.""" def __init__(self, db): self.db = db self.project_id = "p" rows = [{"name": "가"}, {"name": "나"}] # ★2026-09-04 — 번호는 **프로젝트 장부**에서 온다. 접두가 계약에서 # 오는지는 그대로 재고, 발급 자리만 바뀌었다. _assign_short_ids(_Runner(project_db), rows, "character") assert [r["short_id"] for r in rows] == ["ZZ01", "ZZ02"] def test_the_list_step_uses_that_prefix(self, env, project_db, monkeypatch): """★그리고 **스텝이 실제로** 그 접두를 쓰는가 — 호출부에 박혀 있으면 계약을 바꿔도 안 따라온다.""" from app.core.steps import STEP_CLASSES from app.modules.pipeline import entity_lister as el from app.modules.pipeline import grounding_entity_contract as ec monkeypatch.setitem(ec.OWNER_PREFIX, "character", "ZZ") self._cp(env, "visual_world_rules", {"era": "E", "region": "R"}) self._cp(env, "shot_validator", {"scenes": [ {"scene_index": 1, "scene_heading": "h", "shots": [{"shot_index": 1, "description": "d"}]}]}) monkeypatch.setattr(el, "call_structured", lambda **kw: { "characters": [{"name": "가", "shot_count": 1}]}) monkeypatch.setattr(el, "load_prompt", lambda *a, **k: "p") monkeypatch.setattr(el, "load_schema", lambda *a, **k: { "type": "object", "properties": {"characters": {"type": "array", "items": { "type": "object", "properties": {"name": {"type": "string"}, "shot_count": {"type": "integer"}}, "required": ["name", "shot_count"], "additionalProperties": False}}}, "required": ["characters"], "additionalProperties": False}) runner = STEP_CLASSES["entity_all_character"]( step_id="entity_all_character", project_id="p", episode_id="e", db=project_db, project_config={}) out = runner._execute() assert out["data"]["characters"][0]["short_id"] == "ZZ01", \ "★스텝이 접두를 손으로 박고 있다" def test_the_overlay_prefix_table_is_derived_not_retyped(self): """★복수 alias 만 코드로 만들고 다섯 갈래는 **계약에서** 온다.""" from app.modules.pipeline import grounding_entity_contract as ec from app.modules.pipeline.grounding_overlay import _ENTITY_PREFIX for owner, pre in ec.OWNER_PREFIX.items(): assert _ENTITY_PREFIX[owner] == pre assert _ENTITY_PREFIX[f"{owner}s"] == pre def test_a_facet_is_still_not_materialized(self): """★접두표에 facet 이 들어왔다 — 그 문이 약해지지 않았는지 본다.""" from app.modules.pipeline.grounding_overlay import ( materialize_missing_entities) out = materialize_missing_entities( [{"research_subject_id": "r1", "owner_type": "location_part", "surface_form": "가", "source_anchor": "a"}], [{"research_subject_id": "r1", "route": "research", "generation_difficulty": "hard"}], {"location_parts": [], "props": []}) assert out == {}, "★facet 이 base 갈래로 올라왔다"