#!/usr/bin/env python3 """#108 검증 시나리오 acceptance — **주행 전에 고정한다** (2026-08-29). 원고 = **3씬** `#1 버스 A` · `#2 다른 버스 B` · `#3 버스 A 재방문`. lane 축은 **out of scope**(08-25 실물로 도달 증명됨) — n/a 를 합격에 안 섞는다. ## 실물에서 확인한 키만 쓴다 scene_save `data.segments` scene_director `data.scenes[].primary_location` shot_selection `data.scenes[].selected_shot_indices` background_share_plan `data.plan.share_groups[].shot_tags` · `data.plan.shot_plans[tag].{ref_plan,prev_anchor_tag}` era 체크포인트 **없음** — recipe `records.json` 의 `era_assess::` / `era_ref::` / `era_fail::` ★Opik span **이름**이 아니라 **`tags` 의 `op:` 축**이다. 실물 이름은 `gemini-3.7-flash_chat.completion_…` 이고 신원은 `op:era_research_plate` 등에 있다. (2026-08-29 Codex 지적 — 이름으로 세면 **거짓 0**) ## 축 ① scene split == 3 ② `#1==#3` · `#2!=#1` (era 캐시 신원의 전제) ③ era record nonempty assess **2/2** · `era_ref == nonempty assess` · identity=canonical · `identity_fallback` 0 ④ share_plan S1+S3 한 group · 그 group 에 S2 없음 · 어떤 group 도 섞지 않음 ★S3 의 prev anchor 는 **진단값**이지 gate 가 아니다 (08-25 fixture 도 S3sh1 은 `background` 였다) ★era 와 **직교**한다 — 여기가 깨져도 ③⑤ 를 연쇄 미확정으로 만들지 않는다 ⑤ cache-miss 사슬 Opik assess distinct trace **2** · pick **2** · 논리 web_search **2** — 총계로만 건다 pick 은 별도 수 — assess 와 **합산 금지** raw provider span 수는 fallback/retry **진단**으로만 ★**씬 번호로 추론하지 않는다** — 「씬3 이 샀으면 실패」는 씬1 이 먼저 돈다는 전제를 깔았고, 상류가 죽어 씬3 이 그 scope 를 **처음** 사면 정상 cache miss 를 실패로 찍었다 (2026-08-29 실측 · Codex 판정). 씬별 계수는 **진단**으로만. 같은 scope 를 두 번 사면 총계가 3 이상이 되어 거기서 잡힌다. ★③이 온전히 통과할 때만 gate — ③이 깨지면 분모가 성립하지 않으므로 **미확정**이지 어긋남이 아니다 ★「물리 호출」이라 부르지 않는다 — SDK 내부 전송은 한 행 안에 숨는다. usage: read_108_acceptance.py [--since ISO] [--until ISO] """ from __future__ import annotations import json import pathlib import sys HERE = pathlib.Path(__file__).resolve() sys.path.insert(0, str(HERE.parents[2])) # backend sys.path.insert(0, str(HERE.parents[3])) # repo root (tools.opik_...) sys.path.insert(0, str(HERE.parents[3] / "scratchpad")) OK, NG, NA, OUT = "✓", "★", "·", "—" N_SCENES = 3 EXPECT_SCOPES = 2 # 버스 A · 버스 B #: 막차는 두 장소가 **모두 버스 안**이다 EXPECT_ROLE = "location_interior" def _cp(base: pathlib.Path, step: str): f = base / step / "manifest.json" return (json.loads(f.read_text()).get("data") or {}) if f.is_file() else None def _op_tags(row) -> set: return {t[len("op:"):] for t in (row.get("tags") or []) if isinstance(t, str) and t.startswith("op:")} def _era_calls(eid: str, since: str, until: str): """Opik 에서 era assess/pick 을 **`op:` 태그 정확 일치**로 센다. ★span 이름으로 찾으면 안 된다 — 실물 이름은 모델명이고 era 신원은 `tags` 의 `op:era_research_*` 에 있다. trace 는 `metadata.episode_id` 정확 일치로 거른다. 못 붙거나 창이 비면 **None** — 0 으로 안 읽는다. """ try: import _opik_env from tools.opik_prompt_audit.audit.fetch import ( fetch_spans, fetch_trace_index, ) base, ws, proj = _opik_env.opik_target() spans = fetch_spans(base, ws, proj, since, until) tidx = fetch_trace_index(base, ws, proj, since, until) mine = {str(tid) for tid, t in tidx.items() if str(((t.get("metadata") or {}).get("episode_id")) or "") == eid} # ★gate 는 **distinct 부모 trace 수**다 — raw span 은 fallback/retry 로 # 늘어나므로 진단으로만 쓴다 (2026-08-29 Codex 지적). # ★scene_index 는 span 이 아니라 **부모 trace** 의 metadata 에 있다 # (실측: plate=씬1/샷3 · groupbg+pick=씬2/샷4). trig = {"assess": set(), "pick": set()} raw = {"assess": 0, "pick": 0} by_scene = {"assess": {}, "pick": {}} for sp in spans: tid = str(sp.get("trace_id")) if tid not in mine: continue for o in _op_tags(sp): if not o.startswith("era_research_"): continue key = "pick" if o.endswith("_pick") else "assess" raw[key] += 1 trig[key].add(tid) sc = ((tidx.get(tid) or {}).get("metadata") or {}).get("scene_index") by_scene[key][str(sc)] = by_scene[key].get(str(sc), 0) + 1 return ({k: len(v) for k, v in trig.items()}, len(mine), by_scene, raw) except Exception as exc: # noqa: BLE001 return None, 0, {}, f"{exc!r}"[:120] def _web_search_rows(eid: str, since: str, until: str): """DB 의 **논리** web_search 행 — SDK 내부 재전송은 이 한 행에 숨는다. 그래서 「물리 전송 수」라 부르지 않는다 (2026-08-29 Codex). """ try: sys.path.insert(0, str(HERE.parent)) import _db rows = _db.rows( """SELECT metadata_json FROM llm_call_log WHERE episode_id = :e AND operation_type = 'web_search' AND step_name = 'scene_image_pipeline' AND created_at >= :s AND created_at < :u""", {"e": eid, "s": since, "u": (until if until != "9999" else "9999")}) n3 = 0 for (mj,) in rows: try: m = json.loads(mj) if isinstance(mj, str) else (mj or {}) except Exception: m = {} if str(m.get("scene_index")) == "3": n3 += 1 return {"total": len(rows), "scene3": n3} except Exception: return None def main() -> int: argv = [a for a in sys.argv[1:] if not a.startswith("--")] if len(argv) < 2: print(__doc__); return 2 pid, eid = argv[0], argv[1] since = next((a.split("=", 1)[1] for a in sys.argv if a.startswith("--since=")), "2026-08-01T00:00:00") until = next((a.split("=", 1)[1] for a in sys.argv if a.startswith("--until=")), "9999") from app.core.config import settings root = pathlib.Path(settings.projects_dir) / pid base = root / "checkpoints" / "episodes" / eid rec_f = root / "images" / eid / "scene" / "recipe" / "records.json" if not base.is_dir(): raise SystemExit(f"{NG} 체크포인트 폴더를 못 찾았다: {base}\n" " 「없다」가 아니라 **내 조회로 못 찾았다** 로 읽어라") bad = unconf = 0 # ① scene split segs = ((_cp(base, "scene_save") or {}).get("segments") or []) ok1 = len(segs) == N_SCENES print(f"① scene split {len(segs)}개 " + (OK if ok1 else f"{NG} {N_SCENES} 아님")) bad += 0 if ok1 else 1 # ② primary_location loc = {s.get("scene_index"): s.get("primary_location") for s in ((_cp(base, "scene_director") or {}).get("scenes") or [])} print(f"② primary_location {loc or '(못 찾았다)'}") same = bool(loc.get(1)) and loc.get(1) == loc.get(3) diff = bool(loc.get(2)) and loc.get(2) != loc.get(1) print(f" #1==#3 {OK if same else NG} · #2!=#1 {OK if diff else NG}") gate = same and diff bad += 0 if gate else 1 if not gate: print(f" {NG} ★캐시 신원의 전제 — 깨지면 ③⑤ 는 미확정") # ③ era record rec = json.loads(rec_f.read_text()) if rec_f.is_file() else {} if not rec_f.is_file(): print(f"③ {NG} recipe records.json 없음: {rec_f}"); bad += 1 A = {k: v for k, v in rec.items() if k.startswith("era_assess::")} R = [k for k in rec if k.startswith("era_ref::")] F = [k for k in rec if k.startswith("era_fail::")] nonempty = [k for k, v in A.items() if isinstance(v, dict) and (v.get("subjects") or [])] fb = [k for k, v in A.items() if isinstance(v, dict) and v.get("identity_fallback")] print(f"③ era assess {len(A)} (nonempty {len(nonempty)}) · " f"ref {len(R)} · fail {len(F)} · identity_fallback {len(fb)}") # ★payload 의 canonical triple 을 직접 본다 (2026-08-29 Codex BLOCK-1). # 실물 키: identity · scope_id · scope_role · scope_sha · subjects ids = {k: (v.get("identity") if isinstance(v, dict) else None) for k, v in A.items()} trip = {k: ((v.get("scope_id"), v.get("scope_role"), v.get("scope_sha")) if isinstance(v, dict) else None) for k, v in A.items()} scope_ids = {t[0] for t in trip.values() if t} roles = {t[1] for t in trip.values() if t} want_ids = {loc.get(1), loc.get(2)} - {None} print(f" identity={sorted(set(map(str, ids.values())))} · " f"scope_id={sorted(map(str, scope_ids))} · role={sorted(map(str, roles))} · " f"unique triple {len({t for t in trip.values() if t})}") ok3 = False if not gate: print(f" {NA} ②가 깨져 미확정"); unconf += 1 else: e3 = [] if len(A) != EXPECT_SCOPES: e3.append(f"assess 행 {len(A)} != {EXPECT_SCOPES}") if len(nonempty) != EXPECT_SCOPES: e3.append(f"nonempty {len(nonempty)} != {EXPECT_SCOPES}") if len(R) != len(nonempty): e3.append(f"era_ref {len(R)} != nonempty {len(nonempty)}") if fb: e3.append(f"identity_fallback {len(fb)}건") if set(ids.values()) != {"canonical"}: e3.append(f"identity 가 전부 canonical 이 아니다: " f"{sorted(set(map(str, ids.values())))}") if len({t for t in trip.values() if t}) != EXPECT_SCOPES: e3.append("unique (scope_id,role,sha) 가 2개가 아니다") if want_ids and scope_ids != want_ids: e3.append(f"scope_id {sorted(map(str,scope_ids))} != " f"기대 {sorted(map(str,want_ids))}") if any(not (t and t[2]) for t in trip.values()): e3.append("scope_sha 가 빈 행이 있다") # ★막차는 두 장소가 **모두 버스 안**이다 — 하나라도 exterior 면 # 잘못 분류된 것이고, 그러면 캐시 신원이 갈려 #97 이 무의미해진다 # (2026-08-29 Codex). 출력만 하고 안 걸렀던 자리. if roles - {EXPECT_ROLE}: e3.append(f"scope_role 이 {EXPECT_ROLE} 가 아닌 행: " f"{sorted(roles - {EXPECT_ROLE})}") if F: e3.append(f"era_fail {len(F)}건 — fresh 주행에서는 실패다") print(f" {NG + ' ' + ' / '.join(e3) if e3 else OK}") bad += 1 if e3 else 0 ok3 = not e3 # ④ share_plan — ★era 와 직교. 깨져도 연쇄 미확정 안 만든다 plan = ((_cp(base, "background_share_plan") or {}).get("plan") or {}) groups = plan.get("share_groups") or [] plans = plan.get("shot_plans") or {} sc = lambda t: int(str(t).split("sh")[0].lstrip("S")) if "sh" in str(t) else None g_of = {t: g.get("group_key") for g in groups for t in (g.get("shot_tags") or [])} S = {n: {t for t in g_of if sc(t) == n} for n in (1, 2, 3)} # ★acceptance 도 자기 전제를 직접 확인한다 — selected tag 와 share plan # tag 가 서로 덮는지 (2026-08-29 Codex NON-BLOCK). sel = set() for x in ((_cp(base, "shot_selection") or {}).get("scenes") or []): for i in x.get("selected_shot_indices") or []: sel.add(f"S{x.get('scene_index')}sh{i}") cov = sel - set(g_of) extra = set(g_of) - sel print(f"④ share_plan group {len(groups)}개 · selected {len(sel)}개") print(f" selected 인데 plan 에 없음 {sorted(cov) or '없음'} · " f"plan 에만 있음 {sorted(extra) or '없음'}") # ★extra 도 건다 — 「닫았다」고 보고했으면 둘 다 걸어야 한다 (Codex) if cov or extra: print(f" {NG} selected 와 share plan tag 집합이 안 맞는다"); bad += 1 if not groups: print(f" {NA} group 을 못 찾았다 — 미확정"); unconf += 1 else: merged = [g.get("group_key") for g in groups if ({t for t in (g.get("shot_tags") or []) if sc(t) == 2} and {t for t in (g.get("shot_tags") or []) if sc(t) in (1, 3)})] together = [g.get("group_key") for g in groups if ({t for t in (g.get("shot_tags") or []) if sc(t) == 1} and {t for t in (g.get("shot_tags") or []) if sc(t) == 3})] print(f" S1+S3 한 group {OK if together else NG} {together}") print(f" S2 를 섞은 group {OK + ' 없음' if not merged else NG + ' ' + str(merged)}") bad += 0 if (together and not merged) else 1 anch = {t: (plans.get(t) or {}) for t in plans if sc(t) == 3} print(f" {OUT} 진단(gate 아님) S3 ref_plan: " + " · ".join(f"{t}={v.get('ref_plan')}" + (f"→{v.get('prev_anchor_tag')}" if v.get("prev_anchor_tag") else "") for t, v in sorted(anch.items()))) # ⑤ cache-miss 사슬 n, n_tr, by_scene, extra = _era_calls(eid, since, until) if n is None: print(f"⑤ {NA} Opik 조회 실패({extra}) — **미확정**"); unconf += 1 elif n_tr == 0: print(f"⑤ {NA} 이 에피소드의 trace 가 창({since}~{until})에 **없다** — " f"0 을 「호출 없음」으로 안 읽는다. **미확정**"); unconf += 1 else: print(f"⑤ cache-miss 사슬 trace {n_tr}건 · " f"assess **{n['assess']}** · pick **{n['pick']}** (합산 금지)") ws = _web_search_rows(eid, since, until) print(f" {OUT} 진단(gate 아님) 씬별 assess {by_scene.get('assess', {})} · " f"씬별 pick {by_scene.get('pick', {})} · " f"web_search {ws if ws is not None else '(조회 실패)'}") if not ok3: # ★③이 깨졌으면 **분모가 성립하지 않는다.** 두 scope 가 다 안 # 구워졌는데 계수만 세면 「재방문이 캐시를 못 썼다」로 읽히는 # 거짓 어긋남이 된다 (2026-08-29 실측 — 씬1 이 상류 결함으로 # 전멸해 씬3 이 L01 을 **처음** 산 것을 실패로 찍었다). print(f" {NA} ③이 통과하지 않아 **미확정** — 계수는 진단으로만") unconf += 1 else: e5 = [] for lab, got in (("assess", n["assess"]), ("pick", n["pick"])): if got != EXPECT_SCOPES: e5.append(f"{lab} distinct trace {got} != {EXPECT_SCOPES}") if ws is None: print(f" {NA} web_search 조회 실패 — 그 축은 미확정"); unconf += 1 elif ws["total"] != EXPECT_SCOPES: e5.append(f"논리 web_search 행 {ws['total']} != {EXPECT_SCOPES}") print(f" {NG + ' ' + ' / '.join(e5) if e5 else OK + ' 두 scope 를 한 번씩만 샀다'}") bad += 1 if e5 else 0 print(f"\n(lane 축 {OUT} out of scope)") if bad: print(f"\n{NG} 어긋남 {bad}건" + (f" · 미확정 {unconf}" if unconf else "")) return 1 if unconf: print(f"\n{NA} 어긋남 0 이지만 **미확정 {unconf}건** — 통과라 쓰지 마라") return 2 print(f"\n{OK} 다 통과") return 0 if __name__ == "__main__": raise SystemExit(main())