"""C — shot-aware 유료 canary **preflight**. ★유료 0 · 외부 0 · 활성화 0. Codex 가 C 범위를 정했다 (2026-08-31): 합성 원고가 아니라 **실제 에피소드 하나**다. 8샷 합성으로는 실제 catalog 부하(구간당 p50 44 · max 228)에서 모델이 runtime 샷 ID 를 고를 수 있는지 **못 잰다**. ## ★내용을 보고 고르면 체리피킹이다 그래서 **모델 출력을 열기 전에 결정적 selector 를 잠근다.** 고르는 기준은 구조뿐이고, 원고 내용·이름·산문은 **한 줄도 안 읽는다**. ① 후보 `scene_save` + `shot_validator` 가 다 있고, 세계 사실이 있고, 구간이 **2개 이상** ② 고르기 전수 분포의 **중앙 구조**에 가장 가까운 것 — 구간 수 · 구간당 샷 수 · catalog bytes 의 **중앙값 거리** 동률이면 **에피소드 id** 순(안정) ③ 갈래 기존 **구조화 신호**로 무료로 본다. ★단 **고르는 데는 안 쓴다** — 보고와 **coverage debt** 로만 쓴다(문구와 코드를 한 벌로). 믿을 SOT 가 없으면 `coverage unknown`. 산문·이름은 안 읽는다. python tools/grounding_audit/cc_c_preflight.py """ from __future__ import annotations import json import statistics as st import sys from pathlib import Path from typing import Any, Dict, List, Optional, 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_runner as rr # noqa: E402 from tools.grounding_audit.call_payload_table import seal_outbound # noqa: E402 PROJECTS = ROOT.parent / "projects" def _load_cp(p: Path) -> Optional[Dict[str, Any]]: """체크포인트 **통째로** — `build_*` 가 `{"data": ...}` 모양을 받는다.""" try: return json.loads(p.read_text(encoding="utf-8")) except Exception: # noqa: BLE001 return None def _load(p: Path) -> Optional[Dict[str, Any]]: d = _load_cp(p) return (d or {}).get("data") or None def _completed(p: Path) -> bool: """★`status=completed` 이고 실패가 없나 — 명시 gate (Codex NON-BLOCK).""" d = _load_cp(p) or {} if str(d.get("status") or "") != "completed": return False return int(d.get("failed_count") or 0) == 0 def candidates() -> List[Dict[str, Any]]: """후보 전수. ★구조만 본다 — 원고 내용을 안 읽는다.""" from app.modules.pipeline import grounding_shot_catalog as sc from app.modules.pipeline.grounding_chunk_plan import bundle_scenes out: List[Dict[str, Any]] = [] for sv in sorted(PROJECTS.glob( "*/checkpoints/episodes/*/shot_validator/manifest.json")): base = sv.parent.parent S = _load(base / "scene_save" / "manifest.json") V = _load(sv) Wcp = _load_cp(base / "visual_world_rules" / "manifest.json") W = (Wcp or {}).get("data") or {} if not (S and V and W): continue segs = S.get("segments") or [] scenes = V.get("scenes") or [] if not segs or not scenes: continue # ★세계 사실이 **성해야** 한다 — 임의 글자수 문턱이 아니라 # production builder 의 판정을 그대로 쓴다 (Codex C-2). from app.core.world_context import grounding_world_facts_ready if not grounding_world_facts_ready(Wcp): continue # ★★CP 가 **completed** 인지도 명시로 본다 (Codex NON-BLOCK). if not all(_completed(base / k / "manifest.json") for k in ("scene_save", "shot_validator", "visual_world_rules")): continue # ★묶는 규칙은 **한 곳**이다 — 여기 다시 적으면 preflight 가 세는 # 구간과 실제로 굽는 구간이 갈린다. bundles = bundle_scenes( [{"idx": int(sg.get("scene_index") or 0), "length": len(sg.get("text") or "")} for sg in segs]) if len(bundles) < 2: # ★구간 2개 이상 continue shots, sizes = [], [] try: for b in bundles: cat = sc.build_catalog(scenes, [f"scene-{i}" for i in b]) shots.append(len(cat)) sizes.append(len(json.dumps(cat, ensure_ascii=False) .encode("utf-8"))) except ValueError: # 겹친 샷 ID 등 continue # base = projects//checkpoints/episodes/ out.append({ "project_id": base.parent.parent.parent.name, "episode_id": base.name, "chunks": len(bundles), "bundles": bundles, "shots_median": st.median(shots) if shots else 0, "bytes_median": st.median(sizes) if sizes else 0, "shots_max": max(shots) if shots else 0, "chars": sum(len(s.get("text") or "") for s in segs), }) return out def shot_image_counts() -> Dict[str, int]: """에피소드 → **샷에 걸린** 이미지 수. ★못 읽으면 빈 표. ★`episode_id` 로만 세면 평면도·다른 단계 산출까지 세어진다. 사람이 「이 대상이 저 샷에 있나」를 보려면 **그 샷의 그림**이어야 하므로 `still_id` 로 잇는다. """ try: from sqlalchemy import text from app.core.database import SessionLocal with SessionLocal() as db: return dict(db.execute(text(""" SELECT ss.episode_id, count(*) FROM image_asset ia JOIN scene_still ss ON ia.still_id = ss.id GROUP BY 1""")).fetchall()) except Exception as exc: # noqa: BLE001 print(f" ★이미지 수를 못 읽었다 ({exc}) — **없다는 뜻이 아니다**") return {} def choose(cands: List[Dict[str, Any]], *, require_shot_images: bool = True ) -> Tuple[Optional[Dict[str, Any]], Dict[str, Any]]: """**중앙 구조**에 가장 가까운 하나. ★동률은 에피소드 id 순. ★★`require_shot_images` 가 기본 True 다 (2026-08-31, 사용자 지적). 앞 판은 이 조건이 **없어서** 샷 이미지가 한 장도 없는 에피소드를 골랐다. 그 결과 사람 검토 화면이 **글만** 있는 판이 됐고, 「이 대상이 저 샷에 있나」를 샷 설명 글로만 보게 됐다 — 시각 검토가 성립하지 않는다. ★실측(2026-08-31): 후보 55개 중 **47개가 이미 샷 이미지를 갖고 있다.** 그래서 새 시나리오를 지을 필요가 없다 — 고르는 기준만 고치면 된다. """ if not cands: return None, {} if require_shot_images: imgs = shot_image_counts() with_img = [c for c in cands if imgs.get(c["episode_id"], 0) > 0] if not with_img: # ★**조용히 아무거나 고르지 않는다.** 시각 검토가 안 되는 판을 # 「됐다」로 넘기면 사람이 볼 것이 없다. raise LookupError( "샷 이미지가 있는 후보가 하나도 없다 — 시각 검토용 에피소드를 " "먼저 만들어야 한다. 아무거나 고르지 않는다") for c in with_img: c["shot_images"] = imgs.get(c["episode_id"], 0) cands = with_img mid = { "chunks": st.median([c["chunks"] for c in cands]), "shots_median": st.median([c["shots_median"] for c in cands]), "bytes_median": st.median([c["bytes_median"] for c in cands]), } def dist(c): # ★척도가 다르므로 **중앙값 대비 비율**로 잰다 — bytes 가 지배하지 않게 return sum(abs(c[k] - mid[k]) / (mid[k] or 1) for k in mid) ranked = sorted(cands, key=lambda c: (round(dist(c), 6), c["episode_id"])) mid["candidates_with_shot_images"] = len(cands) return ranked[0], mid def owner_coverage(pid: str, eid: str) -> Dict[str, Any]: """이 **에피소드**의 갈래 coverage. ★기존 구조화 SOT 로만 본다. ★앞 판은 `psql` 을 subprocess 로 부르고 host·user·db·비밀번호를 코드에 박고 SQL 을 f-string 으로 이었으며, **`eid` 를 아예 안 썼다** — project 전체를 셌다. 지금 고른 project 가 에피소드 하나뿐이라 우연히 맞았을 뿐, 여러 판을 돈 project 에서는 **거짓**이 된다 (Codex 2026-08-31). → `SessionLocal` + ORM 으로 `entity_episode_link` ↔ `entity_canon` 을 잇고 **project_id 와 episode_id 를 둘 다** 건다. 실패하면 unknown. ★산문·이름을 읽지 않는다. """ try: from sqlalchemy import func from app.core.database import SessionLocal from app.models.project import EntityCanon, EntityEpisodeLink except Exception as exc: # noqa: BLE001 return {"known": False, "why": f"모델을 못 읽었다: {exc}"} db = None try: db = SessionLocal() rows = (db.query(EntityCanon.entity_type, func.count(EntityCanon.id)) .join(EntityEpisodeLink, EntityEpisodeLink.canon_id == EntityCanon.id) .filter(EntityCanon.project_id == pid, EntityEpisodeLink.project_id == pid, EntityEpisodeLink.episode_id == eid) .group_by(EntityCanon.entity_type) .all()) except Exception as exc: # noqa: BLE001 return {"known": False, "why": f"조회 실패: {exc}"} finally: if db is not None: db.close() got = {str(t): int(n) for t, n in rows} if not got: return {"known": False, "why": "이 에피소드에 연결된 canon 이 없다"} return {"known": True, "by_type": got, "note": ("★`location_part` 는 canon 갈래가 없어 이 수에 안 잡힌다. " "없는 갈래는 **coverage debt** 로 남긴다")} def main() -> int: seal_outbound() from app.core import openai_keys from app.modules.pipeline import grounding_chunk as gc from app.modules.pipeline import grounding_shot_catalog as sc cands = candidates() pick, mid = choose(cands) P = print P(f"■ C preflight — shot-aware 유료 canary. ★아직 **안 샀다**") P(f" 후보 {len(cands)} 에피소드 (구조만 보고 걸렀다)") if not pick: P(" ★후보가 없다 — 조건을 만족하는 에피소드가 없다") return 2 P(f" 전수 중앙 구조 구간 {mid['chunks']} · 구간당 샷 " f"{mid['shots_median']} · bytes {mid['bytes_median']:,.0f}") P() P("── §1 고른 에피소드와 **까닭**") P(f" project {pick['project_id']}") P(f" episode {pick['episode_id']}") P(f" 구간 {pick['chunks']} · 구간당 샷(중앙) {pick['shots_median']} · " f"최대 {pick['shots_max']} · bytes(중앙) {pick['bytes_median']:,.0f} · " f"원문 {pick['chars']:,}자") P(" ★고른 기준은 **구조뿐**이다 — 전수 분포의 중앙에 가장 가깝고, 동률은") P(" 에피소드 id 순. **원고 내용·이름·산문을 한 줄도 안 읽었다.**") cov = owner_coverage(pick["project_id"], pick["episode_id"]) if cov.get("known"): P(f" 갈래 coverage(기존 SOT) {cov['by_type']}") P(f" {cov['note']}") else: P(f" 갈래 coverage **unknown** — {cov.get('why')}") P(" ★산문·이름을 읽어 고르지 않는다. 없는 갈래는 **coverage debt**.") P() # ── 실제 payload 와 획득 신원 ── base = PROJECTS / pick["project_id"] / "checkpoints" / "episodes" \ / pick["episode_id"] S = _load(base / "scene_save" / "manifest.json") or {} V = _load(base / "shot_validator" / "manifest.json") or {} Wcp = _load_cp(base / "visual_world_rules" / "manifest.json") segs = {f"scene-{s['scene_index']}": (s.get("text") or "") for s in S.get("segments") or []} # ★★production builder 한 벌 — preflight · C runner · D step 이 **같은 # callable** 을 쓴다. JSON 직렬화로 canary 를 사면 D 가 다른 입력을 쓰게 # 되어 canary 가 production 대표가 아니다 (Codex C-2). from app.core.world_context import build_grounding_world_facts world = build_grounding_world_facts(Wcp) phys = rr.physical_model() slots = int(openai_keys.slot_count()) P("── §2 호출 그래프와 **획득 신원**") P(f"{'호출':22} {'샷':>4} {'payload bytes':>14} {'신원':>26}") P("─" * 74) tot = 0 for n, b in enumerate(pick["bundles"]): ids = [f"scene-{i}" for i in b] cat = sc.build_catalog(V.get("scenes") or [], ids) p = gc.build_chunk_payload(ids, segs, world, shot_catalog=cat) by = len(p["parts"][0]["text"].encode("utf-8")) tot += by ident = gc.acquisition_identity( p, model_alias=rr.MODEL_ALIAS, model_physical=phys, request_contract=rr.PINNED) P(f"{('c%d ' % n) + ','.join(ids):22} {len(cat):>4} {by:>14,} " f"{ident:>26}") P(f"{'merge (구간 산출 뒤)':22} {'-':>4} {'-':>14} {'★주행 때':>26}") P("─" * 74) S_n = pick["chunks"] P(f" 구간 payload 합계 {tot:,} bytes") P() P("── §3 상한") P(f" 논리 구간 {S_n} + merge 1 = **{S_n + 1}**") P(f" dispatch {S_n + 1}") P(f" ★키 슬롯 {slots} — `llm_client.py:427-439` loop 는 Router 밖이라") P(f" `num_retries=0` 이 안 닫는다") P(f" 물리 상한 ({S_n} + 1) × {slots} = **{(S_n + 1) * slots}**") P(f" 요청 계약 {json.dumps(rr.PINNED, ensure_ascii=False)} " "(재시도 0 · tier fallback 끔)") P(" 검색 0 · 이미지 생성 0 · 이미지 검색 0") P() P("── §4 자동으로 볼 것 / 사람만 볼 것") P(" 자동 — schema/runtime enum · 씬 제약 · 격리 · 장부·재개 · Opik/provider 대조") P(" ★사람만 — 샷 ID 가 **의미상** 맞나 · 두 축(hard·notice) · 엔티티·facet 품질") P(" ★VLM 평가 금지 · 고정 대상/작품 이름 프롬프트 금지") P(" 검토 화면(`veto_review`)에 한 화면으로 낸다 —") P(" 원문 인용 + 고른 샷 ID + **각 샷 전문 description** + owner/관계") P() P("── §5 이 주행의 지위") P(" ★**shot-aware 모델 경로 가능성 진단**이다. 다섯 갈래 완료도,") P(" production PASS 도 **아니다**.") P(" ★고른 에피소드에 없는 갈래는 **coverage debt** 로 남기고 뒤에서 닫는다.") P(" ★`location_part` 는 계속 durable debt 다.") return 0 if __name__ == "__main__": raise SystemExit(main())