"""참조 의무는 **조사 앞에서** 선다. ★유료 0. Codex BLOCK (2026-09-01) — 「bundle projection 이 acquisition 완료 뒤 context 멤버를 발명하면 안 됩니다. 중앙 참조 조사 **전에** reference obligation plan 을 만드십시오. 각 obligation 은 서로 **다른 outbound/acquisition identity** 를 가져야 합니다.」 """ from __future__ import annotations import pytest from app.modules.pipeline import grounding_acquisition_adapter as aa from app.modules.pipeline import grounding_acquisition_ledger as gl from app.modules.pipeline import grounding_reference_obligations as ro from app.modules.pipeline import grounding_screen as gs from app.modules.pipeline.grounding_entity_contract import (HOST_CONTEXT, PRODUCER_PAYLOAD) from app.modules.pipeline.grounding_host_context import (PURPOSE_CONTEXT, PURPOSE_DETAIL) EV = {"source_quote": "원문", "occurrences": [{"source_span": {"segment_id": "scene-1"}}]} def _pl(word): return {"coarse_type_label": word, "visual_brief": f"{word} 찾기", "search_terms_native": [word], "language_lock_native": "ko"} def _row(fid, owner, *, parent=None, screen=gs.SCREEN_OBLIGATION, word="가"): return gl.row(owner_type=owner, research_subject_id=f"rs_{fid}", screen=screen, status=gl.RESOLVED, final_id=fid, parent_final_id=parent, source_evidence=dict(EV), producer_payload=_pl(word)) def _led(*rows): return {"rows": list(rows)} def _led_of(row): return {"rows": [row]} class TestEachObligationAsksSomethingDifferent: """★★쓰임만 다시 적은 것이 아니라 **다른 것을 조사한다**.""" LED = _led(_row("L01", "location", word="이발소"), _row("LP01", "location_part", parent="L01", word="회전 간판")) def test_context_and_detail_are_separate_obligations(self): got = ro.plan(self.LED) by = ro.purposes_of(got) assert by["LP01"] == [PURPOSE_DETAIL] assert by["L01"] == [PURPOSE_CONTEXT] def test_they_have_different_subject_ids(self): ids = {r["research_subject_id"] for r in ro.plan(self.LED)["rows"]} assert "L01#context" in ids and "LP01#detail" in ids def test_they_carry_different_queries(self): """★같은 질의면 **같은 것을 두 번 조사한다**.""" got = aa.inputs_from_ledger(ro.plan(self.LED)) q = {t["target"]["subject_id"]: t["target"]["directive_native"] for t in got["targets"]} assert q["L01#context"] != q["LP01#detail"] assert "이발소" in q["L01#context"] assert "회전 간판" in q["LP01#detail"] def test_the_identities_differ(self): from app.modules.pipeline import grounding_central_acquisition as ca got = aa.inputs_from_ledger(ro.plan(self.LED)) ids = {ca.identity_of(t, contract_sha="s") for t in got["targets"]} assert len(ids) == len(got["targets"]) class TestSiblingsShareOneContextObligation: LED = _led(_row("L01", "location", word="이발소"), _row("LP01", "location_part", parent="L01", word="회전 간판"), _row("LP02", "location_part", parent="L01", word="미닫이문")) def test_only_one_context_obligation_stands(self): ctx = [r for r in ro.plan(self.LED)["rows"] if r.get("purpose") == PURPOSE_CONTEXT] assert len(ctx) == 1 assert ctx[0]["final_id"] == "L01" def test_it_covers_every_sibling(self): ctx = [r for r in ro.plan(self.LED)["rows"] if r.get("purpose") == PURPOSE_CONTEXT][0] assert ctx["covers"] == ["L01", "LP01", "LP02"] def test_each_sibling_keeps_its_own_detail(self): det = [r for r in ro.plan(self.LED)["rows"] if r.get("purpose") == PURPOSE_DETAIL] assert sorted(r["final_id"] for r in det) == ["LP01", "LP02"] def test_the_parent_is_researched_once(self): got = aa.inputs_from_ledger(ro.plan(self.LED)) subs = [t["target"]["subject_id"] for t in got["targets"]] assert subs.count("L01#context") == 1 assert len(subs) == 3, "★맥락 1 + 상세 2 여야 한다" class TestTheParentScreenIsNotAskedAgain: """★★부모 장소가 **비대상이어도** LP 의무에서 결정적으로 파생한다.""" LED = _led(_row("L01", "location", screen=gs.SCREEN_NOT_TARGET, word="이발소"), _row("LP01", "location_part", parent="L01", word="회전 간판")) def test_the_context_obligation_still_stands(self): ctx = [r for r in ro.plan(self.LED)["rows"] if r.get("purpose") == PURPOSE_CONTEXT] assert len(ctx) == 1 and ctx[0]["final_id"] == "L01" assert ctx[0]["screen"] == gs.SCREEN_OBLIGATION def test_it_uses_the_parents_own_search_material(self): ctx = [r for r in ro.plan(self.LED)["rows"] if r.get("purpose") == PURPOSE_CONTEXT][0] assert ctx[PRODUCER_PAYLOAD]["coarse_type_label"] == "이발소" def test_the_parent_row_itself_is_still_recorded(self): """★비대상 줄이 사라지면 회계가 깨진다.""" ids = {r["research_subject_id"] for r in ro.plan(self.LED)["rows"]} assert "rs_L01" in ids class TestWithNoParentItInventsNoPlaceId: """★★부모 행이 없어도 **맥락과 상세를 둘 다** 조사한다 (Codex · 09-01). 앞 판은 재료가 없다며 맥락을 통째로 건너뛰었다 — 안전했지만 「장소와 부분을 둘 다 찾는다」는 기능을 닫은 것이었다. 이제 모델이 낸 `host_context` 가 그 재료다. """ #: ★★`evidence_spans` 는 **`grounding_chunk_merge` 가 그 행의 씬에서 #: 찾아 붙인 것**이다 — 여기서 짓는 것은 그 산출 모양을 흉내 낸 것이고, #: 진짜 검증은 `test_five_lanes_to_the_bundle` 이 실제 사슬로 잰다. QUOTE = "원문이 그 장소를 말한 구절" HC = {"state": "explicit_context_only", "parent_local_id": None, "search_subject": "그 장소", "evidence": [QUOTE], "evidence_spans": [{"segment_id": "scene-1", "start": 0, "end": len(QUOTE), "source_quote": QUOTE}]} def _led(self, hc=None): r = _row("LP09", "location_part", word="어떤 부분") if hc is not None: r[HOST_CONTEXT] = hc return _led_of(r) def test_no_place_id_is_conjured(self): ids = {r["final_id"] for r in ro.plan(self._led(self.HC))["rows"]} assert not [x for x in ids if str(x).startswith("L") and not str(x).startswith("LP")] def test_the_part_owns_its_own_context(self): own = ro.context_owner_of(ro.plan(self._led(self.HC))) assert own == {"LP09": ro.OWNER_SELF} def test_both_targets_actually_go_out(self): """★★양성 대조 — 부모 행 없는 LP 하나가 **두 대상**을 낸다.""" got = aa.inputs_from_ledger(ro.plan(self._led(self.HC))) subs = sorted(t["target"]["subject_id"] for t in got["targets"]) assert subs == ["LP09#context", "LP09#detail"] assert got["skipped"] == [] def test_they_ask_different_things(self): got = aa.inputs_from_ledger(ro.plan(self._led(self.HC))) q = {t["target"]["subject_id"]: t["target"]["directive_native"] for t in got["targets"]} assert q["LP09#context"] != q["LP09#detail"] assert q["LP09#context"] == self.HC["search_subject"] # ★검증된 좌표가 증거로 남는다 — 글자만 남기지 않는다 ctx = [r for r in ro.plan(self._led(self.HC))["rows"] if r.get("purpose") == PURPOSE_CONTEXT][0] occ = ctx["source_evidence"]["occurrences"] assert occ[0]["source_span"]["segment_id"] == "scene-1" assert occ[0]["source_quote"] == self.QUOTE assert "어떤 부분" in q["LP09#detail"] def test_they_have_different_identities(self): from app.modules.pipeline import grounding_central_acquisition as ca got = aa.inputs_from_ledger(ro.plan(self._led(self.HC))) ids = {ca.identity_of(t, contract_sha="s") for t in got["targets"]} assert len(ids) == 2 def test_the_context_keeps_the_models_own_evidence(self): ctx = [r for r in ro.plan(self._led(self.HC))["rows"] if r.get("purpose") == PURPOSE_CONTEXT][0] assert ctx["source_evidence"]["source_quote"] == self.HC["evidence"][0] @pytest.mark.parametrize("hc", [ None, {"state": "unresolved", "parent_local_id": None, "search_subject": None, "evidence": []}, {"state": "explicit_context_only", "parent_local_id": None, "search_subject": "", "evidence": ["구절"]}, {"state": "explicit_context_only", "parent_local_id": None, "search_subject": "장소", "evidence": []}, # ★★**검증된 좌표가 없으면 안 나간다** — 글자만 있는 것은 # 「원문 어딘가에 있다」밖에 못 말한다 (Codex BLOCK · 09-01) {"state": "explicit_context_only", "parent_local_id": None, "search_subject": "장소", "evidence": ["구절"]}, {"state": "explicit_context_only", "parent_local_id": None, "search_subject": "장소", "evidence": ["구절"], "evidence_spans": []}, ]) def test_without_material_it_makes_nothing_up(self, hc): """★음성 대조 — 모델이 못 정했으면 **지어내지 않는다**.""" plan = ro.plan(self._led(hc)) ctx = [r for r in plan["rows"] if r.get("purpose") == PURPOSE_CONTEXT] assert len(ctx) == 1 and ctx[0].get("why") == ro.WHY_NO_CONTEXT_SOURCE assert PRODUCER_PAYLOAD not in ctx[0] got = aa.inputs_from_ledger(plan) assert [s["research_subject_id"] for s in got["skipped"]] == \ ["LP09#context"] def test_it_never_reuses_the_parts_own_brief(self): """★★그 부분의 지문을 맥락 질의로 쓰면 **같은 것을 두 번 조사한다**.""" got = aa.inputs_from_ledger(ro.plan(self._led(self.HC))) q = {t["target"]["subject_id"]: t["target"]["directive_native"] for t in got["targets"]} assert q["LP09#context"] != q["LP09#detail"] class TestNothingIsLostOrDoubled: LED = _led(_row("C01", "character", word="이발사"), _row("P01", "prop", screen=gs.SCREEN_NOT_TARGET, word="가위"), _row("L01", "location", word="이발소"), _row("LP01", "location_part", parent="L01", word="회전 간판")) def test_every_lane_survives(self): """★조사 대상인 장소는 **제 맥락 의무가 된다** — 따로 지나가지 않는다. 인물 · 비대상 소품 · `LP01#detail` · `L01#context` = 넷. """ rows = ro.plan(self.LED)["rows"] assert len(rows) == 4 assert gl.accounting({"rows": rows})["rows"] == 4 assert {r["final_id"] for r in rows} == {"C01", "P01", "L01", "LP01"} def test_non_background_lanes_get_no_purpose(self): rows = ro.plan(self.LED)["rows"] assert {r["research_subject_id"] for r in rows if r.get("purpose") is None} == {"rs_C01", "rs_P01"} def test_the_ids_are_unique(self): rows = ro.plan(self.LED)["rows"] ids = [r["research_subject_id"] for r in rows] assert len(ids) == len(set(ids)) class TestTheContractRatchetMoved: """★★모양이 바뀌면 **판이 움직여야** 옛것이 조용히 섞이지 않는다. Codex (2026-09-01) — 「screen config hash 가 producer 판으로 움직이는지, 같은 raw 의 processing stamp 가 움직이는지, obligation 판 변경이 D 의 producer-step config/sidecar 신원을 움직이는지를 양성 대조로 잠그십시오.」 """ def test_the_producer_version_reaches_the_screen_config_hash(self, monkeypatch): import inspect from app.core.steps import grounding_chunk_step as step from app.modules.pipeline import grounding_chunk_adapter as ad src = inspect.getsource(step.GroundingChunkStep._config_hash) assert "ADAPTER_CONTRACT_VERSION" in src, \ "★producer 판이 config hash 에 안 들어간다" assert ad.ADAPTER_CONTRACT_VERSION == \ __import__("app.modules.pipeline.grounding_entity_contract", fromlist=["x"]).PRODUCER_CONTRACT_VERSION def test_the_processing_stamp_moved(self): """★같은 raw 를 새 코드로 읽은 산출이 옛것과 구별돼야 한다.""" from app.modules.pipeline import grounding_chunk as gc assert not gc.PROCESSING_CONTRACT_VERSION.startswith("2."), \ "★해석이 바뀌었는데 판이 그대로다" def test_the_obligation_version_moved(self): """★부모 없는 맥락이 `skipped` 에서 실제 대상으로 바뀌었다.""" assert not ro.OBLIGATION_CONTRACT_VERSION.startswith("1.") def test_the_obligation_version_rides_with_the_plan(self): """★그 판이 산출에 실려야 뒤가 어느 계약으로 만든 것인지 안다.""" got = ro.plan(_led(_row("L01", "location"))) assert got["contract_version"] == ro.OBLIGATION_CONTRACT_VERSION def test_the_acquisition_identity_folds_only_what_goes_out(self): """★음성 대조 — 해석·의무 판은 **구매 신원에 안 들어간다**. 들어가면 후처리만 고쳐도 유료 raw 를 다시 산다 (Codex 08-31 계약). """ import ast import inspect from app.modules.pipeline import grounding_central_acquisition as ca tree = ast.parse(inspect.getsource(ca.identity_of)) names = {ast.unparse(n) for n in ast.walk(tree)} for banned in ("PROCESSING_CONTRACT_VERSION", "OBLIGATION_CONTRACT_VERSION", "PRODUCER_CONTRACT_VERSION"): assert not [x for x in names if banned in x], f"★{banned}"