"""GROUNDING-V2 — A0 후보가 **소리 없이 사라지지 않는다**. ★`carry_reasons` 는 **엔티티**를 세지 후보를 안 센다. 그래서 어느 엔티티에도 안 걸린 후보는 아무 데도 안 세어졌다 — 실측에서 후보 23개 중 장부에 나타난 것이 2개뿐이었다. """ import json import pathlib import pytest from app.modules.pipeline.grounding_carry import ( DISP_CARRIED, DISP_CONTESTED, DISP_DEFERRED, DISP_PROMOTED, DISP_UNRESOLVED, DISPOSITIONS, SCANNED_OWNERS, bindable_owners, build_subjects, disposition_of, ) def _cand(rid, surface, owner="prop", quote="원문"): return {"research_subject_id": rid, "surface_form": surface, "owner_type": owner, "source_anchor": "a", "source_quote": quote} def _run(cands, entities): return build_subjects(entities, project_id="p", episode_id="e", source_step="entity_merge", a0_candidates=cands) class TestEveryCandidateLandsSomewhere: """★통과 조건은 「소리 없이 사라지는 후보 **0건**」이다 (Codex).""" def test_the_ledger_totals_the_candidates(self): cands = [_cand(f"rs_{i}", f"물건{i}") for i in range(9)] led = _run(cands, {"props": [{"name": "물건0"}]})["candidate_ledger"] assert led["total"] == 9 assert sum(led["by_disposition"].values()) == 9 def test_a_matched_candidate_is_carried(self): led = _run([_cand("rs_1", "요금통")], {"props": [{"name": "요금통", "short_id": "P01"}]} )["candidate_ledger"] assert led["by_disposition"][DISP_CARRIED] == 1 def test_a_base_owner_no_entity_wants_is_promoted_not_invisible(self): """★★이게 안 보이던 것이다 — 21개가 아무 데도 안 세어졌다. 그리고 **세는 것으로 끝나지 않는다** — base owner 는 A0 신원 그대로 subject 가 된다. 안 그러면 원고에서 건진 그 인용이 파이프라인에 아예 안 들어간다. """ out = _run([_cand("rs_1", "아무도 안 부르는 것")], {"props": [{"name": "딴것"}]}) led = out["candidate_ledger"] assert led["by_disposition"][DISP_PROMOTED] == 1 assert led["rows"][0]["surface_form"] == "아무도 안 부르는 것" promoted = [s for s in out["subjects"] if s["research_subject_id"] == "rs_1"] assert promoted, "장부에만 적고 subject 로 안 만들었다" assert promoted[0]["quote_source"] == "manuscript" assert promoted[0]["source_quote"] == "원문" def test_two_entities_wanting_one_candidate_is_contested(self): led = _run([_cand("rs_1", "요금통")], {"props": [{"name": "요금통 하나"}, {"name": "요금통 둘"}]})["candidate_ledger"] assert led["by_disposition"][DISP_CONTESTED] == 1 def test_an_owner_with_no_entity_lane_is_deferred_not_unmatched(self): """★★「시도했는데 안 붙었다」와 「시도조차 못 했다」는 다르다. `outlook` 은 `BINDABLE_OWNER_GROUPS` 에 있는데 `ENTITY_KEY_TO_OWNER` 에 갈래가 **없다** — 같은 규칙을 두 곳에 적어 한쪽만 고쳐진 부류다. 고칠 곳이 다르므로 갈라 센다. """ out = _run([_cand("rs_1", "감색 차장 제복", owner="outlook")], {"props": []}) led = out["candidate_ledger"] assert led["by_disposition"][DISP_DEFERRED] == 1 assert led["by_disposition"][DISP_PROMOTED] == 0 assert led["deferred_owners"] == {"outlook": 1} # ★승격하면 그 producer 가 만들 때 같은 대상이 둘이 된다 assert not [s for s in out["subjects"] if s["research_subject_id"] == "rs_1"] def test_location_part_is_tried_but_deferred_when_it_misses(self): """★`location_part` 는 `prop` 과 같은 묶음이라 **결속은 시도한다**. 그래도 안 붙으면 `deferred` 다 — §2-6.5 producer 가 가질 것이다.""" out = _run([_cand("rs_1", "그 아래 요금통", owner="location_part")], {"props": [{"name": "딴것"}]}) assert out["candidate_ledger"]["by_disposition"][DISP_DEFERRED] == 1 assert out["candidate_ledger"]["by_disposition"][DISP_PROMOTED] == 0 def test_location_part_never_binds_to_a_prop_by_name(self): """★★**뒤집힌 시험**이다 (§2-6.5a · Codex BLOCK · 09-01). 앞에는 「묶음이 살아 있는지」를 positive control 로 잠갔다 — `prop` 과 `location_part` 가 한 묶음이라 이름이 겹치면 붙었다. §2-6.5a 로 LP 행이 실제로 생기자 그 문이 **양방향**이 됐고, 다른 부분 대상의 근거가 조용히 건너갔다. 이제 갈랐다. """ out = _run([_cand("rs_1", "요금통", owner="location_part")], {"props": [{"name": "요금통"}]}) assert out["candidate_ledger"]["by_disposition"][DISP_CARRIED] == 0 assert out["candidate_ledger"]["by_disposition"][DISP_DEFERRED] == 1 def test_a_prop_still_binds_to_a_prop_by_name(self): """★음성 대조 — base 갈래의 기존 이름 carry 는 **그대로**다.""" out = _run([_cand("rs_1", "요금통", owner="prop")], {"props": [{"name": "요금통"}]}) assert out["candidate_ledger"]["by_disposition"][DISP_CARRIED] == 1 def test_the_deferred_rule_reads_both_tables(self): """★★한 표만 보면 다른 표가 바뀌어도 안 따라온다. `outlooks` 갈래가 생기는 날 이 시험이 **저절로** 통과를 바꾼다. """ for owner in ("character", "location", "prop"): assert bindable_owners(owner) & SCANNED_OWNERS assert disposition_of(owner) == DISP_PROMOTED # facet 은 뒤 producer 가 가진다 assert disposition_of("location_part") == DISP_DEFERRED assert disposition_of("outlook") == DISP_DEFERRED assert "outlook" not in SCANNED_OWNERS # 모르는 owner 는 **조용히 승격하지 않는다** assert disposition_of("무엇인지 모름") == DISP_UNRESOLVED def test_the_sum_is_enforced_not_just_reported(self): """★합이 안 맞는데 그냥 돌려주면 그 장부는 있으나 마나다.""" from app.modules.pipeline.grounding_carry import DISP_ENTITY_ONLY # ★`entity_only` 는 **A0 후보가 없는 기존 엔티티 행**이다 — 후보 장부가 # 아니라 subject 쪽 처분이라, 후보 합계에는 0 으로 나온다. assert set(DISPOSITIONS) == {DISP_ENTITY_ONLY, DISP_CARRIED, DISP_PROMOTED, DISP_DEFERRED, DISP_CONTESTED, DISP_UNRESOLVED} def test_no_candidates_is_an_empty_ledger_not_a_missing_one(self): """★없는 것을 **없다고** 남긴다 — 칸이 통째로 빠지면 조회가 죽는다.""" led = _run([], {"props": [{"name": "무엇"}]})["candidate_ledger"] assert led["total"] == 0 assert set(led["by_disposition"]) == set(DISPOSITIONS) class TestTheLedgerIsNotAReplacementForCarryReasons: """★둘은 **다른 것을 센다** — 하나로 합치면 한쪽이 사라진다.""" def test_carry_reasons_still_counts_entities(self): out = _run([_cand("rs_1", "요금통")], {"props": [{"name": "요금통"}, {"name": "딴것"}]}) assert sum(out["carry_reasons"].values()) == 2 # 엔티티 2개 assert out["candidate_ledger"]["total"] == 1 # 후보 1개 class TestACandidateThatWasTriedAndFailedIsNotPromoted: """★★「어느 엔티티도 안 불렀다」와 「불렸는데 못 정했다」는 **다르다**. 두 후보가 한 엔티티에 걸린 것이 「서로 다른 두 대상」인지 「같은 것을 두 번 적은 것」인지 **기계적으로 못 가른다**. 승격하면 같은 것을 두 번 조사할 수 있고, 그건 돈이다. 그리고 프로덕션 사슬 시험이 이미 「못 붙인 것을 분류기에 보내지 않는다」를 계약으로 잡고 있었다 — 승격이 그것을 깼었다. """ def test_an_ambiguous_candidate_is_unresolved_not_promoted(self): out = _run([_cand("rs_1", "승차권"), _cand("rs_2", "승차권 뭉치")], {"props": [{"name": "승차권 뭉치"}]}) led = out["candidate_ledger"] assert led["by_disposition"]["unresolved"] == 2 assert led["by_disposition"][DISP_PROMOTED] == 0 assert out["subjects"] == [], "못 정한 것을 분류기에 보냈다" def test_a_candidate_nobody_called_is_still_promoted(self): """★positive control — 걸린 적 없는 것은 그대로 승격한다.""" out = _run([_cand("rs_1", "승차권"), _cand("rs_2", "승차권 뭉치"), _cand("rs_3", "기계식 요금통")], {"props": [{"name": "승차권 뭉치"}]}) led = out["candidate_ledger"] assert led["by_disposition"][DISP_PROMOTED] == 1 assert [s["research_subject_id"] for s in out["subjects"]] == ["rs_3"] def test_the_five_sets_never_overlap(self): """★한 후보가 두 갈래에 들면 합계가 맞아도 장부가 거짓이다.""" out = _run([_cand("rs_1", "승차권"), _cand("rs_2", "승차권 뭉치"), _cand("rs_3", "기계식 요금통"), _cand("rs_4", "제복", owner="outlook")], {"props": [{"name": "승차권 뭉치"}]}) led = out["candidate_ledger"] ids = [r["research_subject_id"] for r in led["rows"]] assert len(ids) == len(set(ids)) == 4 assert sum(led["by_disposition"].values()) == 4 class TestPromotionMustReachTheCompletenessGate: """★★승격이 **반쪽이면 하류가 막힌다**. `completeness_report` 는 「A0 후보가 엔티티 목록에 남았나」를 묻는다. 승격한 후보는 **엔티티에 없어서 승격한 것**이라 그 물음에 당연히 없다 — 그런데 그건 사라진 게 아니라 **다른 자리로 간 것**이다. 실측에서 승격한 7개가 그대로 `missing` 으로 세어져 `complete=False` → 스텝 `partial` → 하류 차단. """ def _report(self, promoted_ids=None): from app.modules.pipeline.grounding_overlay import completeness_report cands = [{"research_subject_id": "rs_1", "owner_type": "prop", "surface_form": "아무도 안 부르는 것"}] return completeness_report(cands, {"prop": [{"name": "딴것"}]}, promoted_ids=promoted_ids) def test_a_promoted_candidate_is_not_counted_missing(self): r = self._report(promoted_ids=["rs_1"]) assert r["missing_count"] == 0 assert r["promoted_count"] == 1 assert r["complete"] is True def test_without_the_promoted_ids_it_is_still_missing(self): """★positive control — 안 넘기면 옛 동작 그대로다.""" r = self._report() assert r["missing_count"] == 1 assert r["complete"] is False def test_it_is_explicit_not_inferred(self): """★★「엔티티에 없으면 승격됐겠지」로 유추하면 **얽혀서 승격 안 된 것**까지 통과시킨다. 명시로 받는다.""" r = self._report(promoted_ids=["rs_다른것"]) assert r["missing_count"] == 1, "유추로 통과시켰다" @pytest.mark.parametrize("bad", [None, [], ["", " "]]) def test_an_empty_promotion_list_changes_nothing(self, bad): assert self._report(promoted_ids=bad)["missing_count"] == 1 def test_the_plan_step_passes_the_ledger_ids(self): """★★호출부가 안 넘기면 이 규칙은 **아무 데도 안 닿는다**. 조건을 함수로 빼고 조립을 호출부에 남기면, 시험은 「무엇을 할지」만 알고 「했는지」는 모른다 — 이 판에서 이미 겪은 부류다. """ import inspect from app.core.steps import grounding_steps src = inspect.getsource(grounding_steps.GroundingPlanStep) assert "promoted_ids=" in src, "plan step 이 승격 목록을 안 넘긴다" assert "candidate_ledger" in src class TestTheRealTwentyThreeCandidatesAreBurnedNotJustClaimed: """★★「23행을 잰다」를 **손 측정이 아니라 자동 끝점**으로 만든다. 앞서 보고에 「디스크에서 23행을 시험했다」고 썼는데, 실제로 도는 사슬 시험의 fixture 는 **4행**이었다 — 23은 한 번 눈으로 본 수였다. 보고가 코드보다 셌다 (Codex). 그래서 **유료로 산 그 23개**를 추적되는 fixture 로 두고 태운다. ★`artifact/` 는 gitignore 라 거기 두면 시험이 못 읽는다. 원본은 Opik trace `06a93c25-…` 이고, 이 파일은 그것을 복구해 **entity_merge 까지 함께** 담은 self-contained 판이다. """ _FX = (pathlib.Path(__file__).resolve().parents[1] / "fixtures" / "grounding" / "a0_recovered_97375a4b.json") @pytest.fixture def real(self): fx = json.loads(self._FX.read_text(encoding="utf-8")) from app.modules.pipeline.grounding_subject import build_subject kept = [] for c in fx["a0_candidates"]: kept.append({**c, **build_subject( project_id="da049582", episode_id="97375a4b", source_anchor=c["source_anchor"], surface_form=c["surface_form"], owner_type=c["owner_type"], provenance={"source_step": "grounding_a0"}), "source_quote": c["source_quote"]}) return fx, kept def test_all_twenty_three_land_somewhere(self, real): fx, kept = real assert len(kept) == 23 out = build_subjects(fx["entity_merge"], project_id="da049582", episode_id="97375a4b", source_step="entity_merge", a0_candidates=kept) led = out["candidate_ledger"] assert led["total"] == 23 assert sum(led["by_disposition"].values()) == 23 assert {r["research_subject_id"] for r in led["rows"]} == \ {c["research_subject_id"] for c in kept} def test_the_measured_split_is_locked(self, real): """★실측한 갈래를 못박는다 — 바뀌면 **왜 바뀌었는지** 말해야 한다.""" fx, kept = real led = build_subjects(fx["entity_merge"], project_id="da049582", episode_id="97375a4b", source_step="entity_merge", a0_candidates=kept)["candidate_ledger"] assert led["by_disposition"] == { "carried": 2, "promoted": 7, "deferred": 14, "contested": 0, "unresolved": 0, "entity_only": 0} assert led["deferred_owners"] == {"location_part": 10, "outlook": 4} def test_the_manuscript_quotes_survive_into_the_subjects(self, real): """★★승격의 몫이 바로 이것이다 — 원문 인용이 파이프라인에 들어간다. 고치기 전에는 원고 기반 subject 가 **2개**였다. """ fx, kept = real out = build_subjects(fx["entity_merge"], project_id="da049582", episode_id="97375a4b", source_step="entity_merge", a0_candidates=kept) by = {} for s in out["subjects"]: by[s.get("quote_source")] = by.get(s.get("quote_source"), 0) + 1 assert by == {"manuscript": 9, "entity_description": 5} # ★원문 인용이 **비어 있지 않다** — 「manuscript」 표만 붙으면 소용없다 assert all(s["source_quote"].strip() for s in out["subjects"] if s.get("quote_source") == "manuscript") def test_the_promoted_ones_clear_the_completeness_gate(self, real): """★★끝점 — 승격이 완전성 게이트까지 닿는지 **실물 23개로** 잰다.""" from app.modules.pipeline.grounding_overlay import completeness_report fx, kept = real out = build_subjects(fx["entity_merge"], project_id="da049582", episode_id="97375a4b", source_step="entity_merge", a0_candidates=kept) promoted = [r["research_subject_id"] for r in out["candidate_ledger"]["rows"] if r["disposition"] == "promoted"] e = fx["entity_merge"] r = completeness_report(kept, { "character": e["characters"], "location": e["locations"], "prop": e["props"]}, promoted_ids=promoted) assert r["missing_count"] == 0 and r["complete"] is True # ★그래도 「추출이 지웠다」는 수는 남는다 assert r["entity_missing_count"] == 7 assert r["promoted_count"] == 7 def test_the_fixture_names_where_it_came_from(self): """★어디서 왔는지 없는 fixture 는 되짚을 수 없다.""" fx = json.loads(self._FX.read_text(encoding="utf-8")) assert fx["opik_trace"] and fx["a0_pack_version"] and fx["model"] assert fx["manuscript_chars"] == 1352, "1.3KB 검증 원고다"