"""B — **실제 체크포인트**로 A 계약을 태운다. ★유료 0 · 외부 0 · 활성화 0. Codex 가 B 의 범위를 못박았다 (2026-08-31): 「A 계약의 **실제 CP 모양 비회귀**」로만 좁힌다. 다섯 갈래 production SOT 완료나 품질 PASS 로 **넓히지 않는다.** ## 무엇을 태우나 실제 `shot_validator` + `scene_save` 체크포인트 → 구간 나누기(생산 관례) → 샷 catalog → runtime enum → (모델 대신) **catalog 에서 뽑은 결속**으로 행을 짓고 → resolve_rows → reduce_episode → facet bind → 불변식 검사 ★모델을 안 부른다. **모델이 낼 법한 모양**을 catalog 에서 만들어 태운다 — 이 단계가 재는 것은 「우리 코드가 실제 CP 모양을 견디나」지 모델 품질이 아니다. ## 재는 것 ① 샷 ID 가 전부 catalog 안이고 씬 결속이 맞나 ② runtime enum 이 그 호출 catalog 로 좁혀지나 (빈 catalog 는 닫히나) ③ 두 구간에 걸친 실물의 샷이 merge 에서 **합쳐지나** ④ 등록 축이 샷 수로 서나 · 미확정이 전파되나 ⑤ facet 결속과 **빚**이 갈리나 (location_part 는 계속 빚) ⑥ 계약이 바뀌면 하류가 **stale** 되나 (지문) ⑦ provider 호출 **0** python tools/grounding_audit/cc_b_verify.py [최대 에피소드 수] """ from __future__ import annotations import hashlib import json import sys from pathlib import Path from typing import Any, Dict, List ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT)) sys.path.insert(0, str(ROOT / "tests")) from tools.grounding_audit.call_payload_table import seal_outbound # noqa: E402 PROJECTS = ROOT.parent / "projects" def _episodes(limit: int): for sv in sorted(PROJECTS.glob( "*/checkpoints/episodes/*/shot_validator/manifest.json")): ss = sv.parent.parent / "scene_save" / "manifest.json" if not ss.exists(): continue try: S = json.loads(ss.read_text(encoding="utf-8")).get("data") or {} V = json.loads(sv.read_text(encoding="utf-8")).get("data") or {} except Exception: # noqa: BLE001 continue if not (S.get("segments") and V.get("scenes")): continue yield sv.parent.parent.parent.parent.parent.name, S, V limit -= 1 if limit <= 0: return def _bundles(segments: List[Dict[str, Any]]) -> List[List[int]]: """생산 관례 그대로. ★묶는 규칙은 **한 곳**이다 — 여기 다시 안 적는다.""" from app.modules.pipeline.grounding_chunk_plan import bundle_scenes return bundle_scenes( [{"idx": int(sg.get("scene_index") or 0), "length": len(sg.get("text") or "")} for sg in segments]) def _model_rows(catalog, segments_text, ids) -> List[Dict[str, Any]]: """모델이 **낼 법한** 행. ★catalog 와 원문에서만 짓는다 — 지어내지 않는다. ★이것은 모델이 아니다. 우리 코드가 **실제 CP 모양**을 견디는지 보려고 쓰는 입력이고, 품질을 재지 않는다. """ rows = [] for n, sid in enumerate(ids): body = segments_text[sid] word = next((w for w in body.split() if len(w) >= 2), None) if not word: continue mine = [c for c in catalog if c["scene_id"] == sid] rows.append({ "owner_type": ("outlook" if n % 3 == 1 else "location_part" if n % 3 == 2 else "prop"), "surface_form": word, "mentions": [{"mention_quote": word, "occurrence_index": 1}], "evidence_quotes": [], "hard_to_generate": True, "viewers_would_notice": True, "visual_brief": "", "search_terms_native": ["가", "나"], "language_lock_native": "어느 말로", "shot_binding_status": ("bound_complete" if mine else "not_in_catalog_shots"), "shot_appearance_ids": [mine[0]["id"]] if mine else [], }) # ★부모가 될 base 행도 하나 — facet 결속을 실제로 태우려고 rows.append({ "owner_type": ("character" if n % 3 == 1 else "location"), "surface_form": word, "mentions": [{"mention_quote": word, "occurrence_index": 1}], "evidence_quotes": [], "hard_to_generate": True, "viewers_would_notice": True, "visual_brief": "", "search_terms_native": ["가", "나"], "language_lock_native": "어느 말로", "shot_binding_status": ("bound_complete" if mine else "not_in_catalog_shots"), "shot_appearance_ids": [mine[0]["id"]] if mine else [], }) return rows def _identity_axes(payload) -> List[str]: """★**production callable** 로 잰다 — 도구가 제 것을 만들면 자기검사다. ①획득 신원이 재료마다 움직이나 ②후처리만 바뀌면 **획득은 그대로**인가. """ from app.modules.pipeline import grounding_chunk as gc kw = dict(model_alias="gpt", model_physical="p", request_contract={"num_retries": 0}) a = gc.acquisition_identity(payload, **kw) out = [] if a != gc.acquisition_identity(payload, **kw): out.append("★같은 입력인데 획득 신원이 흔들린다") if a == gc.acquisition_identity(payload, **{**kw, "model_physical": "q"}): out.append("★모델이 바뀌었는데 획득 신원이 그대로다") if gc.processing_stamp(a, "1.0") == gc.processing_stamp(a, "2.0"): out.append("★후처리가 바뀌었는데 해석 지문이 그대로다") if a != gc.acquisition_identity(payload, **kw): out.append("★후처리 검사가 획득 신원을 흔들었다") return out def verify(limit: int = 8) -> Dict[str, Any]: from app.modules.pipeline import grounding_chunk as gc 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 grounding_shot_catalog as sc tot = {"episodes": 0, "chunks": 0, "rows": 0, "quarantined": 0, "registered": 0, "unresolved": 0, "bindings": 0, "debt": 0, "debt_reasons": {}, "provider_calls": 0, # ★`part_of` 는 행을 **안 합친다** — 합집합 경로를 따로 태워야 한다 "same_referent_merges": 0, "shot_unions": 0} problems: List[str] = [] last_payload = None for pid, S, V in _episodes(limit): segs = {f"scene-{s['scene_index']}": (s.get("text") or "") for s in S["segments"]} shots = V["scenes"] tot["episodes"] += 1 all_rows: List[Dict[str, Any]] = [] for n, b in enumerate(_bundles(S["segments"])): ids = [f"scene-{i}" for i in b] cat = sc.build_catalog(shots, ids) tot["chunks"] += 1 # ② runtime enum 이 이 호출 catalog 로 좁혀지나 p = gc.build_chunk_payload(ids, segs, "세계", shot_catalog=cat) last_payload = p node = (p["schema"]["properties"]["rows"]["items"]["properties"] ["shot_appearance_ids"]) if cat: got = node["items"].get("enum") if got != [c["id"] for c in cat]: problems.append(f"{pid[:8]} c{n}: enum 이 catalog 와 다르다") if "maxItems" in node: problems.append(f"{pid[:8]} c{n}: 있는데 문이 닫혔다") elif node.get("maxItems") != 0: problems.append(f"{pid[:8]} c{n}: 빈 catalog 인데 문이 열렸다") out = gc.resolve_rows(_model_rows(cat, segs, ids), chunk_id=f"c{n}", segment_ids=ids, segments=segs, shot_catalog=cat) tot["rows"] += len(out["rows"]) tot["quarantined"] += len(out["quarantined"]) all_rows += out["rows"] # ① 샷 ID 가 전부 catalog 안이고 씬 결속이 맞나 known = {c["id"] for c in cat} where = sc.scene_of(cat) for r in out["rows"]: for x in r["shot_appearance_ids"]: if x not in known: problems.append(f"{pid[:8]} c{n}: catalog 밖 {x}") elif where[x] not in {o["source_span"]["segment_id"] for o in r["occurrences"]}: problems.append(f"{pid[:8]} c{n}: 씬 밖 결속 {x}") if not all_rows: continue # ★★facet **성공 경로**도 태운다. 판정을 안 주면 결속이 0 이 되고, # 그러면 「비회귀」라고 쓰면서 **한 번도 안 돈 축**을 세게 된다. # 행은 (facet, base) 짝으로 지어졌으므로 그 짝을 관계로 잇는다. decisions = [] # ★★③ **같은 실물 합치기** 경로 — `part_of` 로는 안 탄다 (Codex). # 서로 **다른 chunk** 의, 유효한 샷 ID 를 가진 두 행을 **결정적으로** # 골라 `same_referent` 를 넣는다. 내용을 보고 고르지 않는다 — # chunk 번호와 local_id 순서로만 고른다. bound = [r for r in all_rows if r["shot_binding_status"] == "bound_complete" and r["shot_appearance_ids"]] by_chunk: Dict[str, List[Dict[str, Any]]] = {} for r in bound: by_chunk.setdefault(str(r["local_id"]).split("#")[0], []).append(r) keys = sorted(by_chunk) if len(keys) >= 2: a0 = sorted(by_chunk[keys[0]], key=lambda x: x["local_id"])[0] b0 = sorted(by_chunk[keys[1]], key=lambda x: x["local_id"])[0] if a0["owner_type"] == b0["owner_type"]: decisions.append({"remove_local_id": b0["local_id"], "keep_local_id": a0["local_id"], "relation": cm.REL_SAME}) tot["same_referent_merges"] += 1 want = set(a0["shot_appearance_ids"]) | \ set(b0["shot_appearance_ids"]) else: want = None else: want = None for a, b in zip(all_rows[0::2], all_rows[1::2]): if fb.is_facet(str(a.get("owner_type"))) and \ str(b.get("owner_type")) == fb.parent_of( str(a.get("owner_type"))): decisions.append({"remove_local_id": a["local_id"], "keep_local_id": b["local_id"], "relation": cm.REL_PART_OF}) red = cm.reduce_episode(all_rows, decisions, segments=segs) if want is not None: keeper = next((r for r in red["rows"] if str(r["local_id"]) == a0["local_id"]), None) if keeper is None: problems.append(f"{pid[:8]}: keeper 가 사라졌다") elif set(keeper["shot_appearance_ids"]) != want: problems.append( f"{pid[:8]}: ★합칠 때 샷이 안 합쳐졌다 " f"{sorted(keeper['shot_appearance_ids'])} != {sorted(want)}") else: tot["shot_unions"] += 1 for lid, rec in red["registered"].items(): if rec["registered"] is True: tot["registered"] += 1 elif rec["registered"] is None: tot["unresolved"] += 1 # ⑤ facet 결속과 빚 fbo = fb.bind(red["rows"], red["part_of"], red["registered"]) fb.assert_owner_pairs(fbo["bindings"]) tot["bindings"] += len(fbo["bindings"]) tot["debt"] += len(fbo["debt"]) for d in fbo["debt"]: tot["debt_reasons"][d["reason"]] = \ tot["debt_reasons"].get(d["reason"], 0) + 1 if d["owner_type"] == "location_part" \ and d["reason"] != fb.DEBT_NO_CANON_TYPE: problems.append( f"{pid[:8]}: location_part 빚 사유가 {d['reason']} 다 — " "갈래가 없다는 사유여야 한다") for b in fbo["bindings"]: if b["owner_type"] == "location_part": problems.append(f"{pid[:8]}: ★location_part 를 붙였다") if not b["final_id"] or not b["parent_final_id"]: problems.append(f"{pid[:8]}: ★ID 없이 붙었다") # ⑥ 획득/해석 두 신원 — ★production callable 로 잰다 if last_payload is not None: problems += _identity_axes(last_payload) else: problems.append("★신원 축을 **안 쟀다** — payload 가 하나도 없었다") # ★한 번도 안 돈 축이 있으면 「비회귀」라고 쓰면 안 된다 for name, n in (("facet 결속", tot["bindings"]), ("빚", tot["debt"]), ("등록", tot["registered"]), ("같은 실물 합치기", tot["same_referent_merges"]), ("샷 합집합", tot["shot_unions"])): if n == 0: problems.append(f"★{name} 축이 **한 번도 안 돌았다** — 0 은 " "「통과」가 아니라 「안 쟀다」다") return {"totals": tot, "problems": problems} def main() -> int: seal_outbound() limit = int(sys.argv[1]) if len(sys.argv) > 1 else 8 from app.modules.pipeline import grounding_chunk as gc r = verify(limit) t = r["totals"] print(f"■ B — 실제 체크포인트 {t['episodes']} 에피소드 · 구간 {t['chunks']}") print(f" 팩 {gc.CHUNK_PACK_VERSION} · 후처리 {gc.PROCESSING_CONTRACT_VERSION}") print(f" 행 {t['rows']} · 격리 {t['quarantined']}") print(f" 등록 {t['registered']} · 미확정 {t['unresolved']}") print(f" facet 결속 {t['bindings']} · 빚 {t['debt']} {t['debt_reasons']}") print(f" 같은 실물 합치기 {t['same_referent_merges']} · " f"샷 합집합 확인 {t['shot_unions']}") print(f" provider 호출 **{t['provider_calls']}**") print(" ★미확정 축은 이 판에서 **0 건 났다** — 입력이 성한 것이지 " "「안 잰 것」이 아니다. 그 축의 반례는 단위 시험에 있다.") if r["problems"]: print(f" ★어긋난 것 {len(r['problems'])}건:") for x in r["problems"][:8]: print(f" {x}") else: print(" → A 계약이 실제 CP 모양에서 **비회귀**") print(" ★범위 — 「A 계약의 실제 CP 모양 비회귀」뿐이다. 다섯 갈래 " "production SOT 완료나 품질 PASS 가 **아니다**.") print(" ★location_part 는 계속 **durable debt** 다.") return 1 if r["problems"] else 0 if __name__ == "__main__": raise SystemExit(main())