"""C — **실제 에피소드 하나**로 shot-aware canary. ★`--live` 는 승인 뒤에. `cc_runner` 를 **그대로 쓴다** — 두 벌로 만들면 한쪽만 고쳐진다. 여기서는 실제 체크포인트로 **계획·원문·샷 목록**을 만들어 넘길 뿐이다. ## 자동으로 재는 것 / 사람이 보는 것 Codex 가 못박았다 (2026-08-31) — 자동 schema/runtime enum · 씬 제약 · 격리 · 장부/재개 · Opik/provider 대조 사람만 샷 ID 가 **의미상** 맞나 · 두 축 · 엔티티/facet 품질 ★표적 채점(`cc_scorer`)을 **안 쓴다.** 실제 에피소드에는 심어 둔 표적이 없고, 있는 척하면 그것이 바로 지어낸 정답이다. python tools/grounding_audit/cc_c_runner.py --dry python tools/grounding_audit/cc_c_runner.py --replay # 무구매 python tools/grounding_audit/cc_c_runner.py --live # ★유료 """ from __future__ import annotations import json import sys from pathlib import Path from typing import Any, Dict, List, Tuple ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT)) sys.path.insert(0, str(ROOT / "tests")) from tools.grounding_audit import cc_c_preflight as pf # noqa: E402 from tools.grounding_audit import cc_veto_review as vr # noqa: E402 from tools.grounding_audit import cc_runner as rr # noqa: E402 from tools.grounding_audit.call_payload_table import seal_outbound # noqa: E402 #: ★★**손으로 적은 승인 수**. CP 에서 다시 계산하지 않는다 — 승인 뒤 구간이 #: 하나 늘면 논리 6 · 물리 12 로 **조용히 넓어진다** (Codex 2026-08-31). #: 계획이 이 수와 다르면 **장부·trace·provider 앞에서** 선다. C_APPROVED_LOGICAL = 5 C_APPROVED_SLOTS = 2 #: ★C 전용 Opik 좌표. 실제 에피소드를 **합성 B 이름으로** 기록하면 감사를 #: 오독한다. sender 를 복사하지 않고 좌표만 바꾼다. C_TRACE_NAME = "cc_c_episode_chunk" C_TAG = "op:cc-c-episode" C_THREAD = "cc-c-episode" C_STEP = "grounding_chunk_cc_episode" class ApprovedScopeMismatch(RuntimeError): """계획이 **승인된 수**와 다르다. ★아무것도 안 하고 선다.""" def assert_approved(plan) -> None: """★장부·trace·provider **앞에서** 선다.""" n = len(plan) + 1 if n != C_APPROVED_LOGICAL: raise ApprovedScopeMismatch( f"계획이 논리 {n} 인데 승인은 {C_APPROVED_LOGICAL} 이다 " f"(구간 {len(plan)} + merge 1). 사람이 다시 정해야 한다 — " "장부도 안 열고 provider 도 안 부른다") def build_plan() -> Tuple[Dict[str, Any], List[Dict[str, Any]], Dict[str, str], Dict[str, Any]]: """고른 에피소드로 계획을 짓는다. ★selector 는 preflight 것을 **그대로**.""" from app.core.world_context import build_grounding_world_facts from app.modules.pipeline import grounding_chunk as gc from app.modules.pipeline import grounding_shot_catalog as sc pick, _mid = pf.choose(pf.candidates()) if not pick: raise LookupError("후보 에피소드가 없다") base = (pf.PROJECTS / pick["project_id"] / "checkpoints" / "episodes" / pick["episode_id"]) S = pf._load(base / "scene_save" / "manifest.json") or {} V = pf._load(base / "shot_validator" / "manifest.json") or {} Wcp = pf._load_cp(base / "visual_world_rules" / "manifest.json") world = build_grounding_world_facts(Wcp) segs = {f"scene-{s['scene_index']}": (s.get("text") or "") for s in S.get("segments") or []} plan, cats = [], {} for n, b in enumerate(pick["bundles"]): ids = [f"scene-{i}" for i in b] cat = sc.build_catalog(V.get("scenes") or [], ids) cats[f"c{n}"] = cat plan.append({ "chunk_id": f"c{n}", "segment_ids": ids, "payload": gc.build_chunk_payload(ids, segs, world, shot_catalog=cat), }) return pick, plan, segs, {"world": world, "catalogs": cats} def _lock_extra(pick, plan, cats) -> Dict[str, Any]: """잠금에 **계획된 획득 신원**까지 넣는다 (Codex 2026-08-31). ★앞 판은 project/episode/chunks/shot_ids 만 넣어서, **원문이나 샷 전문이 바뀌면 잠금은 같은데 획득 신원만 달라졌다.** 그러면 부분 장부에 옛 2콜과 새 3콜이 섞이고 dispatch cap 에서야 선다 — 중복 구매와 기록 오염이다. ★merge 신원은 **구간 응답 뒤**에 생기므로 여기 못 넣는다. 다만 구간 신원이 같으면 같은 raw 가 나오고 merge 신원도 그대로 재현된다. """ phys = rr.physical_model() return { "project_id": pick["project_id"], "episode_id": pick["episode_id"], "chunks": len(plan), "shot_ids": sorted(c["id"] for cat in cats.values() for c in cat), "chunk_acquisition_ids": [ _acq(c["payload"], phys) for c in plan], "merge_acquisition": "★구간 응답 뒤에 정해진다", "approved_logical": C_APPROVED_LOGICAL, "approved_slots": C_APPROVED_SLOTS, "trace": {"name": C_TRACE_NAME, "tag": C_TAG, "thread": C_THREAD, "step": C_STEP}, } def _acq(payload, phys) -> str: """★preflight 와 **같은 production callable**.""" from app.modules.pipeline import grounding_chunk as gc return gc.acquisition_identity(payload, model_alias=rr.MODEL_ALIAS, model_physical=phys, request_contract=rr.PINNED) def automatic_checks(got: Dict[str, Any], cats: Dict[str, Any], segs: Dict[str, str]) -> List[str]: """★**구조만** 본다. 의미는 한 줄도 판정하지 않는다. ★★재는 자리가 **격리를 통과한 행**이다 (Codex 2026-08-31). 그래서 0 은 「모델이 계약을 다 지켰다」가 아니라 **「검증기가 위반을 막아냈다」**는 뜻이다. 부르는 쪽은 반드시 **격리 수와 나란히** 적어야 한다 — 앞 보고가 이 구분 없이 「어긋남 0」만 내서 결과를 거꾸로 읽게 했다. """ from app.modules.pipeline import grounding_shot_catalog as sc bad: List[str] = [] known = {cid: {c["id"] for c in cat} for cid, cat in cats.items()} where = {cid: sc.scene_of(cat) for cid, cat in cats.items()} for r in got["reduced"]["rows"]: cid = str(r["local_id"]).split("#")[0] scenes = {o["source_span"]["segment_id"] for o in r["occurrences"]} for x in r["shot_appearance_ids"]: if x not in known.get(cid, set()): # ★합쳐진 행은 다른 구간 ID 를 가질 수 있다 — 전체에서 본다 if not any(x in v for v in known.values()): bad.append(f"{r['local_id']}: catalog 밖 샷 {x}") continue if where[cid].get(x) not in scenes and len(scenes) == 1: bad.append(f"{r['local_id']}: 씬 밖 결속 {x}") for o in r["occurrences"]: sp = o["source_span"] if segs.get(sp["segment_id"], "")[sp["start"]:sp["end"]] \ != o["source_quote"]: bad.append(f"{r['local_id']}: 인용이 그 자리에 없다") return bad def main() -> int: if len(sys.argv) < 3 or sys.argv[1] not in ("--dry", "--live", "--replay"): print(__doc__) return 2 mode, jp = sys.argv[1], Path(sys.argv[2]) if mode in ("--dry", "--replay"): seal_outbound() pick, plan, segs, extra = build_plan() cats = extra["catalogs"] # ★★승인 수는 **손으로 적은 것**이다 — 계획에서 다시 계산하지 않는다. assert_approved(plan) lock_extra = _lock_extra(pick, plan, cats) if mode == "--replay": got = rr.replay(jp, world_facts=extra["world"], plan=plan, segments=segs, shot_catalogs=cats) else: def _send(payload, ident, rid=""): return rr.live_send(payload, ident, rid, trace_name=C_TRACE_NAME, tag=C_TAG, thread=C_THREAD, step=C_STEP) send = _send if mode == "--live" else rr._dry_send got = rr.run(jp, send, world_facts=extra["world"], cap=C_APPROVED_LOGICAL, dispatch_budget=C_APPROVED_LOGICAL, approved_slots=(C_APPROVED_SLOTS if mode == "--live" else None), plan=plan, segments=segs, shot_catalogs=cats, lock_extra=lock_extra) bad = automatic_checks(got, cats, segs) out = jp.with_name(jp.stem + "_run.json") out.write_text(json.dumps({ **{k: v for k, v in got.items() if k != "raw"}, "shot_catalog": [c for cat in cats.values() for c in cat], "note": ("★**shot-aware 경로 가능성 진단**이다. 다섯 갈래 완료도 " "production PASS 도 아니다. 샷 ID 가 의미상 맞나 · 두 축 · " "엔티티/facet 품질은 **사람만** 본다."), "automatic_problems": bad, }, ensure_ascii=False, indent=1, default=str), encoding="utf-8") print(f"■ C {mode[2:]} — {pick['project_id'][:8]}/{pick['episode_id'][:8]}") print(f" 논리 {got['logical']} (산 것 {got['bought']} · 재사용 " f"{got['reused']}) · dispatch {got['dispatched']}/" f"{got['dispatch_budget']}") # ★세는 자리는 **검토 화면과 같은 함수**다 — 두 벌이면 두 수가 갈린다 t = vr._tally({**got, "automatic_problems": bad}) print(f" 모델이 낸 행 {t['model_rows']} = 살아남은 {t['survived']} + " f"격리 {t['quarantined']} {t['kinds']}") print(f" merge 뒤 {t['reduced']} (합쳐진 것 {t['merged']} · " f"삭제 거부 {len(t['refused'])})") print(f" 등록 {t['registered']} {t['why']} · 처분 {t['disposition']}") print(f" 해석 지문 {got['processing_stamps']}") print(f" ★격리 **뒤** 잔존 어긋남 **{len(bad)}건**" + (f" {bad[:3]}" if bad else "") + f" — 위 {t['quarantined']}행은 **검증기가 막은 것**이다") print(f" 적었다: {out}") if mode == "--live": print(f" ★Opik 대조 — tag={C_TAG} · trace={C_TRACE_NAME} · " f"run={got['run_id']}") print(" ★자동은 **구조만** 봤다. 의미는 검토 화면에서 사람이 본다.") return 1 if bad else 0 if __name__ == "__main__": raise SystemExit(main())