"""C(c) — **병렬 구간**의 행을 하나로 줄이는 계약. ★아직 아무 데도 안 붙는다. Codex 가 (c) 를 「확장 merge **설계 후보**」로 좁히며 낸 다섯 (2026-08-31). 이 파일은 그 다섯을 **코드로** 적은 것이고, 붙이는 것은 구조가 승인된 뒤다. ## 왜 기존 `entity_merge` 로는 안 되나 지금 character/location/prop **3갈래**만 읽고 「**타입이 다른데** 같은 대상」·「각 타입 **쌍**」만 찾는다 병렬 구간의 핵심은 **같은 owner 안의 중복**(구간 1의 프롭 ↔ 구간 5의 같은 프롭)이고, C 는 `location_part`·`outlook` 까지 **5갈래**다. 지금 계약으로는 **둘 다 못 본다.** ## 다섯 계약 1. **5 owner · same-owner cross-chunk** 를 받는다. 2. 출현은 `bool` 이 아니라 **`occurrences[]`** 다. 반복 축이 세는 것은 **서로 다른 씬의 수**(`appearance_count`)지 부른 횟수가 아니다 — ★유료 주행이 이것을 바로잡았다(2026-08-31): 모델은 같은 것을 한 씬 안에서 여러 말로 부르고, 자리 수로 세면 한 번 나온 것이 반복 축으로 살아 **고증 예외 축을 영영 못 잰다**. `unique_anchor_count` 는 감사·표시용으로만 둔다. 3. `existing_id | new` 는 **순차 (b) 의 칸**이다. 병렬 (c) 는 코드가 `chunk_id + row_index` 로 **local ID 를 발급**하고, keeper 의 최종 ID 도 **결정적 규칙**으로 정한다 — 모델이 안 만든다. 4. **모든 remove 는 유효 keeper 하나가 없으면 삭제 거부**다. 「provenance 를 든 행만 보호」로는 구간 행이 실제로 사라진다. 5. `location_part` 는 parent 의 **중복이 아니라 부분**일 수 있다. `same_referent` 와 `part_of` 를 **구조로 가른다** — part 를 parent 에 합치면 그 부분이 사라진다. """ from __future__ import annotations import logging from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple logger = logging.getLogger(__name__) CHUNK_MERGE_CONTRACT_VERSION = 1 from app.modules.pipeline.grounding_entity_contract import ( ENTITY_MIN_OCCURRENCES, OWNER_PREFIX, assert_prefix_parity, owners) #: 다섯 갈래 — ★계약 모듈 한 곳에서. 그쪽이 A0 schema 를 읽는다. OWNERS = owners() #: 모델이 낼 수 있는 관계. ★`part_of` 를 `same_referent` 로 접으면 부분이 사라진다. REL_SAME = "same_referent" REL_PART_OF = "part_of" RELATIONS = (REL_SAME, REL_PART_OF) #: 예외 축의 **명시 상태**. ★bool 두 개를 따로 OR 하면 합칠 때 되돌아간다. EXC_YES = "yes" EXC_NO = "no" EXC_UNRESOLVED = "unresolved" #: 합쳐진 행이 그 상태를 들고 오는 칸. EXCEPTION_STATE = "exception_state" #: ★★**원 행 하나하나의 판정 장부.** 앞 판은 「둘 다 참인 행」만 담아서, #: 지워진 행이 `notice=True` 였다는 사실이 **결과 어디에도 없었다** (Codex). #: 이제 모든 member 의 `{local_id, hard, notice}` 를 담고, 상태는 **이 #: 장부 한 벌에서** 계산한다. EXCEPTION_LEDGER = "exception_ledger" def _exception_ledger(rows: Sequence[Dict[str, Any]]) -> List[Dict[str, Any]]: """원 행마다 한 줄. ★재병합에서도 **펴서 겹친 것을 지운다**. 「어느 구간이 뭐라 했는지 남는다」를 **결과로** 보이려면 이 장부가 있어야 한다 — 상태 문자열 하나로는 지워진 행의 판정을 못 되짚는다. """ out: List[Dict[str, Any]] = [] seen: Dict[str, Dict[str, Any]] = {} def _put(entry: Dict[str, Any]) -> None: lid = (entry or {}).get("local_id") if not isinstance(lid, str) or not lid.strip(): raise AssertionError("판정 장부 줄에 `local_id` 가 없다") for k in ("hard_to_generate", "viewers_would_notice"): v = entry.get(k) if type(v) is not bool: # ★`bool("false")` 는 참이다 — 장부에서도 막는다. raise AssertionError( f"{lid}: 장부의 {k} 가 bool 이 아니다 ({v!r})") prev = seen.get(lid) if prev is None: seen[lid] = entry out.append(dict(entry)) return # ★★같은 원 행이 **다른 판정**으로 두 번 오면 순서가 답을 정한다. # 첫 값만 남기면 AB 는 no, BA 는 yes 가 된다 — 실제로 그랬다. if (prev["hard_to_generate"] != entry["hard_to_generate"] or prev["viewers_would_notice"] != entry["viewers_would_notice"]): raise AssertionError( f"{lid}: 같은 원 행에 판정이 둘이다 " f"({prev['hard_to_generate']},{prev['viewers_would_notice']}) 대 " f"({entry['hard_to_generate']},{entry['viewers_would_notice']}) — " "순서가 답을 정하게 둘 수 없다") for r in rows or (): for e in ((r or {}).get(EXCEPTION_LEDGER) or ()): _put(dict(e)) lid = str((r or {}).get("local_id") or "") if lid and lid not in seen: _put({"local_id": lid, "hard_to_generate": _flag(r, "hard_to_generate"), "viewers_would_notice": _flag(r, "viewers_would_notice")}) return out def _exception_state(rows: Sequence[Dict[str, Any]]) -> str: """★★예외 축을 **한 helper** 로 계산한다 (Codex). 어느 한 행이 **둘 다** 참 → `yes` 한쪽만 켜진 행이 있다 → `unresolved` (「아님」이 아니다) 아무 flag 도 없다 → `no` ★`hard` 와 `notice` 를 **따로 OR 하면** (T,F) 행과 (F,T) 행이 만나 한 행의 (T,T) 가 되어 「아무도 그렇게 말한 적 없는 판정」이 선다. 합칠 때도 이 helper 가 낸 **상태를 나른다** — bool 을 안 합친다. """ # ★★**장부 한 벌에서** 계산한다 — 상태 문자열을 따로 이어 붙이면 # 지워진 행의 판정이 새어 나간다 (Codex). seen_any = False for e in _exception_ledger(rows): # ★`bool()` 로 감싸지 않는다 — 장부가 이미 bool 만 담는다. h, n = e["hard_to_generate"], e["viewers_would_notice"] if h and n: return EXC_YES if h or n: seen_any = True return EXC_UNRESOLVED if seen_any else EXC_NO #: 행 하나의 **행선지**. ★`id_contested` 를 「등록 아님」과 섞지 않는다 — #: 하나는 「규칙이 안 열었다」이고 하나는 **「누구인지 못 정했다」**다. DISP_REGISTERED = "registered" DISP_NOT_REGISTERED = "not_registered" DISP_ID_CONTESTED = "id_contested" #: ★★반복 축이 **미확정**이고 예외 축도 안 서면 결과도 미확정이다 — #: 「미등록」이 아니다 (Codex 2026-08-31). 미확정을 미등록으로 접으면 #: 「모델이 못 붙였다」가 조용히 「안 나온다」가 된다. DISP_UNRESOLVED = "registration_unresolved" #: 등록 문턱 — ★**계약 모듈 한 곳**에서. 여기 숫자를 다시 적지 않는다. MIN_OCCURRENCES = ENTITY_MIN_OCCURRENCES def _flag(row: Dict[str, Any], key: str) -> bool: """★`bool(x)` 는 `"false"` 도 참으로 만든다 — **정확히 bool** 만 받는다. 이 부류는 이 저장소에서 반복해 난 결함이다 (상한 `int(x)` · `or` 조건). """ v = (row or {}).get(key) if v is None: return False if type(v) is not bool: raise ValueError( f"{key} 가 bool 이 아니다 ({v!r}) — 「참인 척하는 값」을 안 받는다") return v def local_id(chunk_id: str, row_index: int) -> str: """구간 행의 **local ID**. ★결정적이고 전역에서 안 겹친다. 모델에게 안정 ID 를 써내라고 하지 않는다 — 그건 매번 달라진다. """ c = str(chunk_id or "").strip() if not c: raise ValueError("chunk_id 가 비었다 — 행을 어느 구간 것으로 셀지 못 정한다") if not isinstance(row_index, int) or isinstance(row_index, bool) or row_index < 0: raise ValueError(f"row_index 가 0 이상 int 여야 한다 (받은 것: {row_index!r})") return f"{c}#{row_index}" def canonical_span(occ: Dict[str, Any]) -> Tuple[str, int, int]: """한 출현의 **정본 자리**. ★모델이 쓴 문자열이 아니라 코드가 검증한 좌표다. ★★앞 판은 `(source_anchor, source_quote)` 로 신원을 삼았다. 둘 다 모델이 낸 문자열이라, 같은 대상에 대해 **짧은 인용을 냈다가 긴 인용을 내면** 정렬 순서가 바뀌어 최종 번호가 맞바뀐다 (Codex 재현: A 의 인용만 바꿨더니 A 가 P01→P02, B 가 P02→P01). 그래서 신원은 `(segment_id, start, end)` 다 — **원문 어디인지**이고, `assert_rows` 가 그 구간이 실제로 그 인용을 담는지 본다. 모델 자유 인용은 **audit** 이지 정렬 열쇠가 아니다. """ sp = (occ or {}).get("source_span") or {} a, b = sp.get("start"), sp.get("end") # ★`int(x)` 강제를 안 한다 — `"3"` 도 `True` 도 좌표가 아니다. # 검증은 `_assert_rows` 가 하고, 여기는 **검증된 값만** 읽는다. if type(a) is not int or type(b) is not int or isinstance(a, bool) \ or isinstance(b, bool): return (str(sp.get("segment_id") or ""), -1, -1) return (str(sp.get("segment_id") or ""), a, b) def appearance_count(rows: Iterable[Dict[str, Any]]) -> Tuple[int, bool]: """반복 축이 세는 **등장 단위**. ★샷 단위다. Returns: `(수, 미확정인가)`. 미확정이면 그 수를 **확정값으로 쓰면 안 된다**. ★production `entity_filter._appearance_count` 는 `shot_count` 를 **먼저** 본다. 씬 수로 세면 같은 씬의 서로 다른 샷에 반복 등장한 대상이 1회로 줄어 기존 축이 바뀐다 (Codex 2026-08-31). 그래서 **샷 ID** 로 센다. 상태 셋을 그대로 다룬다 — bound_complete 고유 샷 ID 를 **센다** not_in_catalog_shots **0회** (명시 판정) unresolved **미확정** — false 가 아니다 ★샷 결속이 아예 없는 행(옛 모양)은 씬으로 되돌아가지 **않는다** — 미확정이다. 조용히 씬으로 세면 두 단위가 섞인다. """ from app.modules.pipeline.grounding_shot_catalog import ( BOUND_COMPLETE, BIND_UNRESOLVED, NOT_IN_CATALOG) seen: set = set() unresolved = False for r in rows or (): st = str((r or {}).get("shot_binding_status") or "") if st == BOUND_COMPLETE: seen |= {str(x) for x in ((r or {}).get("shot_appearance_ids") or ())} elif st == NOT_IN_CATALOG: continue # ★명시 0회 else: unresolved = True # unresolved · 빈 상태 둘 다 _ = BIND_UNRESOLVED return len(seen), unresolved def scene_appearance_count(rows: Iterable[Dict[str, Any]]) -> int: """서로 다른 씬의 수. ★**감사·개발 진단용**이다. 2026-08-31 B-only 진단이 이것으로 돌았다. production 반복축을 **대신하지 않는다** — `shot_count` 동등성이 아니다. """ seen: set = set() for r in rows or (): for occ in ((r or {}).get("occurrences") or ()): sp = canonical_span(occ) if sp[0]: seen.add(sp[0]) return len(seen) def unique_anchor_count(rows: Iterable[Dict[str, Any]]) -> int: """서로 다른 정본 자리의 수. ★**등장 단위가 아니다** — 감사·표시용. 등록 판정에 쓰면 한 씬 안의 여러 말이 여러 번 등장한 것이 된다. """ seen: set = set() for r in rows or (): for occ in ((r or {}).get("occurrences") or ()): sp = canonical_span(occ) if sp[0] and sp[1] >= 0: seen.add(sp) return len(seen) def _should_register(rows: Iterable[Dict[str, Any]] ) -> Tuple[Optional[bool], str]: """등록하나. ★**두 축의 union** — 사용자 확정 규칙 그대로. Returns: `(True | False | None, 사유)`. ★`None` 은 **미확정**이지 미등록이 아니다 — 반복 축을 못 정했고 예외 축도 안 선 경우다. 고유 샷 appearances ≥ 2 → 반복 출현 축 OR (만들기 어려움 AND 그곳 사람이 알아챔) → 고증 예외 축 Returns: (등록?, 사유). 사유는 장부에 남긴다 — 「왜 남았나」를 되짚어야 한다. """ rows = list(rows or ()) # ★★**등장 단위**로 센다 — 부른 횟수가 아니다. # ★production `entity_filter._appearance_count` 의 세 단위 # (`shot_count` → `scene_count` → `set(scene_appearances)`) 중 # **scene 단위**다. `shot_count` 동등성은 **주장하지 않는다** — # 구간 payload 는 씬 산문만 갖고 있어 샷 좌표를 모른다. n, bind_unresolved = appearance_count(rows) if n >= MIN_OCCURRENCES: return True, f"shot_appearances={n}" # ★★예외 축은 **helper 하나**가 판정한다 — 합치기 전에도 뒤에도 같다. # ★두 축은 **독립**이다: 반복 축이 미확정이어도 예외 축이 서면 등록된다. st = _exception_state(rows) if st == EXC_YES: return True, "grounding_exception" if st == EXC_UNRESOLVED: # ★구간마다 갈린 것은 「예외 아님」이 아니라 **미확정**이다. return None, "exception_unresolved" if bind_unresolved: # ★★반복 축을 못 정했고 예외 축도 안 섰다 → **미확정**이다. # 「미등록」으로 접으면 「모델이 못 붙였다」가 「안 나온다」가 된다. return None, "shot_binding_unresolved" return False, f"shot_appearances={n}" def validate_decision(d: Dict[str, Any], known: Sequence[str]) -> Optional[str]: """모델이 낸 한 줄이 성립하나. 안 되면 **사유 문자열**을 돌려준다. ★`relation` 이 빠지면 `same_referent` 로 **짐작하지 않는다** — 그 짐작이 `location_part` 를 parent 에 합쳐 부분을 지운다. """ src = str((d or {}).get("remove_local_id") or "") dst = str((d or {}).get("keep_local_id") or "") rel = str((d or {}).get("relation") or "") if src not in known: return f"모르는 remove id {src!r}" if dst not in known: return f"모르는 keep id {dst!r}" if src == dst: return "자기 자신으로 합친다" if rel not in RELATIONS: return f"relation 이 {rel!r} 다 — {RELATIONS} 중 하나여야 한다" return None def _assert_rows(rows: Sequence[Dict[str, Any]], segments: Optional[Dict[str, str]] = None) -> None: """★**줄이기 전에** 입력을 전수로 본다 (Codex BLOCK-3). 앞 판은 빈 `local_id` 를 조용히 버리고 겹친 것은 **마지막 행으로 덮어썼다** — 덮어쓴 행은 아무 데도 안 세어지고 사라진다. ★`segments` 를 주면 **정본 span 이 실제 그 자리인지**까지 본다 — 원문 좌표가 구간 밖이거나 인용이 그 자리에 없으면 **지어낸 것**이다. """ seen: Dict[str, int] = {} for i, r in enumerate(rows or ()): lid = str((r or {}).get("local_id") or "").strip() if not lid: raise AssertionError(f"자리 {i} 에 `local_id` 가 없다 — " "어느 행인지 못 적는다") if lid in seen: raise AssertionError( f"`local_id` 가 겹친다 ({lid} — 자리 {seen[lid]}·{i}) — " "덮어쓰면 한 행이 소리 없이 사라진다") seen[lid] = i owner = str((r or {}).get("owner_type") or "") if owner not in OWNERS: raise AssertionError(f"{lid}: 모르는 owner {owner!r} — {OWNERS}") occ = (r or {}).get("occurrences") if occ is None: raise AssertionError(f"{lid}: `occurrences` 가 없다 — " "「2회 이상」을 셀 수 없다") for j, o in enumerate(occ): sp = (o or {}).get("source_span") or {} seg = str(sp.get("segment_id") or "").strip() a, b = sp.get("start"), sp.get("end") if not seg: raise AssertionError( f"{lid}: occurrence[{j}] 에 `source_span.segment_id` 가 없다 " "— 모델이 쓴 인용은 신원이 못 된다") for name, v in (("start", a), ("end", b)): if type(v) is not int or isinstance(v, bool) or v < 0: raise AssertionError( f"{lid}: occurrence[{j}].source_span.{name} 가 " f"0 이상 int 여야 한다 ({v!r})") if b <= a: raise AssertionError( f"{lid}: occurrence[{j}] 의 span 이 비었다 ({a}~{b})") if segments is not None: text = segments.get(seg) if text is None: raise AssertionError(f"{lid}: 모르는 segment {seg!r}") if b > len(text): raise AssertionError( f"{lid}: span 이 구간 밖이다 ({b} > {len(text)})") q = (o or {}).get("source_quote") if not isinstance(q, str) or not q: # ★빈 인용은 「없는 것」이 아니라 **검증을 건너뛰는 문**이다. raise AssertionError( f"{lid}: occurrence[{j}] 에 `source_quote` 가 없다 — " "빈 값으로 원문 대조를 건너뛸 수 없다") if text[a:b] != q: # ★원문 자리와 인용이 다르면 **지어낸 것**이다. raise AssertionError( f"{lid}: occurrence[{j}] 인용이 그 자리에 없다") for k in ("hard_to_generate", "viewers_would_notice"): _flag(r, k) # ★bool 이 아니면 여기서 선다 def _reduce_rows( rows: Sequence[Dict[str, Any]], decisions: Sequence[Dict[str, Any]], ) -> Dict[str, Any]: """구간 행 + 모델 판정 → **줄인 행**과 장부. ★**모든 remove 는 유효 keeper 하나가 없으면 삭제 거부**다 (BLOCK-4). 「근거를 든 행만 보호」로는 구간 행이 실제로 사라진다. ★`part_of` 는 **안 합친다** — 관계로만 기록한다 (BLOCK-5). """ _assert_rows(rows) by_id = {str(r["local_id"]): r for r in rows} known = list(by_id) plan: Dict[str, str] = {} parts: List[Dict[str, str]] = [] refused: Dict[str, str] = {} claimed_by: Dict[str, List[str]] = {} # ★한 src 에 `part_of` 와 `same_referent` 가 **같이** 오면 상충이다. rel_of: Dict[str, set] = {} for d in decisions or (): _s = str((d or {}).get("remove_local_id") or "") _r = str((d or {}).get("relation") or "") if _s and _r in RELATIONS: rel_of.setdefault(_s, set()).add(_r) for d in decisions or (): src = str((d or {}).get("remove_local_id") or "") why = validate_decision(d, known) if why: if src: refused[src] = why continue if len(rel_of.get(src, ())) > 1: refused[src] = "같은 행에 part_of 와 same_referent 가 같이 왔다" continue dst = str(d["keep_local_id"]) if str(d["relation"]) == REL_PART_OF: # ★부분은 **합치지 않는다.** 관계만 남긴다. parts.append({"part": src, "whole": dst}) continue if by_id[src].get("owner_type") != by_id[dst].get("owner_type"): refused[src] = "owner 가 다른데 같은 실물이라고 한다" continue claimed_by.setdefault(src, []).append(dst) plan[src] = dst for src, dsts in claimed_by.items(): if len(set(dsts)) > 1: # ★한 행을 두 곳으로 — 같은 근거가 두 행에 생긴다. refused[src] = f"keeper 가 {len(set(dsts))}개다" plan.pop(src, None) # ★★**고리(cycle)를 통째로 거부한다** (Codex BLOCK-4). A→B, B→A 를 # 순회 순서대로 풀면 **누가 keeper 가 되는지가 도착 순서로 갈린다** — # 병렬에서 그건 매번 달라진다는 뜻이다. for src in list(plan): seen_path, cur = [], src while cur in plan: if cur in seen_path: for x in seen_path[seen_path.index(cur):]: refused[x] = "합치는 고리가 생겼다 — 누가 남는지 못 정한다" plan.pop(x, None) break seen_path.append(cur) cur = plan[cur] # ★keeper 도 지워지는 중이면 옮겨 봐야 같이 사라진다. removing = set(plan) for src, dst in list(plan.items()): if dst in removing: refused[src] = "keeper 도 지워지는 중이다" plan.pop(src, None) removing.discard(src) from app.modules.pipeline.grounding_binding import union_ids kept: List[Dict[str, Any]] = [] for rid, row in by_id.items(): if rid in plan: continue # ★★`plan` 은 **판정이 온 순서**로 쌓인 dict 다. 그대로 쓰면 같은 # keeper 로 둘이 합쳐질 때 `merged` 순서가 갈리고, 그것이 장부와 # (정렬 전) 출현 목록을 흔든다 — 성질 시험이 잡았다. # **local_id 로 정렬**해서 판정 순서를 지운다. merged = [by_id[s] for s in sorted( (s for s, d in plan.items() if d == rid))] if merged: row = dict(row) row["occurrences"] = _union_occ([row, *merged]) row["merged_from"] = sorted(s for s, d in plan.items() if d == rid) # ★★**두 bool 을 OR 하지 않는다.** (hard T, notice F) 와 # (hard F, notice T) 가 만나 한 행의 (T, T) 가 되면 같은 행 # 규칙이 그 자리에서 되돌아간다 (Codex 재현). # 대신 **helper 가 낸 상태를 나른다.** 원 행의 flag 는 그 행의 # 판정이라 **안 덮어쓴다** — 어느 구간이 뭐라 했는지가 남아야 한다. src_rows = (row, *merged) # ★**모든 member 의 판정을 장부로** 남긴다 — 지워진 행이 뭐라 # 했는지가 결과에 있어야 한다. row[EXCEPTION_LEDGER] = _exception_ledger(src_rows) row[EXCEPTION_STATE] = _exception_state([row]) # ★★**샷 결속도 합친다.** 안 합치면 두 구간에 걸친 실물이 # keeper 쪽 샷만 갖고 「1회」가 된다 — 반복 축이 그 자리에서 # 깨진다(시험이 잡았다). 상태는 **가장 약한 것**을 따른다: # 하나라도 미확정이면 합친 것도 미확정이다. row["shot_binding_status"], row["shot_appearance_ids"] = \ _union_shots(src_rows) gb = union_ids(row, *merged) if gb: from app.modules.pipeline.grounding_binding import FIELD row[FIELD] = gb kept.append(row) if refused: logger.warning("chunk merge: 삭제를 **거부한** 행 %d개 %s", len(refused), sorted(refused)[:5]) # ★★**행 배열도 정본 순서로** 낸다 (Codex). 한 행 안의 출현만 정렬하면 # `rows` 배열이 여전히 **도착 순서**라, 같은 입력에 다른 체크포인트 # bytes 가 나온다. 열쇠는 `evidence_key` 이고 동점은 `local_id` 로 가른다 # (근거가 같아 다투는 행도 순서가 흔들리면 안 된다). kept.sort(key=lambda r: (evidence_key(r), str(r.get("local_id") or ""))) # ★★`part_of` 도 **판정 도착 순서**를 탔다 (Codex). 「입력 순열 전체」로는 # **판정 순열**을 못 대신한다 — 내 성질 시험이 `same_referent` 만 태웠다. parts.sort(key=lambda x: (x["part"], x["whole"])) return { "contract_version": CHUNK_MERGE_CONTRACT_VERSION, "rows": kept, "merged": dict(plan), "part_of": parts, "refused": refused, "counts": {"in": len(by_id), "out": len(kept), "merged": len(plan), "part_of": len(parts), "refused": len(refused)}, } def _union_shots(rows: Sequence[Dict[str, Any]]) -> Tuple[str, List[str]]: """샷 결속을 합친다. ★상태는 **가장 약한 것**을 따른다. 하나라도 `unresolved`(또는 상태가 없음) → 합친 것도 **unresolved** 전부 `not_in_catalog_shots` → 그대로 (명시 0회) 그 밖 → `bound_complete` · ID 합집합 ★미확정을 「없음」으로 접으면 「모델이 못 붙였다」가 「안 나온다」가 된다. """ from app.modules.pipeline.grounding_shot_catalog import ( BOUND_COMPLETE, BIND_UNRESOLVED, NOT_IN_CATALOG) states = [str((r or {}).get("shot_binding_status") or "") for r in rows] ids: set = set() for r in rows: # ★★미확정 member 의 **검증된 감사 ID 도 보존**한다 (Codex). # 앞 판은 `bound_complete` 것만 남겨서, 합친 뒤 그 흔적이 사라졌다. # ★보존하되 상태가 미확정이면 **횟수로는 안 쓴다** — # `appearance_count` 가 상태를 보고 가른다. if str((r or {}).get("shot_binding_status") or "") in ( BOUND_COMPLETE, BIND_UNRESOLVED): ids |= {str(x) for x in ((r or {}).get("shot_appearance_ids") or ())} if any(st not in (BOUND_COMPLETE, NOT_IN_CATALOG) for st in states): return BIND_UNRESOLVED, sorted(ids) if all(st == NOT_IN_CATALOG for st in states): return NOT_IN_CATALOG, [] return BOUND_COMPLETE, sorted(ids) def _union_occ(rows: Sequence[Dict[str, Any]]) -> List[Dict[str, Any]]: """출현을 합친다. ★**정본 span** 이 같으면 한 번만 — **span 순**. ★한 번은 여기가 옛 `source_anchor` 칸을 보고 있어서, 신원을 span 으로 옮긴 뒤 **합칠 때 출현이 통째로 사라졌다**. 「칸을 더하면 그 칸을 복사하는 줄을 찾는다」는 그 부류다. """ # ★★**정본 span 으로 정렬**해서 낸다 — 첫 등장 순으로 두면 **병렬 구간이 # 어느 순서로 도착하느냐**에 따라 저장될 내용이 달라진다. # 실측: A→B 는 [(0,2),(6,8)], B→A 는 [(6,8),(0,2)] 였다. 체크포인트 # bytes 가 그때그때 다르면 하류 지문이 흔들려 **멀쩡한 것이 재생성**된다. seen: Dict[Tuple[str, int, int], Dict[str, Any]] = {} for r in rows: for occ in ((r or {}).get("occurrences") or ()): sp = canonical_span(occ) if sp[0] and sp[1] >= 0 and sp not in seen: seen[sp] = dict(occ) return [seen[k] for k in sorted(seen)] #: 갈래 → 접두. ★**계약 모듈**이 SOT 다 — 여기 손으로 다시 안 적는다. _PREFIX = OWNER_PREFIX def evidence_key(row: Dict[str, Any]) -> Tuple[str, ...]: """★**코드가 확인할 수 있는 근거**로만 만든 열쇠. `local_id` 를 안 쓴다. `local_id` 는 `chunk#row_index` 라 **모델이 행 순서만 바꿔도** 달라진다. 그것을 정렬 열쇠에 넣으면 같은 근거의 두 대상이 run 마다 번호를 맞바꾼다 (Codex 재현: run1 A→P01·B→P02, run2 B→P01·A→P02). ★그다음 판은 `(anchor, quote)` 를 썼는데 **그것도 모델이 쓴 문자열**이라, 인용 길이만 바뀌어도 정렬이 뒤집혔다(Codex 재현). 그래서 열쇠는 **owner + 모든 정본 span** 이다 — `(segment_id, start, end)`, 코드가 검증하는 좌표뿐이다. """ spans = tuple(sorted(canonical_span(o) for o in (row.get("occurrences") or ()))) return (str(row.get("owner_type") or ""),) + tuple( f"{seg}\x1f{a:012d}\x1f{b:012d}" for seg, a, b in spans) def _assign_final_ids(rows: Sequence[Dict[str, Any]], audit_rows: Optional[Sequence[Dict[str, Any]]] = None ) -> Dict[str, Any]: """남는 행들에 **최종 ID 를 한꺼번에** 발급한다. ★결정적이다. 앞 판의 `final_id(row, prefix, taken)` 는 `row` 를 **안 쓰고** 다음 번호만 냈다 — 도착 순서가 달라지면 같은 입력에 다른 ID 가 붙었다. ★★그리고 정렬 열쇠에 `local_id` 를 넣었더니, **모델이 행 순서만 바꿔도** 같은 근거의 두 대상이 번호를 맞바꿨다 (Codex 재현). 이제 열쇠는 `evidence_key` — 원문에서 확인되는 것뿐이다. ★근거가 **똑같은** 두 행은 구조적으로 못 가른다. 그때는 **안정 ID 를 주장하지 않는다** — `contested` 로 세운다. Returns: `ids`(local_id → 최종 ID) · `contested`(근거가 같아 못 가른 열쇠들). """ _assert_rows(rows) # ★★직접 불려도 전수 검증한다 assert_prefix_parity() # ★받는 문과 내보내는 문이 갈리면 여기서 선다 # ★★**다툼은 남은 행 전부에서** 본다(감사), **번호는 `rows` 에만** 준다. # 등록 후보끼리만 보면, 탈락한 행과 근거가 똑같은 것을 못 보고 번호를 # 줘 버린다 — 그 둘은 사실 구별이 안 되는 것이다. seen_all: Dict[Tuple[str, ...], List[str]] = {} for row in (audit_rows if audit_rows is not None else rows): seen_all.setdefault(evidence_key(row), []).append( str(row.get("local_id"))) by_key: Dict[Tuple[str, ...], List[Dict[str, Any]]] = {} for row in rows: by_key.setdefault(evidence_key(row), []).append(row) # ★★다툼은 **감사 대상 전부**에서 낸다 — 등록 후보 안에서만 세면, # 둘 다 탈락한 두 행이 근거가 똑같아도 **아무 데도 안 남는다**. # (등록 ID 에는 절대 안 넣는다. 감사와 번호는 다른 일이다.) contested: List[Dict[str, Any]] = [ {"evidence": list(k), "local_ids": sorted(v)} for k, v in sorted(seen_all.items()) if len(v) > 1 ] blocked_keys = {k for k, v in seen_all.items() if len(v) > 1} ids: Dict[str, str] = {} counters: Dict[str, int] = {} for key in sorted(by_key): if key in blocked_keys: continue # ★못 가르는 것에 번호를 주지 않는다 row = by_key[key][0] pre = _PREFIX[str(row.get("owner_type"))] counters[pre] = counters.get(pre, 0) + 1 ids[str(row.get("local_id"))] = f"{pre}{counters[pre]:02d}" if contested: logger.warning("chunk merge: 근거가 같아 ID 를 못 준 무리 %d개", len(contested)) return {"ids": ids, "contested": contested} # ── ★★공개 경로는 **이것 하나**다 ──────────────────────────────────────── # # Codex (2026-08-31): 「`reduce_rows` 도 `assign_final_ids` 도 `segments` # 없이 `assert_rows(rows)` 만 부르고, `should_register` 는 검증을 아예 안 # 부릅니다. 즉 **실제 경로는 원문 없이 통과합니다.**」 # # 실제로 그랬다 — `source_quote="WRONG"` 인 행이 `reduce_rows` 를 그냥 # 지나갔다. 그래서 위 셋을 `_` 로 닫고, **원문을 반드시 받는 끝점** 하나만 # 남긴다. 시험이 아닌 자리에서 `_` 함수를 부르면 그것이 결함이다. def _settle_host_context(raw: Sequence[Dict[str, Any]], reduced: Dict[str, Any], segments: Dict[str, str]) -> None: """`location_part` 의 **맥락 칸**을 계약대로 내린다. ★판정은 저쪽 모듈이. ★★여러 구간이 서로 다른 것을 냈으면 **한쪽을 고르지 않는다** — `reconcile` 이 뜻이 같으면 근거를 합치고, 다르면 `unresolved` 로 내린다. ★근거 인용이 **원문 그 자리에 있었나**는 기존 span 대조를 태워 넘긴다 — 여기서 원문을 다시 읽으면 같은 규칙이 두 곳이 된다. """ from app.modules.pipeline import grounding_host_context as hc # ★`rows` 는 **평평한 행 목록**이다 — 구간 봉투가 아니다 raw_by_id: Dict[str, Dict[str, Any]] = {} for rr in raw or (): lid = str((rr or {}).get("local_id") or "") if lid: raw_by_id.setdefault(lid, rr) rows_by_id = {str(r.get("local_id") or ""): r for r in (reduced.get("rows") or ())} part_of = list(reduced.get("part_of") or ()) for row in reduced.get("rows") or (): if str(row.get("owner_type") or "") != hc.HOST_CONTEXT_OWNER: continue lid = str(row.get("local_id") or "") # ★합쳐진 member 는 **예외 장부**에 남아 있다 — 새 칸을 만들지 않는다 members = {lid} | {str(e.get("local_id") or "") for e in (row.get(EXCEPTION_LEDGER) or ())} seen, verified = [], [] for src_lid in sorted(x for x in members if x): src = raw_by_id.get(str(src_lid)) if src is None: continue got = (src or {}).get("host_context") if not isinstance(got, dict): continue ev_ok, ev_spans = _verify_context_evidence(got, row, segments) if ev_ok: verified.extend(ev_spans) seen.append(hc.normalize(got, row_local_id=str(src_lid), rows_by_id=rows_by_id, part_of=part_of, evidence_ok=ev_ok)) if not seen: continue state, kept, why = hc.reconcile(seen) out = {"state": state, **kept, **({"why": why} if why else {})} if state == hc.CONTEXT_ONLY and verified: # ★★검증된 **좌표**를 붙인다 — `reconcile` 은 제 계약의 칸만 # 들고 오므로 그 **뒤에** 얹는다. 뒤가 글자만 받으면 어느 자리 # 것인지 다시 못 찾는다 (Codex BLOCK · 09-01). # ★차례를 못박는다 — 구간 도착 순서가 결과를 바꾸면 안 된다. uniq = {(sp["segment_id"], sp["start"], sp["end"]): sp for sp in verified} out["evidence_spans"] = [uniq[k] for k in sorted(uniq)] row["host_context"] = out def _verify_context_evidence(hc_raw: Dict[str, Any], row: Dict[str, Any], segments: Dict[str, str] ) -> Tuple[bool, List[Dict[str, Any]]]: """근거가 **그 부분이 나온 자리**에 정말 있나 — 좌표까지 돌려준다. ★★★앞 판은 에피소드 **전체를 이어 붙인 글자열**에서 인용을 찾았다. 그러면 — search_subject 「원문에 없는 고급 호텔」 ← 원문 어디에도 없다 evidence 「그는 회전 간판 앞에 섰다」 ← 원문 **어딘가**엔 있다 → 통과 (Codex 재현 · 2026-09-01) 반복 장소가 있는 실제 원고에서는 **다른 장소의 인용**이나 **지어낸 장소 이름**으로 참조 조사가 나간다. 프롬프트가 「원문 그대로」라고 말하는 것은 계약이 아니다. 그래서 두 가지를 **그 행의 자리에서** 본다 — ①근거 인용이 **그 행이 나온 씬**에 있다 (기존 span 찾기를 그대로 쓴다) ②`search_subject` 가 **그 검증된 인용 안에** 정확한 글자로 있다 ★②는 이름으로 뜻을 추론하는 것이 아니라 **출처 대조**다 — 모델이 원문에 없는 말을 지어냈는지를 본다. Returns: `(ok, spans)`. `spans` 는 `{segment_id, start, end, source_quote}` 로 **기존 좌표 모양 그대로**다 — 뒤가 다른 모양을 배우지 않게. """ from app.modules.pipeline.grounding_chunk import _find_span ev = [str(x) for x in (hc_raw.get("evidence") or ()) if str(x).strip()] if not ev: return True, [] # ★없는 것은 여기서 판단하지 않는다 # ★★그 행이 **실제로 나온 씬**만 본다 — 에피소드 전체가 아니다 where = [str(((o or {}).get("source_span") or {}).get("segment_id") or "") for o in ((row or {}).get("occurrences") or ())] where = [x for x in where if x and x in segments] if not where: return False, [] spans: List[Dict[str, Any]] = [] for q in ev: try: sp = _find_span(q, 1, where, segments) except ValueError: return False, [] # ★그 자리에 없다 — 다른 씬 것이거나 지어낸 것 spans.append({**sp, "source_quote": q}) subj = str(hc_raw.get("search_subject") or "").strip() if subj and not any(subj in sp["source_quote"] for sp in spans): # ★모델이 근거에 없는 말을 지어냈다 — 그 말로 조사하러 나가면 안 된다 return False, [] return True, spans def reduce_episode(rows: Sequence[Dict[str, Any]], decisions: Sequence[Dict[str, Any]], *, segments: Dict[str, str]) -> Dict[str, Any]: """구간 행 + 판정 → 줄인 행 · 등록 여부 · 최종 ID. ★**공개 경로 하나.** ★`segments` 는 **required keyword-only** 다. 원문 없이 부를 길을 안 둔다 — 그 길이 있으면 지어낸 인용이 그대로 통과한다. ★맨 앞에서 **전수 검증**한다. provider 를 타기 전에, 줄이기 전에. """ if not isinstance(segments, dict) or not segments: raise AssertionError( "`segments` 가 비었다 — 원문 없이 정본 span 을 검증할 수 없다") _assert_rows(rows, segments) # ★★원문 대조까지 여기서 assert_prefix_parity() reduced = _reduce_rows(rows, decisions) _settle_host_context(rows, reduced, segments) # ★★★등록 판정은 **행마다** 한다 (Codex). # `_should_register` 의 입력 의미는 「**같은 실물**의 여러 관측 행」이다. # 줄인 뒤의 행 전부를 한 번에 넣으면 **서로 다른 엔티티의 출현 수와 # 예외 판정이 섞인다** — 각각 한 번뿐인 두 물건이 「2회」가 되어 둘 다 # 등록됐다. 실제로 그랬다. # ★★**판정이 먼저다.** 앞 판은 ID 를 먼저 주는 바람에 **탈락한 후보가 # 번호를 먹었고**, 그 행이 있고 없고에 따라 등록된 것의 ID 가 P01↔P02 # 로 바뀌었다 (Codex 재현). 최종 ID 는 「등록된 엔티티의 신원」이지 # 「이 판에 나온 행의 순번」이 아니다. verdict = {str(r.get("local_id")): _should_register([r]) for r in reduced["rows"]} # ★★미확정(`None`)은 **등록도 미등록도 아니다**. ID 를 주지 않되 # 「안 나온다」로도 접지 않는다 (Codex 2026-08-31). eligible = [r for r in reduced["rows"] if verdict[str(r.get("local_id"))][0] is True] ids = _assign_final_ids(eligible, reduced["rows"]) contested_ids = {lid for g in ids["contested"] for lid in g["local_ids"]} registered: Dict[str, Any] = {} for r in reduced["rows"]: lid = str(r.get("local_id")) ok, why = verdict[lid] if lid in contested_ids: registered[lid] = {"registered": False, "reason": why, "disposition": DISP_ID_CONTESTED, "final_id": None} continue if ok is None: registered[lid] = {"registered": None, "reason": why, "disposition": DISP_UNRESOLVED, "final_id": None} continue registered[lid] = { "registered": ok, "reason": why, "disposition": DISP_REGISTERED if ok else DISP_NOT_REGISTERED, "final_id": ids["ids"].get(lid) if ok else None, } return {**reduced, "registered": registered, "final_ids": ids["ids"], "id_contested": ids["contested"]}