"""GROUNDING-V2 — A0 후보를 **병합된 엔티티에 결속**하는 한 자리. ## 왜 별도 모듈인가 같은 일을 두 곳에서 한다 — production `GroundingPlanStep` 이 `entity_merge` 산출에 결속 측정 `grounding_shadow` 가 저장된 `entity_merge` CP 에 결속 ★**둘이 갈리면 잰 것이 뜻을 잃는다.** 실제로 갈려 있었다: production 은 A0 가 건진 **원문 문장**을 분류기에 넘기는데, 측정 도구는 LLM 이 상상해 쓴 `description` 을 넘겼다. 같은 대상을 다른 근거로 판정해 놓고 「production 을 쟀다」고 쓰게 된다. 그래서 결속을 **한 함수**로 모은다. 도구가 프로덕션과 다른 입력을 보내려면 이 파일을 고쳐야 하고, 고치면 양쪽이 같이 바뀐다. """ from __future__ import annotations import copy import logging from typing import Any, Dict, List, Optional, Sequence, Tuple from app.modules.pipeline.grounding_entity_contract import (FACET_BINDING, HOST_CONTEXT, PRODUCER_PAYLOAD) from app.modules.pipeline.grounding_overlay import _norm from app.modules.pipeline.grounding_subject import build_subject logger = logging.getLogger(__name__) #: ★결속 규칙이 바뀌면 올린다 — 지문에 접혀 resume 이 옛 CP 를 안 건너뛴다. #: 2 = owner 묶음(`{prop, location_part}`) + `contested` 갈래. #: 3 = **승격**(base owner 인데 엔티티가 없으면 A0 신원 그대로 subject) + #: 후보 쪽 장부(다섯 갈래) + 얽힌 후보는 승격 안 함. #: ★안 올리면 옛 `grounding_plan` CP 를 그대로 재사용해 **승격이 아무 데도 #: 안 닿는다** — 고친 것이 안 도는 부류다 (Codex). CARRY_CONTRACT_VERSION = 3 #: 결속을 못 한 사유. ``none`` 만 「새로 발급해도 되는 것」이다. CARRY_MATCHED = "matched" CARRY_NONE = "none" CARRY_AMBIGUOUS = "ambiguous" CARRY_DUPLICATE = "duplicate_surface" #: ★한 후보를 **두 엔티티가 가져간다** — 어느 쪽 것인지 못 정한 것이다. CARRY_CONTESTED = "contested" #: ★★**후보 쪽 장부**의 갈래. `carry_reasons` 는 **엔티티**를 세지 후보를 안 #: 센다 — 그래서 어느 엔티티에도 안 걸린 후보는 **아무 데도 안 세어지고 #: 조용히 사라졌다**. 실측: 후보 23개 중 장부에 나타난 것이 2개뿐이었다. DISP_CARRIED = "carried" # 기존 엔티티에 붙었다 DISP_PROMOTED = "promoted" # ★base owner 인데 엔티티가 없다 — **그대로 승격** DISP_DEFERRED = "deferred" # 뒤 facet producer 가 가질 것이다 DISP_CONTESTED = "contested" # 두 엔티티가 가져갔다 — 못 정했다 DISP_UNRESOLVED = "unresolved" # 그 밖의 미확정 #: ★★**A0 후보가 없는 기존 엔티티 행.** 후보 장부만으로는 이것을 표현 못 한다 #: — 그래서 `GroundingScreenStep` 이 그 행을 `unbound` 로 떨어뜨렸다 #: (Codex BLOCK 2026-09-01: 「등록 LP + A0=[] → binding='unbound'」). #: ★`carried` 로 쓰면 거짓말이다 — 물려받은 후보가 **없다**. DISP_ENTITY_ONLY = "entity_only" DISPOSITIONS = (DISP_CARRIED, DISP_PROMOTED, DISP_DEFERRED, DISP_CONTESTED, DISP_UNRESOLVED, DISP_ENTITY_ONLY) #: ★**뒤 producer 가 가질 갈래.** `location_part` 는 §2-6.5 의 producer 가, #: `outlook` 은 아웃룩 단계가 가진다. 안 붙었다고 승격하면 그 producer 가 #: 만들 때 같은 대상이 둘이 된다. FACET_OWNERS = frozenset({"location_part", "outlook"}) #: 새로 발급하면 **안 되는** 사유 — 후보는 있는데 어느 것인지 못 정한 것. UNBOUND_REASONS = frozenset({CARRY_AMBIGUOUS, CARRY_DUPLICATE, CARRY_CONTESTED}) #: ★결속에서 **같은 대상일 수 있는** owner 묶음. #: #: A0 의 `owner_type` 은 **잠정**이다(계획 §6 — provisional owner). 파이프라인의 #: 엔티티 갈래와 늘 같지 않다. 실측: A0 가 「요금통」을 `location_part`(쇠사슬로 #: 묶여 못 옮긴다)로 적었는데 파이프라인은 `P01` 을 **prop** 으로 저장했다. #: 같은 원고의 다른 화에서는 A0 가 그것을 `prop` 으로 적었다 — 경계에 걸친 #: 대상이라 화마다 갈린다. #: #: ★owner 검사가 막으려던 것은 **사람과 그 사람이 입은 것**이다. `prop` 과 #: `location_part` 는 그 혼동이 아니다 — `location_part` producer 가 §2-6.5 라 #: 오늘은 그런 대상이 prop 으로 저장된다. #: #: ★묶는다고 승격하는 것이 아니다. overlay 는 여전히 `location_part` 를 #: base 갈래로 안 올린다(`grounding_overlay._OWNER_TO_ENTITY`). #: ★★★`prop` 과 `location_part` 를 **갈랐다** (Codex BLOCK 2026-09-01). #: 묶여 있던 동안은 LP 행이 스캔 대상이 아니어서 표가 안 났는데, §2-6.5a 로 #: LP 행이 생기자 **양방향 교차 결속**이 열렸다 — 실측: #: 후보 owner=prop · 이름이 LP 행과 같음 → **LP 행에 붙었다** #: 후보 owner=location_part · 이름이 prop 행과 같음 → **prop 행에 붙었다** #: 즉 다른 부분 대상의 원문 근거·참조 의무가 조용히 건너갔다. BINDABLE_OWNER_GROUPS = ( frozenset({"character"}), frozenset({"outlook"}), frozenset({"location"}), frozenset({"prop"}), frozenset({"location_part"}), ) def bindable_owners(owner_type: str) -> frozenset: """그 owner 와 **같은 대상일 수 있는** owner 들. 못 찾으면 자기 자신만.""" for g in BINDABLE_OWNER_GROUPS: if owner_type in g: return g return frozenset({owner_type}) #: 엔티티 갈래 → owner_type. #: 엔티티 CP 키 → owner. ★**결속을 시도할 갈래**다 — 등록 여부와 다른 축이다. #: ★★§2-6.5a 로 `location_parts` 가 들어왔다 (Codex 2026-09-01): 「이미 등록된 #: LP 행은 **독립 엔티티·고증 대상으로 스캔**하고, 행이 없는 LP 후보는 부모 #: 관계 없이 일반 승격시키지 않는다.」 앞서는 이 표에 LP 칸이 없어 **결속을 #: 시도조차 못 했다** — 실물 23개에서 LP 후보 10개가 그렇게 사라졌다. #: ★`outlooks` 는 **여기 없다.** outlook 엔티티는 `outlook_phase3`(order 19.2) #: 뒤에야 생기는데 이 앞단은 13.65~13.8 이라, 그때는 **존재하지 않는다**. ENTITY_KEY_TO_OWNER = ( ("characters", "character"), ("locations", "location"), ("props", "prop"), ("location_parts", "location_part"), ) #: ★엔티티 쪽에서 **실제로 훑는** owner. `BINDABLE_OWNER_GROUPS` 는 `outlook` 을 #: 결속 가능하다고 적어 두었는데 여기에는 `outlooks` 갈래가 **없다** — 그래서 #: outlook 후보는 결속을 시도조차 못 하면서 「못 붙었다」로도 안 세어졌다. #: 같은 규칙을 두 곳에 적어 한쪽만 고쳐진 부류다. 지금은 **없다는 사실 자체를 #: 코드가 알고** `deferred` 로 센다. SCANNED_OWNERS = frozenset(o for _k, o in ENTITY_KEY_TO_OWNER) def disposition_of(owner_type: str) -> str: """**안 붙은** 후보가 어디로 가는가. ``` facet owner (location_part · outlook) → deferred 뒤 producer 가 가진다 base owner (character·location·prop) → promoted A0 신원 그대로 승격 그 밖 → unresolved ``` ★`location_part` 는 `prop` 과 같은 묶음이라 **결속은 시도한다**. 그래도 안 붙으면 `deferred` 다 — 승격하면 §2-6.5 producer 가 만들 때 같은 대상이 둘이 된다. """ o = (owner_type or "").strip() if o in FACET_OWNERS: return DISP_DEFERRED if o in SCANNED_OWNERS: return DISP_PROMOTED return DISP_UNRESOLVED def build_carry_index(candidates: Sequence[Dict[str, Any]]) -> Dict[str, Any]: """owner 별 표면형 색인. ★같은 표면형이 둘이면 **둘 다 뺀다.** A0 는 「같은 대상을 여러 번 적지 마세요」를 지시받지만 지켜진다는 보장이 없다. 겹친 것을 그냥 두면 아래 짝짓기가 **아무 쪽에나** 붙는다. ★같은 key 를 **덮지 않는다.** 덮으면 겹친 후보 중 하나의 id·anchor·인용이 그 자리에서 사라지고, 뒤에서 「원 후보를 보존한다」가 거짓이 된다. """ index: Dict[Tuple[str, str], List[Dict[str, Any]]] = {} for c in candidates or (): key = (c.get("owner_type") or "", _norm(c.get("surface_form", ""))) if not key[0] or not key[1]: continue index.setdefault(key, []).append(c) return {"index": index, "ambiguous": {k for k, v in index.items() if len(v) > 1}} #: ★★★**이름으로는 절대 결속하지 않는 갈래.** #: 같은 owner 안에서도 「간판」과 「회전 간판」은 **서로 다른 부분**일 수 있다. #: 그래서 LP 는 exact 도 substring 도 신원 결속에 쓰지 않는다 (Codex 2026-09-01). STRUCTURAL_ONLY_OWNERS = frozenset({"location_part"}) #: producer/ledger 가 **명시로 낸** 링크 칸. ★후보가 스스로 적은 `short_id` 는 #: **충분하지 않다** — 실제 A0 산출에는 그 칸이 **아예 없고**(실측: 후보 칸이 #: `owner_type·planned_occurrences·source_anchor·source_quote·surface_form· #: why_candidate` 뿐), 임의로 채워 넣은 값은 「같은 실물」을 증명하지 못한다. #: ★★★producer 가 낸 **링크 한 덩어리**. 흩어진 선택 필드가 아니다 #: (Codex · 09-01) — 흩어 두면 어느 좌표를 믿었는지 사라지고, 저장 위치가 #: 둘이면 그 둘이 부딪혀도 한쪽만 보고 붙게 된다. #: #: {"issuer": …, "contract_version": …, "final_id": …, "local_id": …} LEDGER_LINK = "producer_link" #: 링크 안의 칸 이름. LINK_ISSUER = "issuer" LINK_CONTRACT = "contract_version" LINK_FINAL_ID = "final_id" LINK_LOCAL_ID = "local_id" class LinkConflict(RuntimeError): """링크가 **서로 다른 것**을 가리킨다. ★붙이지 않는다.""" def known_producers() -> Dict[str, frozenset]: """이 코드가 **아는** 발급자와 그 계약 판. ★여기서 손으로 안 적는다.""" from app.modules.pipeline.grounding_entity_contract import KNOWN_PRODUCERS return {k: frozenset(v) for k, v in KNOWN_PRODUCERS.items()} def link_verdict(cand: Dict[str, Any], entities: Dict[str, Sequence[Dict[str, Any]]]) -> str: """이 후보의 링크가 **아무 엔티티에도 안 맞은** 까닭. ★한 번만 정한다. ★엔티티와 **무관한** 흠(모양·발급자·계약·두 자리 충돌)이 먼저다 — 그것이 있으면 어느 엔티티와 대조해도 같은 답이라 좌표 이야기를 할 필요가 없다. ★그 다음이 좌표다: 어디에도 안 맞았으면 `coordinate_mismatch` 다. """ _l, why = verified_link(cand, short_id="", local_id="") if why in (LINK_MALFORMED, LINK_UNKNOWN_ISSUER, LINK_BAD_CONTRACT, LINK_STORAGE_CONFLICT): return why seen: List[str] = [] for key, rows in (entities or {}).items(): del key for e in rows or (): if not isinstance(e, dict): continue _l2, why2 = verified_link( cand, short_id=str(e.get("short_id") or ""), local_id=entity_local_id(e)) if why2 == LINK_OK: return LINK_OK # ★맞는 것이 있다 — 흠이 아니다 seen.append(why2) # ★대조한 것이 **전부** 「그 좌표가 없다」면 그대로 적는다 — 뭉뚱그리면 # 「대상이 그 칸을 안 가졌다」와 「값이 다르다」가 구별되지 않는다. if seen and all(w == LINK_COORD_MISSING for w in seen): return LINK_COORD_MISSING return LINK_COORD_MISMATCH def _all_candidates(carry: Dict[str, Any], owner: str) -> List[Dict[str, Any]]: """그 갈래의 후보 전부. ★사유를 모으려면 **붙은 것 밖**도 봐야 한다.""" return [c for _k, v in (carry or {}).get("index", {}).items() for c in v if str((c or {}).get("owner_type") or "") == owner] #: 링크 칸이 앉는 두 자리. LINK_AT_TOP = "top_level" LINK_AT_PROVENANCE = "provenance" #: 원형 슬롯을 담는 칸 이름. LINK_RAW_SLOTS = "link_raw_slots" def link_raw_slots(cand: Dict[str, Any]) -> List[Dict[str, Any]]: """링크 칸이 **있는 자리마다** 위치와 값을 원형 그대로. ★★★앞 판은 첫 슬롯 하나만 남겨서 두 가지를 잃었다 (Codex · 09-01) — ①값이 `None` 이면 「없음」과 구별이 안 됐다(칸은 있는데 안 남았다) ②두 자리가 다를 때 **무엇과 부딪혔는지** 두 번째 값이 사라졌다 ★칸이 **아예 없을 때만** 빈 목록이다. """ c = cand or {} out: List[Dict[str, Any]] = [] for where, holder in ((LINK_AT_TOP, c), (LINK_AT_PROVENANCE, c.get("provenance") or {})): if isinstance(holder, dict) and LEDGER_LINK in holder: out.append({"location": where, "value": copy.deepcopy(holder[LEDGER_LINK])}) return out def _link_slots(cand: Dict[str, Any]) -> List[Any]: """링크 칸이 **있는** 자리의 값들. ★두 저장 위치를 다 본다. ★★값이 dict 이 아니어도, 비어 있어도 **그대로 담는다** (Codex · 09-01). 앞 판은 `isinstance(dict) and got` 로 걸러서, 칸이 있는데 문자열·`None`· 빈 dict 이면 **「없음」으로 접혔다**. 그러면 producer 계약이 깨진 것이 평범한 「링크 없음」처럼 보이고, 더 나쁘게는 한쪽이 망가졌는데 다른 쪽이 멀쩡하면 **그것만 보고 붙었다**. """ c = cand or {} out: List[Any] = [] for holder in (c, c.get("provenance") or {}): if isinstance(holder, dict) and LEDGER_LINK in holder: out.append(holder[LEDGER_LINK]) return out def _is_malformed(value: Any) -> bool: """링크 모양이 아닌가. ★필수 칸이 없어도 malformed 다.""" if not isinstance(value, dict) or not value: return True return any(not str(value.get(k) or "").strip() for k in LINK_REQUIRED) #: 링크를 못 믿은 **사유**. ★「링크가 없다」와 「링크가 틀렸다」를 가른다 — #: 안 가르면 producer 계약이 어긋난 것이 평범한 「없음」처럼 보인다. LINK_NONE = "no_link" LINK_UNKNOWN_ISSUER = "unknown_issuer" LINK_BAD_CONTRACT = "contract_not_known" LINK_STORAGE_CONFLICT = "storage_places_disagree" #: ★칸은 **있는데** 모양이 아니다 — 문자열·`None`·빈 dict·필수 칸 없음. #: 「없음」으로 접으면 producer 계약이 깨진 것이 안 보인다 (Codex · 09-01). LINK_MALFORMED = "malformed_link" LINK_COORD_MISMATCH = "coordinate_mismatch" LINK_COORD_MISSING = "target_has_no_such_coordinate" LINK_OK = "ok" #: ★**계약이 어긋난** 사유들 — 그냥 「없음」으로 접으면 안 된다. LINK_BROKEN = (LINK_UNKNOWN_ISSUER, LINK_BAD_CONTRACT, LINK_STORAGE_CONFLICT, LINK_COORD_MISMATCH, LINK_COORD_MISSING, LINK_MALFORMED) #: 링크 dict 이 **반드시** 가져야 하는 칸. LINK_REQUIRED = (LINK_ISSUER, LINK_CONTRACT) def entity_local_id(entity: Dict[str, Any]) -> str: """엔티티 행의 `local_id`. ★adapter 는 그것을 **provenance 안**에 둔다. top-level 만 보면 늘 비어 있고, 그러면 「대상에 없는 좌표」를 검사할 수 없어 링크가 낸 엉뚱한 `local_id` 가 **그냥 지나간다** (Codex · 09-01). """ e = entity or {} got = str(e.get("local_id") or "").strip() if got: return got pv = e.get("grounding_provenance") or e.get("provenance") or {} return str((pv or {}).get("local_id") or "").strip() def verified_link(cand: Dict[str, Any], *, short_id: str, local_id: str = "") -> Tuple[Optional[Dict[str, Any]], str]: """이 후보를 그 엔티티에 붙일 **검증된 링크**와 **사유**. ★**전부 맞아야 한다** — ①발급자가 **아는 것**이고 그 발급자의 **아는 계약 판**을 달았다 ②두 저장 위치(top-level·provenance)에 있으면 **서로 같아야** 한다 ③★링크가 낸 좌표는 **대상 쪽에도 있어야 하고 같아야** 한다. 없으면 무시하지 않고 **거절**한다 — 앞 판은 대상 `local_id` 가 비면 링크의 엉뚱한 `local_id` 를 안 봤고, `final_id` 하나가 맞으면 OR 로 통과시켰다. ④적어도 한 좌표는 있어야 한다 Returns: `(링크 or None, 사유)`. 사유는 `LINK_OK` 이거나 못 믿은 까닭이다 — **「없다」와 「틀렸다」를 가른다**. """ links = _link_slots(cand) if not links: return None, LINK_NONE # ★칸 자체가 없다 if any(_is_malformed(v) for v in links): # ★한 자리라도 모양이 아니면 **다른 자리가 멀쩡해도** 안 믿는다 return None, LINK_MALFORMED first = links[0] for other in links[1:]: if other != first: return None, LINK_STORAGE_CONFLICT issuer = str(first.get(LINK_ISSUER) or "").strip() known = known_producers() if issuer not in known: return None, LINK_UNKNOWN_ISSUER if str(first.get(LINK_CONTRACT) or "").strip() not in known[issuer]: return None, LINK_BAD_CONTRACT fid = str(first.get(LINK_FINAL_ID) or "").strip() lid = str(first.get(LINK_LOCAL_ID) or "").strip() want_f, want_l = str(short_id or "").strip(), str(local_id or "").strip() for got, want in ((fid, want_f), (lid, want_l)): if not got: continue if not want: # ★대상에 그 좌표가 **아예 없다** — 검사할 수 없으면 안 믿는다 return None, LINK_COORD_MISSING if got != want: return None, LINK_COORD_MISMATCH if not (fid or lid): return None, LINK_COORD_MISSING return dict(first), LINK_OK def _hits(carry: Dict[str, Any], name: str, owner_type: str): want = _norm(name) if not want: return [] # ★★★**링크를 선언한 후보는 이름 문에 안 들어온다** (Codex BLOCK · 09-01). # 선언해 놓고 그것이 안 맞으면 **이름으로 되살아나** 붙었다 — 실측: # `issuer='bogus'` 인데 이름이 같아서 `carried` 가 되고, 같은 장부 줄에 # `disposition='carried'` 와 `link_fault='unknown_issuer'` 가 **동시에** # 적혔다. 모순이다. 링크를 걸었으면 **링크로만** 붙는다. # ★★구조 전용 갈래는 **이름 문에 아예 안 들어온다** — 양쪽 다. # 앞 판은 한쪽만 막아서, `prop` 행이 `location_part` 후보를 **이름으로** # 가져가 `contested` 를 만들었다 (실측 2026-09-01). ok = bindable_owners(owner_type) - STRUCTURAL_ONLY_OWNERS if owner_type in STRUCTURAL_ONLY_OWNERS: return [] out = [] for k, v in carry["index"].items(): if k[0] not in ok or not (k[1] in want or want in k[1]): continue # ★링크를 선언한 후보는 뺀다 — 그것은 링크로만 붙는다 kept = [c for c in v if not link_raw_slots(c)] if kept: out.append((k, kept)) return out def match_candidate(carry: Dict[str, Any], name: str, owner_type: str, *, short_id: str = "", local_id: str = ""): """살아남은 이름에 맞는 A0 후보. ★한 개로 정해질 때만 돌려준다. 가르는 규칙 셋: 1. **owner 가 같은 묶음이어야 한다**(`BINDABLE_OWNER_GROUPS`). 사람과 그 사람이 입은 것은 서로 다른 대상이라 표면형이 겹쳐도 결속하면 안 된다. `prop`↔`location_part` 는 그 혼동이 아니라 **같은 대상의 잠정 갈래**다. 2. **양방향 포함**으로 본다 — 추출이 「고무줄로 묶인 회수권 뭉치」를 「회수권 뭉치」로 줄여도 잡아야 한다. 뜻으로 묶는 것이 아니라 **내가 넣은 말이 남았는지**를 보는 것이다. 3. **둘 이상 걸리면 안 붙인다.** 「가장 긴 것을 고른다」는 임의 선택이고, 그렇게 붙인 id 는 다른 대상의 근거를 물려받는다. Returns: (후보 or None, 사유). """ # ★★★**링크가 이름보다 세다.** producer 가 낸 링크는 그 행의 사실이고, # 이름은 짐작이다. 같은 씬에 같은 표면형이 둘이면 이름 경로는 「애매하다」로 # 아무것도 안 붙이는데, 링크가 있으면 **어느 것인지 이미 정해져 있다**. # ★링크가 하나도 없으면 아래 이름 경로가 그대로 돈다(기존 동작 무변). linked = [c for _k, v in carry["index"].items() for c in v if str(c.get("owner_type") or "") == owner_type and verified_link(c, short_id=short_id, local_id=local_id)[0]] if linked: if len(linked) > 1: return None, CARRY_DUPLICATE return linked[0], CARRY_MATCHED if owner_type in STRUCTURAL_ONLY_OWNERS: # ★이름으로 짐작하지 않는다 — **producer 가 낸 링크**가 있는 것만. same = [] if not same: return None, CARRY_NONE if len(same) > 1: return None, CARRY_DUPLICATE return same[0], CARRY_MATCHED if not _norm(name): return None, CARRY_NONE hits = _hits(carry, name, owner_type) if not hits: return None, CARRY_NONE if len(hits) > 1: return None, CARRY_AMBIGUOUS _key, cands = hits[0] if len(cands) > 1: return None, CARRY_DUPLICATE return cands[0], CARRY_MATCHED def _originals(carry: Dict[str, Any], name: str, owner_type: str): """걸린 후보를 **전부** 편다 — 하나만 남기면 다른 근거가 사라진다.""" return [ {"research_subject_id": c.get("research_subject_id"), "source_anchor": c.get("source_anchor"), "surface_form": c.get("surface_form"), "source_quote": c.get("source_quote")} for _k, v in _hits(carry, name, owner_type) for c in v ] def build_subjects( entities: Dict[str, Sequence[Dict[str, Any]]], *, project_id: str, episode_id: str, source_step: str, a0_candidates: Optional[Sequence[Dict[str, Any]]] = None, ) -> Dict[str, Any]: """병합된 엔티티(`entity_merge`) → 분류할 subject 목록 + 못 붙인 목록. ★production 과 측정 도구가 **이 함수 하나**를 쓴다. Returns: ``subjects`` 분류기에 넣을 것 (A0 물려받았거나, 후보가 없던 것) ``unbound`` ★분류기에 **안 넣는다** — 후보는 있는데 못 정한 것. 원 후보의 id·anchor·원문 인용을 그대로 담은 ``unresolved`` 행이다. ``carry_reasons`` 사유별 개수 · ``carried`` 물려받은 수 """ carry = build_carry_index(list(a0_candidates or ())) subjects: List[Dict[str, Any]] = [] unbound: List[Dict[str, Any]] = [] reasons: Dict[str, int] = {} carried = 0 # ★**되돌아오는 쪽도 1:1 이어야 한다.** 한 후보가 두 엔티티에 걸리면 어느 # 쪽 것인지 못 정한 것이다. 앞 판은 그것을 「같은 subject id 라 접는다」로 # 삼켜 **뒤에 온 엔티티가 통째로 사라졌다** — 분류도 안 되고 기록도 없다. _claim: Dict[str, int] = {} for key, owner in ENTITY_KEY_TO_OWNER: for e in entities.get(key) or (): if not isinstance(e, dict) or not (e.get("name") or "").strip(): continue hit, why = match_candidate( carry, e["name"].strip(), owner, short_id=str(e.get("short_id") or ""), local_id=entity_local_id(e)) if hit: rid = hit.get("research_subject_id") or "" _claim[rid] = _claim.get(rid, 0) + 1 _contested = {r for r, n in _claim.items() if n > 1} seen_ids: set = set() #: ★후보 id → **믿은 링크**. 장부와 provenance 가 같은 것을 남긴다. used_links: Dict[str, Any] = {} #: ★후보 id → **못 믿은 사유 + 원형 링크**. 「없음」과 「틀림」을 가른다. link_faults: Dict[str, Any] = {} #: ★subject id → 처분. **한 SOT** 다 — 부르는 쪽이 owner 로 다시 추론하면 #: 같은 owner 의 「기존 행」과 「행 없는 후보」를 구별 못 한다. subject_dispositions: Dict[str, str] = {} #: ★실제로 붙은 후보의 id. 장부가 이것으로 `carried` 를 가른다. bound_ids: set = set() #: ★★**결속에 걸렸다 실패한** 후보의 id. 승격 대상이 아니다 — #: 「어느 엔티티도 안 불렀다」와 「불렸는데 못 정했다」는 다르다. entangled: set = set() for key, owner in ENTITY_KEY_TO_OWNER: for e in entities.get(key) or (): if not isinstance(e, dict): continue name = (e.get("name") or "").strip() if not name: continue sid = (e.get("short_id") or "").strip() a0, why = match_candidate(carry, name, owner, short_id=sid, local_id=entity_local_id(e)) # ★믿은 링크를 **원형 그대로** 남긴다 — 안 남기면 나중에 「무엇을 # 믿고 붙였나」를 되짚을 수 없다 (Codex · 09-01). used_link = None if a0: used_link, _why_link = verified_link( a0, short_id=sid, local_id=entity_local_id(e)) # ★★링크를 냈는데 **못 믿은** 후보의 사유를 모은다. 안 모으면 # producer 계약이 어긋난 것이 평범한 「링크 없음」처럼 보인다. if a0 and (a0.get("research_subject_id") or "") in _contested: a0, why = None, CARRY_CONTESTED reasons[why] = reasons.get(why, 0) + 1 if a0: carried += 1 bound_ids.add(a0.get("research_subject_id") or "") if used_link: used_links[a0.get("research_subject_id") or ""] = used_link subj = { **{k: v for k, v in a0.items() if k in ("contract_version", "research_subject_id", "project_id", "episode_id", "source_anchor", "surface_form", "owner_type", "canon_id", "bind_state")}, # ★★**기대 span 을 실어 보낸다** (실측 2026-09-01). # 앞 판은 이 칸이 없어 subject 에 span 이 **하나도** 안 # 실렸고, 그래서 판별 줄의 `source_evidence` 도 실경로에서 # **늘 비어 있었다**(6줄 중 0줄). 뒤(19.2 이후 중앙 획득)가 # 결속할 좌표가 사라진다. # ★없으면 칸도 안 만든다 — 지어낸 좌표는 증거가 아니다. **({"occurrences": copy.deepcopy(a0["occurrences"])} if a0.get("occurrences") else {}), # ★producer 판정을 **원형 그대로** 물려준다 **({PRODUCER_PAYLOAD: copy.deepcopy(a0[PRODUCER_PAYLOAD])} if a0.get(PRODUCER_PAYLOAD) else {}), **({FACET_BINDING: copy.deepcopy(a0[FACET_BINDING])} if a0.get(FACET_BINDING) else {}), **({HOST_CONTEXT: copy.deepcopy(a0[HOST_CONTEXT])} if a0.get(HOST_CONTEXT) else {}), "provenance": {**(a0.get("provenance") or {}), "short_id": sid or None, "carried_from": "grounding_a0", **({LEDGER_LINK: used_link} if used_link else {})}, # ★원문 문장이다 — LLM 이 쓴 description 이 아니다. "source_quote": a0.get("source_quote") or "", "quote_source": "manuscript", } elif why == CARRY_NONE: subj = build_subject( project_id=project_id, episode_id=episode_id, source_anchor=sid or f"name:{name}", surface_form=name, owner_type=owner, provenance={"short_id": sid or None, "source_step": source_step, "carry_reason": why}) subj["source_quote"] = e.get("description") or "" subj["quote_source"] = "entity_description" # ★엔티티 쪽에도 있으면 싣는다 — adapter 는 그것을 # `grounding_provenance` 안에 둔다. _occ = ((e.get("grounding_provenance") or {}) .get("occurrences")) if _occ: subj["occurrences"] = copy.deepcopy(_occ) else: # ★후보는 있는데 어느 것인지 못 정했다. 분류기에 **안 보낸다** — # 보낼 근거가 원문이 아니라 상상 묘사라 판정이 무의미하고, # 표본 N개를 그냥 사게 된다. # ★★**걸렸다 실패한 후보는 승격하지 않는다.** 두 후보가 한 # 엔티티에 걸린 것이 「서로 다른 두 대상」인지 「같은 것을 두 번 # 적은 것」인지 **기계적으로 못 가른다**. 승격하면 같은 것을 두 # 번 조사할 수 있고, 그건 돈이다. 못 정한 것은 `unresolved` 다. for _o in _originals(carry, name, owner): if _o.get("research_subject_id"): entangled.add(_o["research_subject_id"]) unbound.append({ "route": "unresolved", "route_override_reason": f"a0_carry_{why}", "sample_count": 0, "unanswered": 0, "_short_id": sid or None, "_surface_form": name, "_owner_type": owner, "research_subject_id": None, "a0_candidates": _originals(carry, name, owner), }) continue rsid = subj["research_subject_id"] if rsid in seen_ids: # ★여기 오는 것은 **새로 발급한 것끼리** 겹친 경우뿐이다 # (물려받은 것은 위에서 `contested` 로 갈라진다). 같은 범위· # 같은 anchor·같은 표면형이면 같은 대상이라 접어도 안전하다. continue seen_ids.add(rsid) # ★★★**subject 마다 처분을 여기서 낸다** (Codex BLOCK 2026-09-01). # 후보 장부만으로는 「A0 후보가 없는 기존 엔티티 행」을 표현 못 해 # `GroundingScreenStep` 이 그 행을 `unbound` 로 떨어뜨렸다. # ★`carried` 로 쓰면 거짓말이다 — 물려받은 후보가 **없다**. subject_dispositions[rsid] = (DISP_CARRIED if a0 else DISP_ENTITY_ONLY) subjects.append(subj) # ★★★**사유는 후보별 전체 대조가 끝난 뒤에 정한다** (Codex · 09-01). # 앞 판은 엔티티를 돌며 `setdefault` 로 남겨서, 링크가 **둘째** 엔티티를 # 정확히 가리켜도 첫 엔티티와의 어긋남이 먼저 적혔다 — 붙은 줄에 # `carried` 와 `link_fault` 가 같이 실렸다. 출력만 숨기면 **내부 상태가 # 여전히 틀린 것**이라, 아예 여기서 한 번 셈한다. for c in (a0_candidates or ()): rid = str((c or {}).get("research_subject_id") or "") if not rid or rid in bound_ids or not link_raw_slots(c): continue # ★붙었거나 링크를 안 걸었다 link_faults[rid] = {"reason": link_verdict(c, entities), LINK_RAW_SLOTS: link_raw_slots(c)} # ★★**두 번째 pass** — 첫 pass 는 엔티티를 돌았으므로, 어느 엔티티도 안 # 부른 후보는 아직 아무것도 안 됐다. base owner 인 것은 **그대로 승격**해 # 자기 신원으로 subject 가 된다. A0 가 원고에서 건진 대상이라 조사 대상이 # 맞고, 승격 안 하면 그 원문 인용이 파이프라인에 아예 안 들어간다. for c in (a0_candidates or ()): rid = (c or {}).get("research_subject_id") or "" if not rid or rid in bound_ids or rid in _contested or rid in seen_ids: continue if rid in entangled: continue # ★★★**링크를 걸었는데 안 맞은 후보는 승격도 안 된다** (Codex · 09-01). # 이름으로 되살아나는 문은 닫았는데 여기로 되살아나면 같은 일이다 — # 계약이 어긋난 것을 「엔티티가 없어서 온 것」으로 읽으면 안 된다. if link_raw_slots(c): continue if disposition_of((c or {}).get("owner_type") or "") != DISP_PROMOTED: continue seen_ids.add(rid) subjects.append({ **{k: v for k, v in c.items() if k in ("contract_version", "research_subject_id", "project_id", "episode_id", "source_anchor", "surface_form", "owner_type", "canon_id", "bind_state")}, **({"occurrences": copy.deepcopy(c["occurrences"])} if c.get("occurrences") else {}), **({PRODUCER_PAYLOAD: copy.deepcopy(c[PRODUCER_PAYLOAD])} if c.get(PRODUCER_PAYLOAD) else {}), **({HOST_CONTEXT: copy.deepcopy(c[HOST_CONTEXT])} if c.get(HOST_CONTEXT) else {}), **({FACET_BINDING: copy.deepcopy(c[FACET_BINDING])} if c.get(FACET_BINDING) else {}), "provenance": {**(c.get("provenance") or {}), # ★엔티티가 없어서 온 것이다 — short_id 가 없다. "short_id": None, "promoted_from": "grounding_a0"}, # ★원문 문장이다. 승격의 몫이 바로 이것을 살리는 것이다. "source_quote": c.get("source_quote") or "", "quote_source": "manuscript", }) ledger = _candidate_ledger(list(a0_candidates or ()), carry, entities, bound_ids, contested=_contested, entangled=entangled, used_links=used_links, link_faults=link_faults) logger.info("grounding carry: 후보 %d개 중 %d개 결속 · 못 붙임 %d · %s · 장부 %s", len(a0_candidates or ()), carried, len(unbound), reasons, ledger["by_disposition"]) return {"subjects": subjects, "unbound": unbound, "subject_dispositions": subject_dispositions, "carry_reasons": reasons, "carried": carried, # ★★후보 **하나하나**가 어디로 갔는지. `carry_reasons` 는 엔티티를 # 세지 후보를 안 센다 — 그래서 어느 엔티티에도 안 걸린 후보는 # 아무 데도 안 세어지고 조용히 사라졌다. "candidate_ledger": ledger} def _candidate_ledger(cands: List[Dict[str, Any]], carry: Dict[str, Any], entities: Dict[str, Sequence[Dict[str, Any]]], bound_ids: set, *, contested: set, entangled: set, used_links: Optional[Dict[str, Any]] = None, link_faults: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """A0 후보 **전부**가 넷 중 하나로 간다. ★남거나 겹치면 **선다**. 「소리 없이 사라지는 후보 0건」이 이 함수의 통과 조건이다 (Codex). 합이 안 맞는데 그냥 돌려주면, 그 장부는 있으나 마나다. """ rows: List[Dict[str, Any]] = [] for c in cands: rid = (c or {}).get("research_subject_id") or "" owner = ((c or {}).get("owner_type") or "").strip() slots = link_raw_slots(c) if rid and rid in bound_ids: d = DISP_CARRIED elif rid and slots: # ★★링크를 **걸었는데 아무 엔티티에도 안 맞았다.** 이름으로 # 되살리지도, 승격시키지도 않는다 — 계약이 어긋난 것이다. d = DISP_UNRESOLVED elif rid and rid in contested: d = DISP_CONTESTED elif rid and rid in entangled: # ★불렸는데 못 정했다 — 안 불린 것과 다르다. 승격 안 한다. d = DISP_UNRESOLVED else: # ★안 붙은 후보의 행선지는 **owner 가 정한다** — facet 은 뒤 # producer 가 가지고, base 는 A0 신원 그대로 승격한다. d = disposition_of(owner) row = {"research_subject_id": rid, "owner_type": owner, "surface_form": (c or {}).get("surface_form") or "", "disposition": d} if d == DISP_UNRESOLVED and slots: # ★원형과 사유를 **여기서도** 남긴다 — 링크를 걸었다는 사실 자체가 # 기록돼야 계약 drift 가 보인다. row[LINK_RAW_SLOTS] = copy.deepcopy(slots) # ★★믿은 링크를 **원형 그대로** 남긴다 (Codex · 09-01). 처분만 남기면 # 「무엇을 믿고 붙였나」가 사라진다. link = (used_links or {}).get(rid) if link: row[LEDGER_LINK] = dict(link) # ★★못 믿은 링크는 **사유와 원형**을 남긴다 — 「없음」과 갈린다. # ★★★단 **안 붙은 것에만** 붙인다 (실측 2026-09-01): 링크가 두 번째 # 엔티티를 정확히 가리켜도, 첫 엔티티와 대조할 때 「어긋남」이 한 번 # 기록되면 그 사유가 **붙은 줄에 그대로 실렸다** — `carried` 와 # `link_fault` 가 한 줄에 같이 있는 그 모순이다. # ★사유는 이미 **후보별로 한 번** 정해졌다 — 붙은 후보에는 애초에 # 들어오지 않는다. 여기서 다시 가릴 필요가 없다. fault = (link_faults or {}).get(rid) if fault: row["link_fault"] = str(fault.get("reason") or "") slots = fault.get(LINK_RAW_SLOTS) or [] if slots: # ★★모양이 아니어도, `None` 이어도, 두 자리가 달라도 **전부** # 위치째 남긴다 — 무엇이 왔길래 거절했는지 못 보면 계약 # drift 를 못 고친다. row[LINK_RAW_SLOTS] = copy.deepcopy(slots) rows.append(row) counts = {d: sum(1 for r in rows if r["disposition"] == d) for d in DISPOSITIONS} # ★★**합만 세면 항진식이다** (Codex). 후보마다 행 하나를 무조건 넣고 그 # 행 수를 다시 세므로 합은 **언제나** 맞는다 — 빈 id 도 중복 id 도 못 잡는다. # 실제로 같은 `research_subject_id` 두 후보가 장부는 2행인데 subject 는 # 1개였고(뒤엣것이 `seen_ids` 에서 사라짐) 아무 소리도 안 났다. # 그래서 **id 집합으로** 잰다. got = [r["research_subject_id"] for r in rows] want = [str((c or {}).get("research_subject_id") or "") for c in cands] blank = [i for i, v in enumerate(want) if not v.strip()] if blank: raise AssertionError( f"후보에 `research_subject_id` 가 빈 것이 {len(blank)}개 있다 " f"(자리 {blank[:5]}) — 장부가 어느 후보인지 못 적는다") dup = sorted({v for v in want if want.count(v) > 1}) if dup: raise AssertionError( f"후보 id 가 겹친다: {dup[:5]} — 겹치면 뒤엣것이 조용히 사라진다") if set(got) != set(want) or len(got) != len(want): raise AssertionError( f"장부가 후보 집합과 다르다: 장부 {len(got)}행 / 후보 {len(want)}개 " f"· 장부에만 {sorted(set(got) - set(want))[:5]} " f"· 후보에만 {sorted(set(want) - set(got))[:5]}") bad = [r for r in rows if r["disposition"] not in DISPOSITIONS] if bad: raise AssertionError( f"갈래 밖 행 {len(bad)}개: {[r['disposition'] for r in bad][:5]}") if sum(counts.values()) != len(rows): raise AssertionError( f"갈래별 합이 행 수와 다르다: {sum(counts.values())} != {len(rows)}") return { "total": len(cands), "by_disposition": counts, # ★`deferred` 는 **왜** 미뤄졌는지를 owner 별로 남긴다 — 그 producer 가 # 생기면 여기가 0이 되어야 한다. "deferred_owners": { o: sum(1 for r in rows if r["disposition"] == DISP_DEFERRED and r["owner_type"] == o) for o in sorted({r["owner_type"] for r in rows if r["disposition"] == DISP_DEFERRED})}, "rows": rows, }