"""다섯 갈래가 계약 사슬을 지나는가 — **결정적 fixture 끝점**. ★유료 0. ★★★**이것은 replay 가 아니다** (Codex 정정 2026-09-01). 아래 `_rows_and_ relations()` 가 `EXPECTED_TARGETS`/`AXIS_BASIS` 로 모델 행과 샷 ID 를 **새로 만든다**. 즉 저장된 raw 응답을 푸는 것이 아니라, **양성 대조 fixture** 로 reducer→binder→adapter→outcome 을 태우는 것이다. 진짜 replay 라면 저장된 raw chunk 응답을 sender 없이 풀어야 한다 — 호환 raw 가 생기기 전에는 이것을 「producer 실증」이라고 부르지 않는다. Codex ⓑ (2026-09-01) — 「2씬 canary 의 outlook 2개는 검증 근거로 쓰지 마십시오. 원고의 구조 의무가 `location`+`location_part` 뿐이고 explicit outlook 묘사가 없는데 legacy 산출에 outlook 2 가 나온 **숫자만으로** outlook research 를 잴 수 없습니다. 이미 있는 `period_episode` fixture 가 명시적 근거·세계사실·음성대조· 기대 anchor 를 갖습니다. 그것으로 먼저 **provider 0 replay/free 끝점**을.」 여기서 잇는 사슬 — 모델 행 → reduce_episode(등록·part_of 장부) → grounding_facet_binding.bind(구조적 결속) → grounding_chunk_adapter.to_entity_rows(하류 CP 모양) → reference_acquisition.acquisition_outcome / downstream_blocked ★이름·부분문자열로 아무것도 안 가른다. 결속은 **ID 와 갈래**로만 한다. ★사람 대기(HITL)를 만들지 않는다 — 못 구하면 `reference_unavailable` 로 적고 참조 없이 내려간다. """ from __future__ import annotations import pytest from app.modules.pipeline import grounding_chunk_adapter as ad from app.modules.pipeline import grounding_chunk_merge as cm from app.modules.pipeline import grounding_facet_binding as fb from app.modules.pipeline import reference_acquisition as ra from tests.grounding.fixtures import period_episode as ep SEGS = ep.segment_texts() #: facet → 부모로 쓸 표적 key. ★표에서 온다 — 이름을 안 본다. PARENT_KEY = {"owner_outlook": "owner_character", "owner_location_part": "owner_location", "plain_location_part": "owner_location"} def _rows_and_relations(): """fixture 표적을 **모델이 낸 행** 모양으로 만든다. ★두 축(`hard_to_generate`/`viewers_would_notice`)은 `AXIS_BASIS` 에서만 온다 — 여기서 지어내지 않는다. 음성 대조는 근거가 없으므로 False 다. """ rows, rel, key_of = [], [], {} for i, t in enumerate(ep.EXPECTED_TARGETS): scene, word, nth = t["at"][0] lid = f"c0#{i}" key_of[t["key"]] = lid axis = ep.AXIS_BASIS.get(t["key"]) or {} rows.append({ "local_id": lid, "owner_type": t["owner"], "surface_form": word, "occurrences": [{"source_span": ep.span_of(scene, word, nth), "source_quote": word}], "shot_binding_status": "bound_complete", # ★두 샷에 나와야 등록 문턱을 넘는다 — 갈래와 무관한 공통 규칙 "shot_appearance_ids": [f"s{scene}#1", f"s{scene}#2"], "hard_to_generate": bool(axis.get("hard")), "viewers_would_notice": bool(axis.get("notice")), }) for child, parent in PARENT_KEY.items(): rel.append({"remove_local_id": key_of[child], "keep_local_id": key_of[parent], "relation": cm.REL_PART_OF}) return rows, rel, key_of @pytest.fixture(scope="module") def chain(): rows, rel, key_of = _rows_and_relations() red = cm.reduce_episode(rows, rel, segments=SEGS) bound = fb.bind(red["rows"], red["part_of"], red["registered"]) cp = ad.to_entity_rows(red) return {"reduced": red, "bound": bound, "cp": cp, "key_of": key_of} class TestTheFixtureItselfIsHonest: def test_it_still_holds_together(self): ep.assert_planted() def test_every_owner_appears_at_least_once_as_a_positive(self): pos = [t for t in ep.EXPECTED_TARGETS if (ep.AXIS_BASIS.get(t["key"]) or {}).get("hard")] assert {t["owner"] for t in pos} == { "character", "location", "location_part", "outlook", "prop"} def test_the_negatives_say_why_they_are_not_grounded(self): """★★음성 대조도 **근거가 있다** — 다만 두 축이 False 다. 내가 처음 쓴 시험은 「음성이면 `AXIS_BASIS` 에 아예 없어야 한다」였는데 fixture 는 **왜 아닌지**를 적어 둔다. 그쪽이 낫다 — 「안 적었다」와 「아니라고 적었다」가 갈린다. """ neg = [t for t in ep.EXPECTED_TARGETS if not (ep.AXIS_BASIS.get(t["key"]) or {}).get("hard")] assert neg, "★음성 대조가 없다" for t in neg: axis = ep.AXIS_BASIS[t["key"]] assert axis["hard"] is False and axis["notice"] is False assert axis["basis"].strip(), "★왜 아닌지를 안 적었다" class TestAllFiveReachRegistration: def test_every_owner_gets_a_final_id(self, chain): red, key_of = chain["reduced"], chain["key_of"] by_owner = {} for r in red["rows"]: rec = red["registered"].get(r["local_id"]) or {} by_owner.setdefault(r["owner_type"], []).append(rec) for owner in ("character", "location", "location_part", "outlook", "prop"): recs = by_owner.get(owner) or [] assert recs, f"★{owner} 행이 아예 없다" assert any(x.get("final_id") for x in recs), \ f"★{owner} 가 하나도 등록 안 됐다" def test_the_part_never_borrows_the_base_prefix(self, chain): """★★base 갈래 **우회 등록 0건** — 이것이 계약의 핵심이다.""" from app.modules.pipeline.grounding_entity_contract import ( owner_of_final_id) red = chain["reduced"] for r in red["rows"]: fid = (red["registered"].get(r["local_id"]) or {}).get("final_id") if not fid: continue assert owner_of_final_id(fid) == r["owner_type"], ( f"★{fid} 가 {r['owner_type']} 인데 다른 갈래로 읽힌다") class TestBothFacetsBindStructurally: def test_the_outlook_binds_to_its_character(self, chain): got = {b["local_id"]: b for b in chain["bound"]["bindings"]} b = got[chain["key_of"]["owner_outlook"]] assert b["parent_owner_type"] == "character" assert b["parent_local_id"] == chain["key_of"]["owner_character"] def test_the_location_part_binds_to_its_location(self, chain): got = {b["local_id"]: b for b in chain["bound"]["bindings"]} b = got[chain["key_of"]["owner_location_part"]] assert b["parent_owner_type"] == "location" assert b["parent_local_id"] == chain["key_of"]["owner_location"] def test_nothing_is_left_as_debt(self, chain): assert chain["bound"]["debt"] == [], chain["bound"]["debt"] def test_binding_never_reads_a_surface_form(self): """★★결속은 **ID 와 갈래**로만 — 글자를 안 본다 (AST 로 확인).""" import ast import inspect tree = ast.parse(inspect.getsource(fb)) for n in ast.walk(tree): if isinstance(n, ast.Attribute) and n.attr in ( "startswith", "endswith", "lower", "upper"): raise AssertionError("★결속이 글자를 본다") class TestTheDownstreamCheckpointKeepsThemApart: def test_the_part_gets_its_own_key(self, chain): rows = chain["cp"]["rows"] assert "location_parts" in rows and "locations" in rows lp = {r["short_id"] for r in rows["location_parts"]} lo = {r["short_id"] for r in rows["locations"]} assert lp and lo and not (lp & lo) def test_the_outlook_is_not_a_base_row(self, chain): """★`outlook` 은 `character_outlook` 이 따로 쓴다 — base 행이 아니다.""" for key, made in chain["cp"]["rows"].items(): for r in made: assert not str(r["short_id"]).startswith("O"), (key, r) skipped = {s["owner_type"] for s in chain["cp"]["skipped"]} assert "outlook" in skipped def test_nothing_vanished_without_a_reason(self, chain): made = sum(len(v) for v in chain["cp"]["rows"].values()) assert made + len(chain["cp"]["skipped"]) == len( chain["reduced"]["rows"]) class TestNobodyWaitsForAHuman: """★사람 대기를 production 에 만들지 않는다 (사용자 확정 2026-08-31).""" @pytest.mark.parametrize("status", [ra.STATUS_SELECTED, "no_match_after_retry", "retryable"]) def test_the_outcome_is_one_of_two(self, status): assert ra.acquisition_outcome(status) in ( ra.STATUS_SELECTED, ra.STATUS_UNAVAILABLE) @pytest.mark.parametrize("status", [ra.STATUS_SELECTED, "no_match_after_retry", "retryable", "무엇인지 모름"]) def test_downstream_is_never_blocked(self, status): assert ra.downstream_blocked(status) is False # ───────────────────────────────────────────────────────────────────── # 고증 의무 — ★`build_subjects` 는 **production 함수**다. 손으로 짓지 않는다. # ───────────────────────────────────────────────────────────────────── def _plan_env(**kw): """`era_research.assess_plan_cached` 의 **명시 envelope** 모양 그대로. ★내가 지어낸 모양을 쓰면 시험은 초록인데 실제 경로는 안 돈다 — 오늘 이미 두 번 그랬다. `{"status": …, "plan": …}` 이고 status 는 셋 중 하나다. """ from app.modules.pipeline import era_research as era return {"status": era.PLAN_OK, "plan": {"assess_sha": "sha", "subjects": [{"q": "무엇"}]}, **kw} @pytest.fixture(scope="module") def screened(chain): """★★**production 모양 그대로** 태운다 (Codex 2026-09-01). `GroundingScreenStep` 은 ①`build_subjects` ②그 **`candidate_ledger` 가 낸 disposition** ③`build_population(subjects, a0_candidates, dispositions)` 순서로 간다. 앞 판은 내가 `disposition_of(owner)` 로 **다시 지어** 냈는데, 그러면 같은 owner 의 「기존 행」과 「행 없는 후보」를 **구별 못 한다**. """ from app.modules.pipeline import grounding_carry as gc from app.modules.pipeline import grounding_screen as gs ents = {k: list(v) for k, v in chain["cp"]["rows"].items()} a0 = _a0_candidates(chain) built = gc.build_subjects(ents, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=a0) ledger = built.get("candidate_ledger") or {} # ★production 과 **같은 합치기** — subject 쪽 처분이 먼저, 후보 장부가 덮는다 disp = dict(built.get("subject_dispositions") or {}) disp.update({str(r.get("research_subject_id") or ""): str(r.get("disposition") or "") for r in (ledger.get("rows") or [])}) pop = gs.build_population(built["subjects"], a0, disp) got = gs.screen_subjects( pop, dispositions=disp, world_facts_block=ep.WORLD_FACTS, cache_get=lambda k: None, cache_put=lambda k, v: v, assess_fn=lambda **kw: _plan_env()) return {"built": built, "screen": got, "dispositions": disp, "population": pop} def _a0_candidates(chain): """★★**실제 A0 산출 모양**으로 만든다 (Codex 2026-09-01). 실측: 저장된 A0 후보의 칸은 `owner_type · planned_occurrences · source_anchor · source_quote · surface_form · why_candidate` **뿐**이다 — `short_id` 도, 어떤 id 칸도 **없다**. A0 는 엔티티 ID 가 생기기 **전에** 원문에서 후보를 낸다. ★앞 판은 내가 `short_id` 를 지어 넣어 성공 경로를 만들었다. 그러면 시험이 **production 이 안 보내는 모양**을 재는 것이고, 이름으로 붙은 것과 신원으로 붙은 것을 구별하지 못한다. 그래서 여기서는 **base 갈래(C/L/P)** 만 후보로 만든다 — LP 는 A0 만으로는 붙지 않는 것이 계약이다. """ from app.modules.pipeline.grounding_carry import STRUCTURAL_ONLY_OWNERS from app.modules.pipeline.grounding_entity_sync_ext import CHUNK_OWNER_KEYS from app.modules.pipeline.grounding_subject import build_subject back = {v: k for k, v in CHUNK_OWNER_KEYS.items()} out = [] for key, rows in chain["cp"]["rows"].items(): owner = back.get(key, key) if owner in STRUCTURAL_ONLY_OWNERS: continue for r in rows: name = r.get("name") or "" # ★id 는 **production 이 발급하는 방식 그대로** — 손으로 안 짓는다 out.append({ "owner_type": owner, "surface_form": name, "source_anchor": "본문", "planned_occurrences": 2, "why_candidate": "그 시대 특유의 형태다", **build_subject(project_id="p", episode_id="e", source_anchor="본문", surface_form=name, owner_type=owner, provenance={"source_step": "grounding_a0"}), "source_quote": name}) return out class TestTheObligationLedgerCoversWhatWeRegistered: def test_every_registered_row_has_a_subject(self, chain, screened): """★조용히 빠진 것이 없다 — 등록된 것은 판별 장부에 **줄이 있다**. ★★한동안 `xfail(strict)` 였다: 등록 6 인데 subject 4 였고, `location_part` 가 DB 까지 등록되고도 **고증 대상이 되지 못했다**. `ENTITY_KEY_TO_OWNER` 에 `location_parts` 를 넣어 **이미 등록된 `LP##` 행을 스캔·결속**하게 하니 닫혔다(Codex 설계: 일반 승격이 아니다). """ made = sum(len(v) for v in chain["cp"]["rows"].values()) subs = len(screened["built"]["subjects"]) unbound = len(screened["built"].get("unbound") or []) assert subs + unbound >= made > 0, (subs, unbound, made) def test_the_ledger_has_one_row_per_subject(self, screened): got = screened["screen"] assert len(got["rows"]) == len(screened["built"]["subjects"]) def test_it_did_not_buy_anything(self, screened): """★판별을 갈아 끼웠으니 **한 푼도 안 썼다** — 그래도 셈은 돈다.""" got = screened["screen"] assert got["assess_reused"] == 0 assert got["assess_bought"] == len(got["rows"]) def test_a_registered_location_part_binds_as_an_entity_row(self, screened): """★★§2-6.5a 뒤 — LP 는 **제 갈래로 등록**되므로 base 행으로 붙는다. `facet_obligations` 는 「사야 하는데 붙을 데가 없는 것」이다. 그 목록에 `location_part` 가 남아 있으면 「다섯 갈래 완료」는 거짓이다. """ from app.modules.pipeline import grounding_screen as gs rows = {r["research_subject_id"]: r for r in screened["screen"]["rows"]} lp = [r for r in rows.values() if r["owner_type"] == "location_part"] assert lp, "★LP 줄이 아예 없다" for r in lp: # ★이미 등록된 행이라 **entity_row** 로 붙는다 — 승격이 아니다 assert r["binding"] == gs.BIND_ENTITY_ROW, r assert r["is_facet"] is True, "★구조 축이 사라졌다" left = {r["owner_type"] for r in gs.facet_obligations(screened["screen"])} assert "location_part" not in left, ( "★등록된 LP 가 아직 producer 대기로 남아 있다") def test_the_outlook_is_still_the_open_one(self, screened): """★★남은 것을 **그대로 적는다** — 「완료」로 접지 않는다. `outlook` 은 `character_outlook` 이 SOT 라 base 행으로 안 간다. 그 갈래의 참조 획득은 §2-6.5 의 **다음 단계**다. """ from app.modules.pipeline import grounding_carry as gc assert gc.disposition_of("outlook") == gc.DISP_DEFERRED class TestTheFacetContractIsThreeAxesNotOne: """★★★Codex 설계 (2026-09-01) — `FACET_OWNERS` 를 **두 뜻**으로 쓰지 않는다. ①구조 facet `FACET_PARENT` — LP→location · outlook→character ②스캔할 갈래 `ENTITY_SYNC_OWNER_TYPES` — 이미 있는 행을 결속 ③일반 승격 가능 `GENERIC_PROMOTION_OWNERS` — 부모 없이 서도 되는 것 LP 는 ①②는 맞지만 ③은 **아니다** — 부모 없이 만들면 고아가 된다. """ def test_the_three_axes_say_different_things(self): from app.modules.pipeline import grounding_entity_contract as ec from app.modules.pipeline import grounding_facet_binding as fb from app.modules.pipeline.grounding_carry import SCANNED_OWNERS assert fb.parent_of("location_part") == "location" # ① assert "location_part" in SCANNED_OWNERS # ② assert "location_part" not in ec.GENERIC_PROMOTION_OWNERS # ③ # outlook 은 ① 뿐이다 — 스캔도 일반 승격도 아니다 assert fb.parent_of("outlook") == "character" assert "outlook" not in SCANNED_OWNERS assert "outlook" not in ec.GENERIC_PROMOTION_OWNERS def test_a_parentless_candidate_creates_nothing(self): """★acceptance ② — 행도 부모도 없는 LP 후보는 **새 ID 0**.""" from app.modules.pipeline.grounding_overlay import ( materialize_missing_entities) got = materialize_missing_entities( [{"research_subject_id": "r1", "owner_type": "location_part", "surface_form": "간판"}], [{"research_subject_id": "r1", "route": "research", "generation_difficulty": "hard"}], {"location_parts": []}) assert got == {}, got def test_a_missing_key_invents_no_phantom_row(self): """★acceptance ④ — `location_parts` 키가 없으면 **phantom 0**.""" from app.modules.pipeline.grounding_overlay import ( materialize_missing_entities) got = materialize_missing_entities( [{"research_subject_id": "r1", "owner_type": "location_part", "surface_form": "간판"}], [{"research_subject_id": "r1", "route": "research", "generation_difficulty": "hard"}], {"characters": [], "locations": [], "props": []}) assert "location_parts" not in got def test_the_base_kinds_still_get_promoted(self): """★acceptance ③ — C/L/P 일반 승격 **비회귀**.""" from app.modules.pipeline.grounding_overlay import ( materialize_missing_entities) got = materialize_missing_entities( [{"research_subject_id": "r1", "owner_type": "prop", "surface_form": "됫박"}], [{"research_subject_id": "r1", "route": "research", "generation_difficulty": "hard"}], {"props": []}) assert [r["short_id"] for r in got.get("props", [])] == ["P01"] def test_an_orphan_stays_recorded_with_a_reason(self, screened): """★acceptance ⑥ — 등록된 LP 는 빠지되 **남은 것은 사유와 함께** 남는다.""" from app.modules.pipeline import grounding_screen as gs left = gs.facet_obligations(screened["screen"]) for r in left: assert r["owner_type"] in ("outlook",), r assert r["binding"] == gs.BIND_DEFERRED_PRODUCER def test_the_part_of_ledger_survives_the_scan(self, chain): """★acceptance ① — 스캔을 열어도 `part_of` 장부는 그대로다.""" assert chain["reduced"]["part_of"], "★part_of 가 사라졌다" pairs = {(r["part"], r["whole"]) for r in chain["reduced"]["part_of"]} assert pairs, pairs class TestALocationPartNeverBindsByName: """★★★Codex BLOCK (2026-09-01) — LP 는 **이름으로 결속하지 않는다.** 같은 owner 안에서도 「간판」과 「회전 간판」은 서로 다른 부분일 수 있다. 그리고 실제 A0 후보에는 **id 칸이 아예 없다**(실측: `owner_type · planned_occurrences · source_anchor · source_quote · surface_form · why_candidate` 뿐). 그러니 「`short_id` 가 있으면 우선」은 실데이터에서 **작동하지 않는 길**이고, 임의로 채운 값은 같은 실물을 증명하지 못한다. """ ENTS = {"props": [{"name": "됫박", "short_id": "P01"}], "location_parts": [{"name": "회전 간판", "short_id": "LP01"}]} def _run(self, cand): from app.modules.pipeline import grounding_carry as gc got = gc.build_subjects(self.ENTS, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=[cand]) rows = (got.get("candidate_ledger") or {}).get("rows") or [] return got, {r["research_subject_id"]: r["disposition"] for r in rows} def _cand(self, owner, surface, **over): from app.modules.pipeline.grounding_subject import build_subject return {"owner_type": owner, "surface_form": surface, "source_anchor": "본문", "planned_occurrences": 2, "why_candidate": "그 시대 특유", **build_subject(project_id="p", episode_id="e", source_anchor="본문", surface_form=surface, owner_type=owner, provenance={"source_step": "grounding_a0"}), "source_quote": surface, **over} def test_the_real_a0_shape_has_no_id_field(self): """★근거 — 실제 저장된 A0 산출을 본다.""" import json import pathlib fx = pathlib.Path("tests/fixtures/grounding/a0_recovered_97375a4b.json") if not fx.exists(): pytest.skip("실제 A0 산출이 없다") got = json.loads(fx.read_text(encoding="utf-8"))["a0_candidates"][0] assert "short_id" not in got assert not [k for k in got if k.endswith("_id")], sorted(got) @pytest.mark.parametrize("surface", ["회전 간판", "간판", "이발소 회전 간판"]) def test_an_exact_or_partial_name_never_binds(self, surface): from app.modules.pipeline import grounding_carry as gc _got, d = self._run(self._cand("location_part", surface)) assert set(d.values()) == {gc.DISP_DEFERRED}, d def test_a_bare_short_id_is_not_enough(self): """★후보가 스스로 적은 `short_id` 는 **링크가 아니다**.""" from app.modules.pipeline import grounding_carry as gc _got, d = self._run(self._cand("location_part", "아무거나", short_id="LP01")) assert set(d.values()) == {gc.DISP_DEFERRED}, d def _link(self, **kw): from app.modules.pipeline.grounding_chunk_adapter import ( ADAPTER_CONTRACT_VERSION, ADAPTER_ISSUER) return {"issuer": ADAPTER_ISSUER, "contract_version": ADAPTER_CONTRACT_VERSION, **kw} def _with_link(self, link, ents=None): from app.modules.pipeline import grounding_carry as gc ents = ents or {"location_parts": [ {"name": "회전 간판", "short_id": "LP01", "local_id": "c0#1"}]} cand = self._cand("location_part", "다른 이름", **{gc.LEDGER_LINK: link}) return gc.build_subjects(ents, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=[cand]) def test_a_producer_issued_link_does_bind(self): """★음성 대조의 반대쪽 — **검증된 링크**면 붙는다.""" for link in (self._link(final_id="LP01"), self._link(local_id="c0#1")): assert self._with_link(link)["carried"] == 1, link def test_an_unknown_issuer_is_refused(self): """★★발급자 이름이 **비어 있지 않기만 하면** 믿던 것 (Codex · 09-01).""" got = self._with_link({**self._link(final_id="LP01"), "issuer": "아무 문자열"}) assert got["carried"] == 0 def test_a_missing_or_wrong_contract_is_refused(self): for link in (self._link(final_id="LP01", contract_version="9.999"), {"issuer": self._link()["issuer"], "final_id": "LP01"}): assert self._with_link(link)["carried"] == 0, link def test_two_coordinates_that_disagree_are_refused(self): """★★★`final_id` 와 `local_id` 가 **서로 다른 것**을 가리키면 모순이다.""" got = self._with_link(self._link(final_id="LP99", local_id="c0#1")) assert got["carried"] == 0 def test_two_storage_places_that_disagree_are_refused(self): """★★top-level 과 provenance 에 **다른 링크**가 있으면 거절한다.""" from app.modules.pipeline import grounding_carry as gc cand = self._cand("location_part", "다른 이름", **{gc.LEDGER_LINK: self._link(final_id="LP01")}, provenance={gc.LEDGER_LINK: self._link(final_id="LP99")}) got = gc.build_subjects( {"location_parts": [{"name": "회전 간판", "short_id": "LP01", "local_id": "c0#1"}]}, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=[cand]) assert got["carried"] == 0 def test_the_link_it_trusted_is_kept_verbatim(self): """★★★**무엇을 믿고 붙였나**가 남아야 한다 (Codex 조건 ④). 앞 판은 처분과 `short_id` 만 남겨서, 어떤 링크를 믿었는지 사라졌다. """ from app.modules.pipeline import grounding_carry as gc link = self._link(final_id="LP01") got = self._with_link(link) kept = [(s.get("provenance") or {}).get(gc.LEDGER_LINK) for s in got["subjects"]] assert link in kept, kept led = [r.get(gc.LEDGER_LINK) for r in got["candidate_ledger"]["rows"]] assert link in led, led def test_the_two_kinds_never_cross(self): """★★`prop` 과 `location_part` 가 **양방향으로** 안 넘어간다.""" from app.modules.pipeline import grounding_carry as gc _got, d = self._run(self._cand("prop", "회전 간판")) assert set(d.values()) == {gc.DISP_PROMOTED}, d _got, d = self._run(self._cand("location_part", "됫박")) assert set(d.values()) == {gc.DISP_DEFERRED}, d def test_a_plain_prop_still_carries(self): """★비회귀 — base 갈래의 기존 이름 carry 는 그대로다.""" from app.modules.pipeline import grounding_carry as gc _got, d = self._run(self._cand("prop", "됫박")) assert set(d.values()) == {gc.DISP_CARRIED}, d def test_a_registered_part_with_no_candidate_is_an_entity_row(self): """★★Codex 끝점 ① — 등록 LP + A0=[] → `entity_row` · 새 ID 0.""" from app.modules.pipeline import grounding_carry as gc from app.modules.pipeline import grounding_screen as gs got = gc.build_subjects(self.ENTS, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=[]) disp = dict(got.get("subject_dispositions") or {}) lp = [s for s in got["subjects"] if s["owner_type"] == "location_part"] assert len(lp) == 1, lp rsid = lp[0]["research_subject_id"] assert disp[rsid] == gc.DISP_ENTITY_ONLY assert gs.binding_of(rsid, disp) == gs.BIND_ENTITY_ROW # ★새로 만든 것이 없다 assert ((got.get("candidate_ledger") or {}).get("rows") or []) == [] def test_both_survive_when_they_cannot_be_linked(self): """★★Codex 끝점 ② — 억지로 합치지 않고 **둘 다 남는다**.""" from app.modules.pipeline import grounding_carry as gc got, d = self._run(self._cand("location_part", "회전 간판")) # 엔티티 쪽은 제 subject 로 남는다 assert [s for s in got["subjects"] if s["owner_type"] == "location_part"] # 후보 쪽은 따로 `deferred` 로 남는다 assert set(d.values()) == {gc.DISP_DEFERRED} class TestAnOutlookCandidateIsActuallyBurned: """★NON-BLOCK 정정 (Codex) — 앞 시험은 `left` 가 **비어도 통과**했다. CP 에 outlook 이 없어서 그 갈래를 **태우지도 않았다**.""" def test_exactly_one_outlook_stays_deferred_with_a_reason(self): from app.modules.pipeline import grounding_carry as gc from app.modules.pipeline import grounding_screen as gs from app.modules.pipeline.grounding_subject import build_subject cand = {"owner_type": "outlook", "surface_form": "긴 겉옷", "source_anchor": "본문", "planned_occurrences": 2, "why_candidate": "그 시대 특유", **build_subject(project_id="p", episode_id="e", source_anchor="본문", surface_form="긴 겉옷", owner_type="outlook", provenance={"source_step": "grounding_a0"}), "source_quote": "긴 겉옷"} built = gc.build_subjects({"characters": []}, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=[cand]) disp = dict(built.get("subject_dispositions") or {}) disp.update({r["research_subject_id"]: r["disposition"] for r in (built["candidate_ledger"]["rows"] or [])}) pop = gs.build_population(built["subjects"], [cand], disp) got = gs.screen_subjects( pop, dispositions=disp, world_facts_block=ep.WORLD_FACTS, cache_get=lambda k: None, cache_put=lambda k, v: v, assess_fn=lambda **kw: _plan_env()) left = gs.facet_obligations(got) assert len(left) == 1, left assert left[0]["owner_type"] == "outlook" assert left[0]["binding"] == gs.BIND_DEFERRED_PRODUCER class TestTheFrontEndKeepsTheEvidenceForLater: """★★★outlook 은 **19.2 뒤에야** 실물이 생긴다 (order: screen 13.68 · `outlook_phase3` 19.2). 그래서 앞단은 **의무와 증거만** 남기고 실제 참조는 중앙 획득 한 곳이 산다 — 그런데 앞 판의 줄에는 판별 결과뿐이라 **뒤에서 결속할 증거가 없었다** (실측 · Codex · 09-01). """ def _screen_one(self, owner, **over): from app.modules.pipeline import grounding_carry as gc from app.modules.pipeline import grounding_screen as gs from app.modules.pipeline.grounding_subject import build_subject cand = {"owner_type": owner, "surface_form": "긴 겉옷", "source_anchor": "본문", "planned_occurrences": 2, "why_candidate": "그 시대 특유", **build_subject(project_id="p", episode_id="e", source_anchor="본문", surface_form="긴 겉옷", owner_type=owner, provenance={"source_step": "grounding_a0"}), "source_quote": "발목까지 오는 긴 겉옷을 앞을 여며 묶었다", **over} built = gc.build_subjects({"characters": []}, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=[cand]) disp = dict(built.get("subject_dispositions") or {}) disp.update({r["research_subject_id"]: r["disposition"] for r in built["candidate_ledger"]["rows"]}) pop = gs.build_population(built["subjects"], [cand], disp) got = gs.screen_subjects( pop, dispositions=disp, world_facts_block=ep.WORLD_FACTS, cache_get=lambda k: None, cache_put=lambda k, v: v, assess_fn=lambda **kw: _plan_env()) return got, cand def test_every_row_carries_its_source_evidence(self): got, cand = self._screen_one("outlook") for r in got["rows"]: ev = r["source_evidence"] assert ev["surface_form"] == cand["surface_form"] assert ev["source_quote"] == cand["source_quote"] assert ev["source_anchor"] == cand["source_anchor"] assert ev["payload_sha"], "★지문이 없다" def test_it_never_invents_a_span(self): """★★없는 좌표를 **만들지 않는다** — 지어낸 span 은 증거가 아니다.""" got, _c = self._screen_one("outlook") for r in got["rows"]: ev = r["source_evidence"] assert "occurrences" not in ev assert "source_spans" not in ev and "spans" not in ev def test_it_keeps_spans_verbatim_when_they_exist(self): spans = [{"segment_id": "scene-1", "start": 3, "end": 7}] got, _c = self._screen_one("outlook", occurrences=[ {"source_span": spans[0], "source_quote": "긴 겉옷"}]) kept = [r["source_evidence"].get("occurrences") for r in got["rows"]] assert any(k and k[0]["source_span"] == spans[0] for k in kept), kept def test_the_facet_debt_is_a_view_not_a_second_ledger(self): """★★`facet_debt` 는 rows 를 거른 **view** 다 — 사본이 아니다.""" from app.modules.pipeline import grounding_screen as gs got, _c = self._screen_one("outlook") left = gs.facet_obligations(got) assert left, "★outlook 을 안 태웠다" for r in left: assert r in got["rows"], "★사본을 만들었다" assert r["source_evidence"]["source_quote"] def test_a_base_owner_keeps_evidence_too(self): """★음성 대조 — facet 만이 아니라 **모든 정본 행**이 담는다.""" got, _c = self._screen_one("prop") assert all(r["source_evidence"]["surface_form"] for r in got["rows"]) class TestALinkMustMatchEveryCoordinateItClaims: """★★★Codex BLOCK (09-01) — 「좌표가 안 어긋난다」가 **실제 행 모양**에서는 성립하지 않았다. adapter 는 `short_id` 를 top-level 에, `local_id` 는 **`grounding_provenance` 안**에 둔다. `build_subjects` 가 top-level 만 읽어 `want_l=""` 이 되고, 검사는 `if lid and want_l and …` 라 **대상에 없는 좌표를 그냥 지나쳤다**. 그래서 `final_id` 하나만 맞으면 엉뚱한 `local_id` 를 단 링크도 통과했다. """ ENT = {"name": "회전 간판", "short_id": "LP01", "grounding_provenance": {"local_id": "c0#1"}} def _link(self, **kw): from app.modules.pipeline.grounding_entity_contract import ( PRODUCER_CONTRACT_VERSION, PRODUCER_ISSUER) return {"issuer": PRODUCER_ISSUER, "contract_version": PRODUCER_CONTRACT_VERSION, **kw} def _run(self, link=None, ent=None): from app.modules.pipeline import grounding_carry as gc from app.modules.pipeline.grounding_subject import build_subject cand = {"owner_type": "location_part", "surface_form": "다른 이름", "source_anchor": "본문", "source_quote": "x", **build_subject(project_id="p", episode_id="e", source_anchor="본문", surface_form="다른 이름", owner_type="location_part", provenance={"source_step": "grounding_a0"})} if link is not None: cand[gc.LEDGER_LINK] = link got = gc.build_subjects({"location_parts": [dict(ent or self.ENT)]}, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=[cand]) return got, got["candidate_ledger"]["rows"][0] def test_the_entity_local_id_is_read_from_its_provenance(self): """★adapter 는 그것을 **provenance 안**에 둔다 — top-level 만 보면 빈다.""" from app.modules.pipeline.grounding_carry import entity_local_id assert entity_local_id(self.ENT) == "c0#1" assert entity_local_id({"local_id": "x"}) == "x" assert entity_local_id({}) == "" def test_a_bogus_local_id_is_refused_even_when_final_matches(self): """★★Codex 재현 그대로.""" from app.modules.pipeline import grounding_carry as gc got, row = self._run(self._link(final_id="LP01", local_id="bogus-local-id")) assert got["carried"] == 0 assert row["link_fault"] == gc.LINK_COORD_MISMATCH slots = row[gc.LINK_RAW_SLOTS] assert slots[0]["value"]["local_id"] == "bogus-local-id" def test_the_symmetric_case_is_refused_too(self): from app.modules.pipeline import grounding_carry as gc got, row = self._run(self._link(local_id="c0#1", final_id="LP99")) assert got["carried"] == 0 assert row["link_fault"] == gc.LINK_COORD_MISMATCH def test_a_coordinate_the_target_does_not_have_is_refused(self): """★「없으면 무시」 금지 — 검사할 수 없으면 **안 믿는다**.""" from app.modules.pipeline import grounding_carry as gc got, row = self._run(self._link(final_id="LP01", local_id="c0#1"), ent={"name": "회전 간판", "short_id": "LP01"}) assert got["carried"] == 0 assert row["link_fault"] == gc.LINK_COORD_MISSING @pytest.mark.parametrize("link,reason", [ ({"issuer": "아무개", "contract_version": "1", "final_id": "LP01"}, "unknown_issuer"), (None, None), ]) def test_broken_is_told_apart_from_absent(self, link, reason): """★★★계약 drift 가 평범한 「링크 없음」처럼 보이면 안 된다.""" from app.modules.pipeline import grounding_carry as gc got, row = self._run(link if link is None else self._link(**{ k: v for k, v in link.items() if k != "contract_version"}, **{"contract_version": link["contract_version"]})) assert got["carried"] == 0 if reason is None: assert "link_fault" not in row, "★없음인데 사유가 붙었다" assert gc.LINK_RAW_SLOTS not in row else: assert row["link_fault"] in gc.LINK_BROKEN assert row[gc.LINK_RAW_SLOTS], "★원형 슬롯이 안 남았다" def test_both_coordinates_agreeing_is_the_positive_control(self): from app.modules.pipeline import grounding_carry as gc got, row = self._run(self._link(final_id="LP01", local_id="c0#1")) assert got["carried"] == 1 assert "link_fault" not in row assert row[gc.LEDGER_LINK]["final_id"] == "LP01" assert row[gc.LEDGER_LINK]["local_id"] == "c0#1" class TestTheProducerEmitsTheLinkItselfNow: """★★★받는 쪽만 있고 **내는 쪽이 없으면** 그 계약은 실경로에서 검증된 적이 없다 (Codex · 09-01). 그래서 「같은 reduced 행에서 엔티티와 후보를 **함께** 투영해 링크를 발행하는 결정적 함수」를 지금 무료로 완성한다. ★manifest 배선은 **D cutover 몫**이다 — adapter 는 아직 inert 다. """ def _reduced(self): from app.modules.pipeline import grounding_chunk_merge as cm rows = [ {"local_id": "c0#0", "owner_type": "location", "surface_form": "국밥집", "occurrences": [{"source_span": ep.span_of(2, "국밥집"), "source_quote": "국밥집"}], "shot_binding_status": "bound_complete", "shot_appearance_ids": ["s2#1", "s2#2"], "hard_to_generate": True, "viewers_would_notice": True}, {"local_id": "c0#1", "owner_type": "location_part", "surface_form": "이발소 회전 간판", "occurrences": [{"source_span": ep.span_of(1, "이발소 회전 간판"), "source_quote": "이발소 회전 간판"}], "shot_binding_status": "bound_complete", "shot_appearance_ids": ["s1#1", "s1#2"], "hard_to_generate": True, "viewers_would_notice": True}, ] return cm.reduce_episode(rows, [ {"remove_local_id": "c0#1", "keep_local_id": "c0#0", "relation": cm.REL_PART_OF}], segments=ep.segment_texts()) def _project(self): from app.modules.pipeline import grounding_chunk_adapter as ad return ad.project_rows_and_candidates( self._reduced(), project_id="p", episode_id="e") def test_every_registered_row_gets_a_candidate_that_points_at_it(self): from app.modules.pipeline import grounding_chunk_adapter as ad got = self._project() made = sum(len(v) for v in got["rows"].values()) linked = [c for c in got["candidates"] if ad.LINK_KEY in c] # ★후보는 **등록 안 된 행 몫까지** 나온다(증거 carry) — 링크는 등록된 # 행 수만큼만 달린다. assert made > 0 and len(linked) == made assert len(got["candidates"]) >= made by_final = {} for _k, rows in got["rows"].items(): for r in rows: by_final[r["short_id"]] = r for c in linked: link = c[ad.LINK_KEY] row = by_final[link["final_id"]] pv = row["grounding_provenance"] assert link["local_id"] == pv["local_id"], (link, pv) assert link["issuer"] == ad.ADAPTER_ISSUER assert link["contract_version"] == ad.ADAPTER_CONTRACT_VERSION def test_the_link_it_emits_is_the_one_carry_accepts(self): """★★끝점 — **내는 쪽과 받는 쪽이 실제로 맞물리는지**.""" from app.modules.pipeline import grounding_carry as gc got = self._project() built = gc.build_subjects( {k: list(v) for k, v in got["rows"].items()}, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=got["candidates"]) from app.modules.pipeline import grounding_chunk_adapter as ad linked = [c for c in got["candidates"] if ad.LINK_KEY in c] led = built["candidate_ledger"] assert led["by_disposition"][gc.DISP_CARRIED] == len(linked) assert led["by_disposition"][gc.DISP_DEFERRED] == 0, ( "★LP 가 안 붙었다 — 내는 쪽과 받는 쪽이 어긋난다") assert not [r for r in led["rows"] if r.get("link_fault")] def test_it_is_deterministic(self): assert self._project()["candidates"] == self._project()["candidates"] def test_it_invents_no_occurrence(self): """★occurrence 가 없는 행은 그 칸도 **안 만든다**.""" from app.modules.pipeline import grounding_chunk_adapter as ad got = ad.project_rows_and_candidates( {"rows": [], "registered": {}, "part_of": []}, project_id="p", episode_id="e") assert got["candidates"] == [] def test_exactly_one_step_wires_it(self): """★★**뒤집었다** (2026-09-01 D 활성화). 앞에는 「아무도 안 부른다」였다 — 그런데 그것이 바로 결함이었다: producer 가 후보를 안 내니 판별이 옛 유료 갈래를 다시 샀다 (Codex 재현). 이제 **부르는 자리가 하나**인지 잠근다. """ import ast from pathlib import Path as _P root = _P(__file__).resolve().parents[2] / "app" hits = [] for f in root.rglob("*.py"): if f.name == "grounding_chunk_adapter.py": continue try: tree = ast.parse(f.read_text(encoding="utf-8")) except SyntaxError: # noqa: PERF203 continue for n in ast.walk(tree): if (isinstance(n, ast.Attribute) and n.attr == "project_rows_and_candidates"): hits.append(f.name) assert hits == ["grounding_chunk_step.py"], \ f"★부르는 자리가 하나가 아니다: {hits}" class TestAMalformedLinkIsNotAnAbsentLink: """★★★Codex BLOCK (09-01) — 칸이 **있는데** 모양이 아니면 그것은 「없음」이 아니다. 앞 판은 `isinstance(dict) and got` 로 걸러서 — producer_link='broken' → 「없음」으로 접힘 top-level='broken' + provenance=정상 링크 → **정상 것만 보고 붙음** producer_link={} → 「없음」으로 접힘 """ ENT = {"name": "회전 간판", "short_id": "LP01", "grounding_provenance": {"local_id": "c0#1"}} def _good(self): from app.modules.pipeline.grounding_entity_contract import ( PRODUCER_CONTRACT_VERSION, PRODUCER_ISSUER) return {"issuer": PRODUCER_ISSUER, "contract_version": PRODUCER_CONTRACT_VERSION, "final_id": "LP01", "local_id": "c0#1"} def _run(self, cand_extra): from app.modules.pipeline import grounding_carry as gc from app.modules.pipeline.grounding_subject import build_subject cand = {"owner_type": "location_part", "surface_form": "x", "source_anchor": "본문", "source_quote": "x", **build_subject(project_id="p", episode_id="e", source_anchor="본문", surface_form="x", owner_type="location_part", provenance={"source_step": "grounding_a0"}), **cand_extra} got = gc.build_subjects({"location_parts": [dict(self.ENT)]}, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=[cand]) return got, got["candidate_ledger"]["rows"][0] @pytest.mark.parametrize("value", ["broken", {}, {"final_id": "LP01"}]) def test_a_present_but_malformed_link_is_explicit(self, value): from app.modules.pipeline import grounding_carry as gc got, row = self._run({gc.LEDGER_LINK: value}) assert got["carried"] == 0 assert row["link_fault"] == gc.LINK_MALFORMED assert row[gc.LINK_RAW_SLOTS] == [ {"location": gc.LINK_AT_TOP, "value": value}], "★원형이 안 남았다" def test_a_none_value_is_still_malformed_not_absent(self): from app.modules.pipeline import grounding_carry as gc got, row = self._run({gc.LEDGER_LINK: None}) assert got["carried"] == 0 assert row["link_fault"] == gc.LINK_MALFORMED def test_one_broken_place_poisons_a_good_one(self): """★★한쪽이 망가지면 **다른 쪽이 멀쩡해도** 안 믿는다.""" from app.modules.pipeline import grounding_carry as gc got, row = self._run({gc.LEDGER_LINK: "broken", "provenance": {gc.LEDGER_LINK: self._good()}}) assert got["carried"] == 0 assert row["link_fault"] == gc.LINK_MALFORMED def test_an_absent_key_carries_no_fault(self): """★음성 대조 — 칸이 **아예 없으면** 사유도 원형도 안 붙는다.""" from app.modules.pipeline import grounding_carry as gc got, row = self._run({}) assert got["carried"] == 0 assert "link_fault" not in row and gc.LINK_RAW_SLOTS not in row def test_one_good_place_binds(self): from app.modules.pipeline import grounding_carry as gc got, _row = self._run({gc.LEDGER_LINK: self._good()}) assert got["carried"] == 1 def test_two_good_places_bind(self): from app.modules.pipeline import grounding_carry as gc got, _row = self._run({gc.LEDGER_LINK: self._good(), "provenance": {gc.LEDGER_LINK: self._good()}}) assert got["carried"] == 1 class TestEveryExpectedSpanSurvivesToTheScreen: """★★★조건 1의 첫 줄 — **모든 기대 span 보존**. 실측으로 두 번 새고 있었다. ①`build_subjects` 가 `occurrences` 를 subject 에 **안 실었다** → 판별 줄의 `source_evidence` 가 실경로에서 **늘 비었다**(6줄 중 0줄) ②emitter 가 **등록 안 된 행의 후보를 안 냈다** → `outlook` 의 원문 증거가 통째로 사라졌다(기대 7 중 1 유실) """ def _chain(self): from app.modules.pipeline import (era_research as era, grounding_carry as gc, grounding_chunk_adapter as ad, grounding_chunk_merge as cm, grounding_screen as gs) rows = [] for i, t in enumerate(ep.EXPECTED_TARGETS): scene, word, nth = t["at"][0] axis = ep.AXIS_BASIS.get(t["key"]) or {} rows.append({ "local_id": f"c0#{i}", "owner_type": t["owner"], "surface_form": word, "occurrences": [{"source_span": ep.span_of(scene, word, nth), "source_quote": word}], "shot_binding_status": "bound_complete", "shot_appearance_ids": [f"s{scene}#1", f"s{scene}#2"], "hard_to_generate": bool(axis.get("hard")), "viewers_would_notice": bool(axis.get("notice"))}) red = cm.reduce_episode(rows, [], segments=ep.segment_texts()) got = ad.project_rows_and_candidates(red, project_id="p", episode_id="e") built = gc.build_subjects( {k: list(v) for k, v in got["rows"].items()}, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=got["candidates"]) disp = dict(built.get("subject_dispositions") or {}) disp.update({r["research_subject_id"]: r["disposition"] for r in built["candidate_ledger"]["rows"]}) pop = gs.build_population(built["subjects"], got["candidates"], disp) sc = gs.screen_subjects( pop, dispositions=disp, world_facts_block=ep.WORLD_FACTS, cache_get=lambda k: None, cache_put=lambda k, v: v, assess_fn=lambda **kw: {"status": era.PLAN_OK, "plan": {"assess_sha": "s", "subjects": [{"q": "x"}]}}) return red, got, sc @staticmethod def _spans(occ_holder): return {(o["source_span"]["segment_id"], o["source_span"]["start"], o["source_span"]["end"]) for o in occ_holder} def test_not_one_expected_span_is_lost(self): red, _got, sc = self._chain() want = set() for r in red["rows"]: want |= self._spans(r["occurrences"]) have = set() for r in sc["rows"]: have |= self._spans((r.get("source_evidence") or {}) .get("occurrences") or []) assert want and want == have, f"★빠진 span: {sorted(want - have)}" def test_the_outlook_evidence_survives_without_an_entity_row(self): """★★`outlook` 은 여기서 엔티티 행이 안 생긴다 — 그래도 증거는 간다.""" from app.modules.pipeline import grounding_chunk_adapter as ad from app.modules.pipeline import grounding_screen as gs _red, got, sc = self._chain() assert "outlooks" not in got["rows"], "★outlook 을 base 행으로 만들었다" ol = [c for c in got["candidates"] if c["owner_type"] == "outlook"] assert len(ol) == 1 assert ad.LINK_KEY not in ol[0], "★가리킬 행이 없는데 링크를 달았다" assert ol[0]["occurrences"], "★증거가 안 실렸다" left = gs.facet_obligations(sc) assert [r["owner_type"] for r in left] == ["outlook"] def test_a_row_without_occurrences_gets_no_invented_ones(self): from app.modules.pipeline import grounding_carry as gc built = gc.build_subjects( {"props": [{"name": "가위", "short_id": "P01"}]}, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=[]) for s in built["subjects"]: assert "occurrences" not in s class TestTheEmitterGivesEachRowItsOwnIdentity: """★★★Codex BLOCK (09-01) — 후보 신원이 겹쳤다. 앞 판은 `source_anchor` 를 `segment_id` 만으로 만들어서, 같은 씬 안의 **같은 owner·같은 표면형** 두 행이 서로 다른 span·`local_id`·`final_id` 를 가져도 **같은 `research_subject_id`** 를 받았다. 같은 이름의 다른 것은 한 씬에도 있다. 실제 89행에서 우연히 안 겹친 것은 계약의 증명이 아니다. """ TEXT = "표 하나와 표 둘이 있었다." def _rows(self, *starts): return [{ "local_id": f"c0#{i}", "owner_type": "prop", "surface_form": "표", "occurrences": [{"source_span": {"segment_id": "scene-1", "start": s, "end": s + 1}, "source_quote": "표"}], "shot_binding_status": "bound_complete", "shot_appearance_ids": ["s1#1", "s1#2"], "hard_to_generate": True, "viewers_would_notice": True} for i, s in enumerate(starts, start=1)] def _project(self, *starts): from app.modules.pipeline import grounding_chunk_adapter as ad from app.modules.pipeline import grounding_chunk_merge as cm red = cm.reduce_episode(self._rows(*starts), [], segments={"scene-1": self.TEXT}) return red, ad.project_rows_and_candidates( red, project_id="p", episode_id="e") def test_same_scene_same_word_different_spans_get_two_identities(self): from app.modules.pipeline import grounding_chunk_adapter as ad a, b = self.TEXT.index("표"), self.TEXT.index("표", 1) _red, got = self._project(a, b) ids = [c["research_subject_id"] for c in got["candidates"]] assert len(ids) == 2 and len(set(ids)) == 2, ids # ★각자 **제 좌표**를 가리킨다 pairs = {(c[ad.LINK_KEY]["final_id"], c[ad.LINK_KEY]["local_id"]) for c in got["candidates"]} assert pairs == {("P01", "c0#1"), ("P02", "c0#2")}, pairs def test_both_bind_at_the_receiving_end(self): """★★Codex 끝점 — `build_subjects` 에서 **carried=2**. 이름 경로는 「둘 걸렸다 → 애매하다」로 아무것도 안 붙인다. 그래서 **링크가 이름보다 세야** 한다 — producer 가 낸 링크는 그 행의 사실이고 이름은 짐작이다. """ from app.modules.pipeline import grounding_carry as gc a, b = self.TEXT.index("표"), self.TEXT.index("표", 1) _red, got = self._project(a, b) built = gc.build_subjects( {k: list(v) for k, v in got["rows"].items()}, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=got["candidates"]) led = built["candidate_ledger"] assert led["by_disposition"][gc.DISP_CARRIED] == 2, led assert {r[gc.LEDGER_LINK]["final_id"] for r in led["rows"]} == { "P01", "P02"} def test_a_row_without_a_span_stops_instead_of_a_blank_anchor(self): """★빈 anchor 로 후보를 만들면 그것이 곧 신원 충돌의 씨앗이다.""" from app.modules.pipeline import grounding_chunk_adapter as ad red, _got = self._project(self.TEXT.index("표")) for r in red["rows"]: r["occurrences"] = [] with pytest.raises(ad.MissingSourceSpan): ad.project_rows_and_candidates(red, project_id="p", episode_id="e") def test_the_emitter_checks_uniqueness_itself(self): """★받는 쪽이 세우기 **전에** emitter 가 스스로 본다.""" import inspect from app.modules.pipeline import grounding_chunk_adapter as ad src = inspect.getsource(ad.project_rows_and_candidates) assert "CandidateIdCollision" in src assert hasattr(ad, "CandidateIdCollision") def test_a_link_only_wins_when_it_is_verified(self): """★음성 대조 — 링크가 이름을 이기는 것은 **검증된** 링크뿐이다.""" from app.modules.pipeline import grounding_carry as gc from app.modules.pipeline import grounding_chunk_adapter as ad a, b = self.TEXT.index("표"), self.TEXT.index("표", 1) _red, got = self._project(a, b) for c in got["candidates"]: c[ad.LINK_KEY]["issuer"] = "아무개" built = gc.build_subjects( {k: list(v) for k, v in got["rows"].items()}, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=got["candidates"]) led = built["candidate_ledger"] assert led["by_disposition"][gc.DISP_CARRIED] == 0 assert all(r.get("link_fault") == gc.LINK_UNKNOWN_ISSUER for r in led["rows"]) class TestEveryLinkSlotIsKeptWithItsLocation: """★★★「원형 보존」이 두 경우 틀렸다 (Codex · 09-01) — `None` 은 칸이 있는데 **안 남았고**, 두 자리가 다르면 **둘째를 잃었다**.""" ENT = {"name": "회전 간판", "short_id": "LP01", "grounding_provenance": {"local_id": "c0#1"}} def _good(self, **kw): from app.modules.pipeline.grounding_entity_contract import ( PRODUCER_CONTRACT_VERSION, PRODUCER_ISSUER) return {"issuer": PRODUCER_ISSUER, "contract_version": PRODUCER_CONTRACT_VERSION, "final_id": "LP01", "local_id": "c0#1", **kw} def _row(self, extra): from app.modules.pipeline import grounding_carry as gc from app.modules.pipeline.grounding_subject import build_subject cand = {"owner_type": "location_part", "surface_form": "x", "source_anchor": "본문", "source_quote": "x", **build_subject(project_id="p", episode_id="e", source_anchor="본문", surface_form="x", owner_type="location_part", provenance={"source_step": "grounding_a0"}), **extra} got = gc.build_subjects({"location_parts": [dict(self.ENT)]}, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=[cand]) return got["candidate_ledger"]["rows"][0] @pytest.mark.parametrize("value", [None, "", {}, "broken"]) def test_a_present_value_is_kept_whatever_it_is(self, value): from app.modules.pipeline import grounding_carry as gc row = self._row({gc.LEDGER_LINK: value}) assert row[gc.LINK_RAW_SLOTS] == [ {"location": gc.LINK_AT_TOP, "value": value}] def test_both_places_are_kept_when_they_disagree(self): from app.modules.pipeline import grounding_carry as gc other = self._good(final_id="LP99") row = self._row({gc.LEDGER_LINK: self._good(), "provenance": {gc.LEDGER_LINK: other}}) slots = row[gc.LINK_RAW_SLOTS] assert [x["location"] for x in slots] == [gc.LINK_AT_TOP, gc.LINK_AT_PROVENANCE] assert slots[1]["value"] == other, "★무엇과 부딪혔는지 잃었다" assert row["link_fault"] == gc.LINK_STORAGE_CONFLICT def test_no_link_has_no_slots(self): from app.modules.pipeline import grounding_carry as gc row = self._row({}) assert gc.LINK_RAW_SLOTS not in row and "link_fault" not in row class TestTheEmitterNeverInventsAnObligation: """★★★Codex 가 짚은 자리 (09-01) — **미등록 행이 뜻밖의 의무를 만들었다.** 「등록 안 된 행도 후보는 낸다」로 넓혔더니, producer 가 **일부러 등록 안 한** 행(문턱 미달)까지 후보가 되어 `promoted` 로 승격됐다 — 실측: 등록 1인데 **의무 2**. 없던 참조 의무가 생긴 것이다. 후보를 내는 자리는 **둘뿐이다** — ①등록된 행 → 링크를 달아 낸다 ②여기서 엔티티가 **아예 안 생기는 갈래**(`outlook`) → 증거만, 링크 없이 """ TEXT = "표 하나와 종이 있었다." def _row(self, lid, s, e, shots, hard, owner="prop"): return {"local_id": lid, "owner_type": owner, "surface_form": self.TEXT[s:e], "occurrences": [{"source_span": {"segment_id": "scene-1", "start": s, "end": e}, "source_quote": self.TEXT[s:e]}], "shot_binding_status": "bound_complete", "shot_appearance_ids": shots, "hard_to_generate": hard, "viewers_would_notice": hard} def _project(self, rows): from app.modules.pipeline import grounding_chunk_adapter as ad from app.modules.pipeline import grounding_chunk_merge as cm red = cm.reduce_episode(rows, [], segments={"scene-1": self.TEXT}) return red, ad.project_rows_and_candidates(red, project_id="p", episode_id="e") def test_a_row_the_producer_did_not_register_gets_no_candidate(self): red, got = self._project([ self._row("c0#1", 0, 1, ["s1#1", "s1#2"], True), self._row("c0#2", 6, 7, ["s1#1"], False)]) assert red["registered"]["c0#2"]["registered"] is False assert len(got["candidates"]) == 1, [ c["surface_form"] for c in got["candidates"]] assert got["candidates"][0]["surface_form"] == "표" def test_an_owner_with_no_entity_row_here_still_carries_evidence(self): from app.modules.pipeline import grounding_chunk_adapter as ad _red, got = self._project([ self._row("c0#1", 0, 1, ["s1#1", "s1#2"], True), self._row("c0#3", 6, 7, ["s1#1", "s1#2"], True, owner="outlook")]) ol = [c for c in got["candidates"] if c["owner_type"] == "outlook"] assert len(ol) == 1 and ad.LINK_KEY not in ol[0] assert ol[0]["occurrences"], "★증거가 안 실렸다" def test_the_gate_is_the_ledger_not_the_owner(self): """★★★문은 **`registered` 하나**다 (Codex · 09-01) — owner 로 가르면 heuristic 이다. 미등록 `outlook` 까지 다 내면 모집단과 뒤 비용이 는다. """ from app.modules.pipeline import grounding_chunk_adapter as ad red, got = self._project([ self._row("c0#1", 0, 1, ["s1#1", "s1#2"], True), self._row("c0#2", 6, 7, ["s1#1"], False), self._row("c0#3", 6, 7, ["s1#1", "s1#2"], True, owner="outlook"), self._row("c0#4", 0, 1, ["s1#1"], False, owner="outlook")]) reg = {k: v.get("registered") for k, v in red["registered"].items()} assert reg["c0#2"] is not True and reg["c0#4"] is not True got_pairs = {(c["owner_type"], ad.LINK_KEY in c) for c in got["candidates"]} assert got_pairs == {("prop", True), ("outlook", False)}, got_pairs assert len(got["candidates"]) == 2 def test_the_obligation_count_matches_what_was_registered(self): """★끝점 — 없던 의무가 안 생긴다.""" from app.modules.pipeline import era_research as era from app.modules.pipeline import grounding_carry as gc from app.modules.pipeline import grounding_screen as gs _red, got = self._project([ self._row("c0#1", 0, 1, ["s1#1", "s1#2"], True), self._row("c0#2", 6, 7, ["s1#1"], False)]) built = gc.build_subjects( {k: list(v) for k, v in got["rows"].items()}, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=got["candidates"]) disp = dict(built.get("subject_dispositions") or {}) disp.update({r["research_subject_id"]: r["disposition"] for r in built["candidate_ledger"]["rows"]}) pop = gs.build_population(built["subjects"], got["candidates"], disp) sc = gs.screen_subjects( pop, dispositions=disp, world_facts_block="세계 사실", cache_get=lambda k: None, cache_put=lambda k, v: v, assess_fn=lambda **kw: {"status": era.PLAN_OK, "plan": {"assess_sha": "s", "subjects": [{"q": "x"}]}}) assert len(sc["rows"]) == 1, [r["owner_type"] for r in sc["rows"]] assert len(gs.obligation_ids(sc)) == 1 class TestADeclaredLinkNeverFallsBackToNames: """★★★Codex BLOCK (09-01) — 링크를 **선언해 놓고 안 맞으면** base 갈래가 **이름으로 되살아나** 붙었다. 같은 장부 줄에 `disposition='carried'` 와 `link_fault='unknown_issuer'` 가 **동시에** 적혔다 — 모순이다. """ ENTS = {"props": [{"name": "표", "short_id": "P01", "grounding_provenance": {"local_id": "c0#1"}}], "characters": [{"name": "이발사", "short_id": "C01", "grounding_provenance": {"local_id": "c0#2"}}], "locations": [{"name": "국밥집", "short_id": "L01", "grounding_provenance": {"local_id": "c0#3"}}]} def _cand(self, owner, surface, link=None): from app.modules.pipeline import grounding_carry as gc c = {"owner_type": owner, "surface_form": surface, "research_subject_id": f"a0-{owner}", "source_anchor": "본문", "source_quote": surface} if link is not None: c[gc.LEDGER_LINK] = link return c def _run(self, cand): from app.modules.pipeline import grounding_carry as gc got = gc.build_subjects({k: [dict(x) for x in v] for k, v in self.ENTS.items()}, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=[cand]) rid = cand["research_subject_id"] row = next(r for r in got["candidate_ledger"]["rows"] if r["research_subject_id"] == rid) mine = [s for s in got["subjects"] if s["research_subject_id"] == rid] return got, row, mine @pytest.mark.parametrize("owner,surface", [ ("prop", "표"), ("character", "이발사"), ("location", "국밥집")]) @pytest.mark.parametrize("bad", ["issuer", "coord", "malformed"]) def test_an_invalid_link_binds_nothing_and_promotes_nothing( self, owner, surface, bad): from app.modules.pipeline import grounding_carry as gc from app.modules.pipeline.grounding_entity_contract import ( PRODUCER_CONTRACT_VERSION, PRODUCER_ISSUER) good = {"issuer": PRODUCER_ISSUER, "contract_version": PRODUCER_CONTRACT_VERSION, "final_id": "P01", "local_id": "c0#1"} link = ({**good, "issuer": "bogus"} if bad == "issuer" else {**good, "final_id": "ZZ99", "local_id": "c9#9"} if bad == "coord" else "broken") got, row, mine = self._run(self._cand(owner, surface, link)) assert got["carried"] == 0, "★이름으로 되살아났다" assert row["disposition"] == gc.DISP_UNRESOLVED assert row["link_fault"] in gc.LINK_BROKEN assert row[gc.LINK_RAW_SLOTS], "★원형이 안 남았다" assert mine == [], "★승격으로 되살아났다" @pytest.mark.parametrize("owner,surface", [ ("prop", "표"), ("character", "이발사"), ("location", "국밥집")]) def test_a_candidate_without_a_link_still_carries_by_name(self, owner, surface): """★비회귀 — 링크를 **안 건** 후보는 기존 이름 경로 그대로.""" from app.modules.pipeline import grounding_carry as gc got, row, _mine = self._run(self._cand(owner, surface)) assert got["carried"] == 1 assert row["disposition"] == gc.DISP_CARRIED def test_a_valid_link_binds_only_to_its_own_coordinate(self): """★이름이 겹쳐도 **제 좌표**로만 붙는다.""" from app.modules.pipeline import grounding_carry as gc from app.modules.pipeline.grounding_entity_contract import ( PRODUCER_CONTRACT_VERSION, PRODUCER_ISSUER) got, row, mine = self._run(self._cand("prop", "표", { "issuer": PRODUCER_ISSUER, "contract_version": PRODUCER_CONTRACT_VERSION, "final_id": "P01", "local_id": "c0#1"})) assert row["disposition"] == gc.DISP_CARRIED assert (mine[0].get("provenance") or {})["short_id"] == "P01" class TestASpanThatStartsAtZeroKeepsItsIdentity: """★`start=0` 은 정상 좌표다 — 참/거짓으로 보면 잃는다.""" def test_zero_offsets_make_a_real_anchor(self): from app.modules.pipeline import grounding_chunk_adapter as ad assert ad._span_anchor({"segment_id": "scene-1", "start": 0, "end": 3}) == "scene-1:0-3" assert ad._span_anchor({"segment_id": "scene-1", "start": 0, "end": 0}) == "scene-1:0-0" def test_a_missing_offset_makes_no_anchor(self): from app.modules.pipeline import grounding_chunk_adapter as ad assert ad._span_anchor({"segment_id": "scene-1", "start": 0}) == "" assert ad._span_anchor({"start": 0, "end": 1}) == "" class TestABoundRowNeverCarriesAFault: """★★★같은 모순이 **다른 모양으로** 남아 있었다 (제 실측 · 09-01). 링크가 **둘째** 엔티티를 정확히 가리켜도, 첫 엔티티와 대조할 때 「어긋남」이 한 번 기록되면 그 사유가 **붙은 줄에 그대로 실렸다** — `disposition='carried'` 와 `link_fault='coordinate_mismatch'` 가 한 줄에 같이 있었다. """ ENTS = {"props": [ {"name": "가", "short_id": "P01", "grounding_provenance": {"local_id": "c0#1"}}, {"name": "나", "short_id": "P02", "grounding_provenance": {"local_id": "c0#2"}}]} def _run(self, link): from app.modules.pipeline import grounding_carry as gc cand = {"owner_type": "prop", "surface_form": "나", "research_subject_id": "a0-x", "source_anchor": "본문", "source_quote": "나", gc.LEDGER_LINK: link} got = gc.build_subjects({k: [dict(x) for x in v] for k, v in self.ENTS.items()}, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=[cand]) return got, got["candidate_ledger"]["rows"][0] def _link(self, **kw): from app.modules.pipeline.grounding_entity_contract import ( PRODUCER_CONTRACT_VERSION, PRODUCER_ISSUER) return {"issuer": PRODUCER_ISSUER, "contract_version": PRODUCER_CONTRACT_VERSION, **kw} def test_a_link_to_the_second_row_binds_without_a_fault(self): from app.modules.pipeline import grounding_carry as gc got, row = self._run(self._link(final_id="P02", local_id="c0#2")) assert got["carried"] == 1 assert row["disposition"] == gc.DISP_CARRIED assert "link_fault" not in row, "★붙었는데 사유가 붙었다" assert gc.LINK_RAW_SLOTS not in row def test_a_link_to_the_first_row_also_binds_without_a_fault(self): """★음성 대조 — 차례가 바뀌어도 같다.""" from app.modules.pipeline import grounding_carry as gc got, row = self._run(self._link(final_id="P01", local_id="c0#1")) assert got["carried"] == 1 and "link_fault" not in row def test_a_link_that_matches_nothing_keeps_its_fault(self): from app.modules.pipeline import grounding_carry as gc got, row = self._run(self._link(final_id="P99", local_id="c9#9")) assert got["carried"] == 0 assert row["disposition"] == gc.DISP_UNRESOLVED assert row["link_fault"] in gc.LINK_BROKEN assert row[gc.LINK_RAW_SLOTS] def test_no_row_ever_has_both(self): """★★불변식 — 한 줄에 `carried` 와 사유가 **같이 있을 수 없다**.""" from app.modules.pipeline import grounding_carry as gc for link in (self._link(final_id="P01", local_id="c0#1"), self._link(final_id="P02", local_id="c0#2"), self._link(final_id="P99"), self._link(issuer="bogus", final_id="P02"), "broken"): _got, row = self._run(link) assert not (row["disposition"] == gc.DISP_CARRIED and "link_fault" in row), (link, row) class TestTheProducerDecisionSurvivesToTheScreen: """★★★Codex BLOCK (09-01) — C(c) 가 **유료로 낸 구조화 판정**이 emitter→subject→screen 에서 통째로 사라졌다. 잃으면 D 에서 **다시 사거나 정보 없이 판단**한다. 그리고 쓰임이 갈린다 — `coarse_type_label` 은 **VLM 종류 판정에만**, `visual_brief`·검색어·언어 잠금은 **검색 저작에만**. 섞어 보내면 VLM 이 「무엇인가」 대신 「어떻게 생겼나」로 답한다. """ ROW = {"local_id": "c0#1", "owner_type": "prop", "surface_form": "고무신", "occurrences": [{"source_span": {"segment_id": "scene-1", "start": 0, "end": 3}, "source_quote": "고무신"}], "shot_binding_status": "bound_complete", "shot_appearance_ids": ["s1#1", "s1#2"], "hard_to_generate": True, "viewers_would_notice": True, "coarse_type_label": "고무신의 앞부분", "visual_brief": "검정 고무신", "search_terms_native": ["검정 고무신"], "language_lock_native": "ko"} def _chain(self, row=None, owner=None): from app.modules.pipeline import era_research as era from app.modules.pipeline import grounding_carry as gc from app.modules.pipeline import grounding_chunk_adapter as ad from app.modules.pipeline import grounding_chunk_merge as cm from app.modules.pipeline import grounding_screen as gs r = dict(row or self.ROW) if owner: r["owner_type"] = owner red = cm.reduce_episode([r], [], segments={"scene-1": "고무신이 놓여 있었다."}) got = ad.project_rows_and_candidates(red, project_id="p", episode_id="e") built = gc.build_subjects( {k: list(v) for k, v in got["rows"].items()}, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=got["candidates"]) disp = dict(built.get("subject_dispositions") or {}) disp.update({x["research_subject_id"]: x["disposition"] for x in built["candidate_ledger"]["rows"]}) pop = gs.build_population(built["subjects"], got["candidates"], disp) sc = gs.screen_subjects( pop, dispositions=disp, world_facts_block="세계 사실", cache_get=lambda k: None, cache_put=lambda k, v: v, assess_fn=lambda **kw: {"status": era.PLAN_OK, "plan": {"assess_sha": "s", "subjects": [{"q": "x"}]}}) return red, got, built, sc def test_every_structured_field_survives_verbatim(self): from app.modules.pipeline import grounding_entity_contract as ec red, got, built, sc = self._chain() want = ec.producer_payload(red["rows"][0]) assert set(want) == set(ec.PRODUCER_PAYLOAD_FIELDS), sorted(want) assert got["candidates"][0][ec.PRODUCER_PAYLOAD] == want assert any(s.get(ec.PRODUCER_PAYLOAD) == want for s in built["subjects"]) assert any(r.get(ec.PRODUCER_PAYLOAD) == want for r in sc["rows"]) def test_the_outlook_keeps_it_without_an_entity_row(self): """★entity hit 유무와 **무관하게** reduced 행에서 받는다.""" from app.modules.pipeline import grounding_chunk_adapter as ad from app.modules.pipeline import grounding_entity_contract as ec red, got, _built, sc = self._chain(owner="outlook") assert "outlooks" not in got["rows"] want = ec.producer_payload(red["rows"][0]) c = got["candidates"][0] assert ad.LINK_KEY not in c assert c[ec.PRODUCER_PAYLOAD] == want assert any(r.get(ec.PRODUCER_PAYLOAD) == want for r in sc["rows"]) def test_it_is_not_mixed_into_the_manuscript_evidence(self): """★★원고가 말한 것과 모델이 판단한 것을 **섞지 않는다**.""" from app.modules.pipeline import grounding_entity_contract as ec _red, _got, _built, sc = self._chain() row = next(r for r in sc["rows"] if r.get(ec.PRODUCER_PAYLOAD)) ev = row["source_evidence"] for k in ec.PRODUCER_PAYLOAD_FIELDS: assert k not in ev, f"★{k} 가 원문 증거에 섞였다" def test_the_judge_never_sees_how_it_looks(self): """★★★막는 것은 **한 방향** 하나다 (2026-09-02 정정). 앞 판은 두 표가 **안 겹치는지**를 잠갔다. 그런데 검색 지시문 저작기(`grounding_ref_brief.build_brief_user`)가 `coarse_type_label` 을 **요구한다** — 겹침을 금지한 탓에 첫 대상이 저작 전에 `BriefInputsMissing` 으로 죽었다 (Codex BLOCK). 진짜 금지는 **심판이 `visual_brief` 를 보는 것**이다. 그것을 주면 VLM 이 「무엇인가」가 아니라 「어떻게 생겼나 · 맞게 생겼나」로 답한다 — 사람만 하는 판정이다 (사용자 확정 08-31). """ from app.modules.pipeline import grounding_entity_contract as ec assert "visual_brief" not in ec.JUDGE_FIELDS assert "search_terms_native" not in ec.JUDGE_FIELDS assert "language_lock_native" not in ec.JUDGE_FIELDS # ★반대 방향은 **열려 있어야** 한다 — 부류 이름 없이는 못 찾는다. assert "coarse_type_label" in ec.SEARCH_FIELDS # ★`surface_form` 은 producer 가 **판단한** 것이 아니라 reduced 행이 # 이미 가진 정본 표기다. 그래서 payload 밖이지만 저작기는 읽는다 # — chunk schema 에 있는지는 `test_the_two_split_tables_are_covered_too`. assert set(ec.JUDGE_FIELDS) | set(ec.SEARCH_FIELDS) <= set( ec.PRODUCER_PAYLOAD_FIELDS) | {"surface_form"} def test_a_row_without_those_fields_gets_no_empty_payload(self): """★없는 것을 **만들지 않는다**.""" from app.modules.pipeline import grounding_entity_contract as ec bare = {k: v for k, v in self.ROW.items() if k not in ec.PRODUCER_PAYLOAD_FIELDS} bare["shot_binding_status"] = "bound_complete" bare["shot_appearance_ids"] = ["s1#1", "s1#2"] bare["hard_to_generate"] = True bare["viewers_would_notice"] = True _red, got, _built, _sc = self._chain(row=bare) payload = got["candidates"][0].get(ec.PRODUCER_PAYLOAD) or {} assert "coarse_type_label" not in payload assert "visual_brief" not in payload def test_the_contract_version_was_bumped(self): """★모양이 바뀌면 **판을 올린다** — 옛 판이 조용히 섞이면 안 된다. ★★판 번호를 **못박지 않는다** (2026-09-01). 앞 판은 `"2."` 으로 시작하는지를 잠갔는데, 그러면 **다음에 모양이 바뀔 때 이 시험이 올리는 것을 막는다** — 래칫이 거꾸로 선다. 지금 것은 `KNOWN_PRODUCERS` 가 그 판을 **받는가**와, 옛 판을 **버리지 않는가**다. """ from app.modules.pipeline import grounding_entity_contract as ec got = ec.KNOWN_PRODUCERS[ec.PRODUCER_ISSUER] assert ec.PRODUCER_CONTRACT_VERSION in got # ★이미 적힌 링크가 갑자기 「모르는 계약」이 되면 안 된다 assert got == ec.PRODUCER_CONTRACT_ACCEPTED assert len(set(got)) == len(got), "★같은 판이 두 번 있다"