"""production 경로 카나리 — 실험 스크립트가 아니라 **실제 코드**로 몇 샷. 실험(`opus_pilot.py`)은 자체 프롬프트·자체 합의 규칙을 쓴다. 그래서 스틸 판정 팩이 실제로 판정에 실리는지, 판정 정책(조건부 이중 / G+Q 동시)이 production 경로에서 도는지, 재판정·critique 가 배선돼 있는지는 **실험으로 확인되지 않는다.** 여기서는 `run_multiroll_select` 를 still_recipe_service 와 **같은 방식으로** 조립해 태운다. 바꾸는 것은 하나뿐이다 — 롤 생성 `gen_fn` 이 이미 있는 롤 이미지를 돌려준다(유료 재생성 없음). 판정·결함·수정·재판정은 전부 실제로 돈다. ★범위: **실행·record 검증**이다 — 프로덕션 지문(extra_fingerprint 의 GQ 모델·정책·팩 내용 스탬프)은 still_recipe_service 가 조립하므로 여기서는 검증되지 않는다(Codex 재리뷰 HIGH). 지문 이동은 완주 스텝 resume 실측이 담당한다(#77 JIT 경로 — 8/9 검증됨). 사용: .venv/bin/python canary_production.py [--stems S13sh3,S18sh5,...] """ from __future__ import annotations import argparse import json import shutil import sys from pathlib import Path from typing import Any, Dict, List sys.path.insert(0, str(Path(__file__).resolve().parent)) from app.core.config import settings # noqa: E402 from app.modules.pipeline.multiroll_gemini import ( # noqa: E402 GQ_DISAGREE_MARGIN, QWEN_JUDGE_MODEL, SELECT_DUAL_MARGIN_SKIP, STILL_FIX_REJUDGE_HEADER_PACK_VERSION, STILL_JUDGE_PACK_VERSION, load_fix_rejudge_header, make_gemini_critique_fn, make_gemini_judge_fn, make_gq_critique_fn, make_nb2_gen_fn, resolve_judge_pack_version, resolve_judge_texts, resolve_select_judge_models, resolve_select_judge_model_physical, ) from app.modules.pipeline.multiroll_select import ( # noqa: E402 build_critique_schema, build_judge_schema, roll_labels, run_multiroll_select, ) PROJ = "e716bafb-24bb-42b7-aea0-fdb383844ee8" EPI = "d6a9aa85-b75e-400c-980c-4ee7e876a15b" ROOT = Path(__file__).resolve().parent.parent RECIPE = ROOT / f"projects/{PROJ}/images/{EPI}/scene/recipe" OUT = ROOT / "artifact/20260806_opus_pilot/canary" DEFAULT_STEMS = ["S13sh3", "S18sh5", "S62sh4"] def size_index() -> Dict[int, List[Path]]: idx: Dict[int, List[Path]] = {} for p in (ROOT / f"projects/{PROJ}").rglob("*.png"): idx.setdefault(p.stat().st_size, []).append(p) return idx def resolve_refs(entries, idx): out = [] for e in entries or []: label, path = e.get("label", ""), str(e.get("path", "")) if path.startswith(" None: ap = argparse.ArgumentParser() ap.add_argument("--stems", default=",".join(DEFAULT_STEMS)) args = ap.parse_args() stems = [s.strip() for s in args.stems.split(",") if s.strip()] OUT.mkdir(parents=True, exist_ok=True) records = json.loads((RECIPE / "records.json").read_text("utf-8")) idx = size_index() # ★표시는 실행과 같아야 한다 (Codex 재리뷰 HIGH — 기본 selector v6 를 # 찍으면서 실행은 v7 이던 표시 거짓, GQ ON 인데 margin-skip 문구 출력). models = resolve_select_judge_models() gq_on = QWEN_JUDGE_MODEL in models print(f"판정 팩(스틸) : " f"{resolve_judge_pack_version(STILL_JUDGE_PACK_VERSION)}") if gq_on: print(f"선정 판정 모델 : {models} (G+Q 동시 — 둘째 생략 " f"없음, 불일치 문턱 {GQ_DISAGREE_MARGIN})") else: print(f"선정 판정 모델 : {models}" f" (둘째 생략 문턱 {SELECT_DUAL_MARGIN_SKIP})") print(f"물리 모델 : {resolve_select_judge_model_physical()}") print(f"재판정 ON : {settings.multiroll_fix_rejudge_enabled}") print(f"critique ON : {settings.still_recipe_critique_enabled}\n") summary = [] for stem in stems: rec = records.get(stem) if not rec: print(f"{stem}: 기록 없음 — 건너뜀") continue labels = sorted(p.stem.split("_")[-1].upper() for p in RECIPE.glob(f"{stem}_[abc].png")) n = len(labels) if not n: print(f"{stem}: 후보 롤 없음 — 건너뜀") continue # still_recipe_service 와 같은 조립 # ★스틸 전용 팩 v7 + with_physics=True — 서비스와 동일 조립. # 스키마만 physics 필수로 올리고 팩을 기본(v6)에 두면 "서술 지시 # 없는 sys + physics 필수 스키마" 모순 조립이 되고(build_judge_ # schema docstring 경고), 스키마를 빼면 G+Q 1순위 리스크(Qwen × # physics 필수 스키마)를 검증하지 못한다 (Codex 리뷰 BLOCK-4). jt = resolve_judge_texts( n, judge_name="judge_still", pack_version=STILL_JUDGE_PACK_VERSION) judge_fn = make_gemini_judge_fn( judge_sys=jt["judge_sys"], judge_schema=build_judge_schema( roll_labels(n), with_physics=True), project_config=None, step_tag="canary_judge") # 서비스(still_recipe_service)와 같은 분기 — G+Q ON 이면 관찰→취합 if bool(getattr(settings, "multiroll_gq_judge_enabled", False)): critique_fn = make_gq_critique_fn( critique_schema=build_critique_schema(), project_config=None, step_tag="canary_critique") else: critique_fn = make_gemini_critique_fn( critique_sys=jt["critique_sys"], critique_schema=build_critique_schema(), project_config=None, step_tag="canary_critique") rj_texts = resolve_judge_texts( 2, judge_name="judge_still", pack_version=STILL_JUDGE_PACK_VERSION) fix_rejudge_fn = make_gemini_judge_fn( judge_sys=rj_texts["judge_sys"], judge_schema=build_judge_schema( roll_labels(2), with_physics=True), project_config=None, step_tag="canary_fix_rejudge", prompt_header=load_fix_rejudge_header( STILL_FIX_REJUDGE_HEADER_PACK_VERSION)) # ★sanitizer — 서비스(:make_shot_gen_fn)와 동일 배선. 없으면 # 시신·핏자국류 수정 호출이 SAFETY 거절에서 그대로 죽는다 # (2026-08-10 첫 G+Q 실측이 S13sh3 수정 단계에서 이걸로 죽음 — # 프로덕션은 sanitizer 1회 재시도로 살아남는 경로). from app.modules.prompt_sanitizer import PromptSanitizer fix_gen_fn = make_nb2_gen_fn( project_id=PROJ, episode_id=EPI, operation_type="canary_fix", sanitizer=PromptSanitizer(project_config=None)) # 롤 생성만 대역 — 이미 있는 롤을 그 자리에 놓는다(유료 0) def gen_fn(tag, prompt, labeled_refs, out_path: Path, _stem=stem) -> Path: lab = out_path.stem.split("_")[-1].lower() src = RECIPE / f"{_stem}_{lab}.png" if not src.exists(): raise FileNotFoundError(src) shutil.copy(src, out_path) return out_path roll_prompts = rec.get("roll_prompts") or {} refs = resolve_refs( (rec.get("roll_refs") or {}).get(labels[0]) or rec.get("refs"), idx) stem_out = OUT / stem record: Dict[str, Any] = {} try: sel, record = run_multiroll_select( tag=f"canary_{stem}", prompt=rec.get("prompt", ""), labeled_refs=refs, out_stem=stem_out, gen_fn=gen_fn, judge_fn=judge_fn, critique_fn=critique_fn, fix_gen_fn=fix_gen_fn, roll_count=n, critique_enabled=bool( settings.still_recipe_critique_enabled), fix_head=jt.get("fix_head", ""), fix_tail=jt.get("fix_tail", ""), fix_label=jt.get("fix_label", "ORIGINAL PHOTOGRAPH"), fix_rejudge_fn=fix_rejudge_fn, roll_prompts=roll_prompts or None, ) except Exception as exc: # noqa: BLE001 # 샷 단위 격리 — 한 샷의 실패(SAFETY 거절 등)가 남은 샷의 # 검증을 막으면 카나리아의 목적(경로 전수 확인)이 죽는다. summary.append({"stem": stem, "오류": f"{type(exc).__name__}: {exc}"[:300]}) print(f"{stem:9} 실패 — {type(exc).__name__}: {exc}") continue rj = record.get("fix_rejudge") or {} crit = record.get("critique") or {} gq = record.get("gq") or {} line = { "stem": stem, "이전_선정": rec.get("selected"), "카나리_선정": record.get("selected"), "판정_점수": record.get("totals"), # G+Q 합의 경로(agree/gemini_priority/combined/single_*) — # 새 체계의 핵심 신호. OFF 실행에서는 빈 값이 정상이다. # ★프로덕션 지문은 여기서 검증하지 않는다(모듈 docstring). "gq": ({"route": gq.get("route"), "gap": gq.get("gap")} if gq else "—"), "결함": len(crit.get("issues") or []), "qwen_관찰": len(crit.get("qwen_observations") or []) if "qwen_observations" in crit else "—", "수정함": "fix_prompt" in record, "재판정": (f"{rj.get('winner')} (수정본승={rj.get('fix_won')})" if rj else "—"), } summary.append(line) print(f"{stem:9} 이전={line['이전_선정']} → 카나리=" f"{line['카나리_선정']} 점수={line['판정_점수']} " f"gq={line['gq']} 결함 {line['결함']}건" f"(관찰 {line['qwen_관찰']}) 수정={line['수정함']} " f"재판정 {line['재판정']}") (OUT / "summary.json").write_text( json.dumps(summary, ensure_ascii=False, indent=1), "utf-8") print(f"\n기록: {OUT/'summary.json'}") if __name__ == "__main__": main()