"""provider 직전 payload 측정 — Gemini 요청이 **완전히 직렬화된 뒤 HTTP 전송 직전**. ★목적은 **provider 전송 0 상태의 payload 조립 확인**이다 (Codex 2026-09-02). 이미지 문(`reserve_current_call`) 검증이 아니다 — 여기서는 이미지 예산을 깔지 않으므로 그 문은 no-op 이다. 이미지 문은 별도 시험·2단계에서 cap 0/1 로 본다. 입구는 **public** `SceneImageService.generate_single_scene_image(still_id)` 다 — readiness 문·asset 문을 **우회하지 않는다**. 문이 서면 그 자리를 그대로 적는다 (`stopped_at`). 우회가 있으면 그것은 payload 조립 probe 지 production E2E 가 아니다 — 이 도구는 우회를 하나도 안 만든다. 0 을 잠그는 문 셋: ① socket 문 — 바깥 주소로 connect 하면 즉시 예외 (DB 는 localhost 라 지난다) ② urlopen sentinel — `gemini_image_client` 가 `urllib.request.urlopen` 을 부르는 순간 요청 body 를 캡처하고 **즉시 멈춘다**. 가짜 bytes 를 돌려주지 않는다 — 돌려주면 그 뒤 scene LVM/readback 이 돌아 「VLM 0」이 깨진다 ③ 글 예산 cap 0 — 이미지 요청 앞에 글/VLM 호출이 있으면 provider 앞에서 거절되고 그 사실이 기록된다(우회 아님 · 그대로 보고) 캡처하는 것: 순서 있는 parts — text 는 앞 120자, inline 은 mime·bytes·sha256. 기대값은 그 샷의 render prompt card 에 사이드카가 적은 멤버(`grounding_bundle_members`) 의 `content_sha256` — production 이 적은 것을 production 함수로 읽는다. """ from __future__ import annotations import argparse import base64 import hashlib import json import os import socket import sys from datetime import datetime, timezone, timedelta from pathlib import Path from typing import Any, Dict, List, Optional KST = timezone(timedelta(hours=9)) PROBE_CONTRACT_VERSION = "1.202609021500" LOCAL_HOSTS = ("127.0.0.1", "::1", "localhost") class ProbeStop(RuntimeError): """캡처가 끝났다 — 여기서 **즉시** 멈춘다. 전송 0.""" class OutboundBlocked(RuntimeError): """바깥 주소로 나가려 했다 — socket 문.""" def _is_local(addr: Any) -> bool: try: host = addr[0] if isinstance(addr, (tuple, list)) else str(addr) except Exception: # noqa: BLE001 return False return str(host).lower() in LOCAL_HOSTS def install_socket_door(record: List[Dict[str, Any]]): """바깥 connect 를 막는다. ★localhost 만 지난다(DB).""" real_connect = socket.socket.connect def _connect(self, addr, *a, **k): if not _is_local(addr): record.append({"blocked": str(addr)}) raise OutboundBlocked(f"바깥 주소 {addr!r} — 이 probe 는 전송 0 이다") return real_connect(self, addr, *a, **k) socket.socket.connect = _connect # type: ignore[assignment] return lambda: setattr(socket.socket, "connect", real_connect) def _parts_of(body: Dict[str, Any]) -> List[Dict[str, Any]]: out: List[Dict[str, Any]] = [] for c in body.get("contents") or (): for i, p in enumerate(c.get("parts") or ()): inline = p.get("inline_data") or p.get("inlineData") if inline: raw = base64.b64decode(str(inline.get("data") or "")) out.append({"order": i, "kind": "inline", "mime": inline.get("mime_type") or inline.get("mimeType"), "bytes": len(raw), "sha256": hashlib.sha256(raw).hexdigest()}) else: text = str(p.get("text") or "") out.append({"order": i, "kind": "text", "chars": len(text), "head": text[:120], # ★A/B 대조 — 두 arm 의 글이 byte 로 같은지 (전문은 안 남긴다) "sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(), # ★A/B 전문 대조 때만(env) — 기본은 안 남긴다 **({"full_text": text} if os.environ.get("CANARY_PROBE_FULL_TEXT") == "1" else {})}) return out def install_urlopen_sentinel(captured: List[Dict[str, Any]]): """`urllib.request.urlopen` 을 갈아 끼운다 — 캡처 뒤 **즉시** 예외.""" import urllib.request as _ur real = _ur.urlopen def _sentinel(req, *a, **k): url = getattr(req, "full_url", str(req)) # ★키가 붙은 query 는 적지 않는다 — host 와 path 만 from urllib.parse import urlsplit u = urlsplit(url) data = getattr(req, "data", None) or b"" try: body = json.loads(bytes(data).decode("utf-8")) except Exception: # noqa: BLE001 body = {} captured.append({ "host": u.netloc, "path": u.path, "method": getattr(req, "get_method", lambda: "?")(), "body_bytes": len(data), "parts": _parts_of(body) if isinstance(body, dict) else [], "generation_config": (body.get("generationConfig") or body.get("generation_config") if isinstance(body, dict) else None), }) raise ProbeStop("요청을 캡처했다 — 전송 0 으로 멈춘다") _ur.urlopen = _sentinel return lambda: setattr(_ur, "urlopen", real) def expected_members(project_id: str, episode_id: str, scene_index: int, shot_index: Optional[int]) -> List[Dict[str, Any]]: """그 샷의 render prompt card 에 **production 이 적은** 사이드카 멤버.""" from app.modules.pipeline import grounding_reference_bundle as rb from app.services.scene_generation_coordinator import lookup_render_prompt_card rpc = lookup_render_prompt_card(project_id=project_id, episode_id=episode_id, scene_index=scene_index, shot_index=shot_index) out = [] for m in (rpc or {}).get(rb.RPC_MEMBERS_KEY) or (): out.append({"subject_final_id": m.get("subject_final_id"), "purpose": m.get("purpose"), "outcome": m.get("outcome"), "member_identity": m.get("member_identity"), "content_sha256": m.get("content_sha256")}) return out def run(*, run_id: str, project_id: str, episode_id: str, scene_index: int, shot_index: int, actor_id: str = "canary-exit-probe") -> Dict[str, Any]: from tools.grounding_audit import canary_isolation as ci from tools.grounding_audit.canary_text_budget import canary_text_scope ci.assert_engine_is_canary(run_id) # ★실제 엔진이 canary 인지 — 먼저 from app.core.database import SessionLocal # ★ORM 모델 등록은 production 의 정식 입구 `register_models()` 로 (Codex 2026-09-02) — # DDL/DML 없이 모델만 등록한다. 한 모델만 올리면 `entity_canon.project_id` 가 가리키는 # `project_registry` 를 mapper 가 못 엮는다(실측 · probe 프로세스만의 문제). from app.core.database import register_models register_models() from app.models.project import SceneStill from app.services.scene_image_service import SceneImageService db = SessionLocal() still = (db.query(SceneStill) .filter(SceneStill.episode_id == episode_id, SceneStill.scene_index == scene_index, SceneStill.shot_index == shot_index).first()) if still is None: return {"contract": PROBE_CONTRACT_VERSION, "stopped_at": "still.not_found", "why": f"씬 {scene_index} 샷 {shot_index} 의 still 이 DB 에 없다 — " "scene_detail sync 전이다", "captured": [], "provider_calls": 0} expected = expected_members(project_id, episode_id, scene_index, shot_index) captured: List[Dict[str, Any]] = [] blocked: List[Dict[str, Any]] = [] restore_sock = install_socket_door(blocked) restore_url = install_urlopen_sentinel(captured) stopped_at, err = None, None text_budget: Optional[Dict[str, Any]] = None try: with canary_text_scope(cap=0) as tb: # ★글/VLM 은 provider 앞에서 거절 try: SceneImageService(db, project_id, actor_id).generate_single_scene_image( str(still.id)) stopped_at = "returned_without_send" except ProbeStop: stopped_at = "urlopen_sentinel" except OutboundBlocked as e: stopped_at, err = "socket_door", str(e) except Exception as e: # noqa: BLE001 stopped_at, err = f"gate:{type(e).__name__}", str(e)[:400] text_budget = dict(tb.snapshot()) finally: restore_url(); restore_sock() exp_sha = {m["content_sha256"] for m in expected if m.get("content_sha256")} got_sha = {p["sha256"] for c in captured for p in c["parts"] if p["kind"] == "inline"} return { "contract": PROBE_CONTRACT_VERSION, "measured_kst": datetime.now(KST).isoformat(timespec="seconds"), "entry": "SceneImageService.generate_single_scene_image (public · 우회 0)", "still": {"id": str(still.id), "scene_index": scene_index, "shot_index": shot_index, "visible": still.visible_entities_json}, "stopped_at": stopped_at, "error": err, "text_budget": text_budget, "socket_blocked": blocked, "captured": captured, "expected_members": expected, "verdict": { "provider_send": 0, "expected_sha_in_request": sorted(exp_sha & got_sha), "expected_sha_missing": sorted(exp_sha - got_sha), "other_inline_in_request": len(got_sha - exp_sha), "★means": ("`urlopen_sentinel` 에서 섰고 expected_sha_missing 이 [] 이면 사이드카 참조가 " "실제 요청 parts 에 실린 것. 다른 자리에서 섰으면 그 문이 먼저 선 것이다 — " "우회하지 않는다"), }, } def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--run-id", required=True) ap.add_argument("--project-id", required=True) ap.add_argument("--episode-id", required=True) ap.add_argument("--scene", type=int, required=True) ap.add_argument("--shot", type=int, required=True) ap.add_argument("--out", required=True) a = ap.parse_args() from tools.grounding_audit import canary_bootstrap as cbs, canary_isolation as ci os.environ.update(cbs.prepare_env(a.run_id)) ci.assert_app_modules_not_loaded() got = run(run_id=a.run_id, project_id=a.project_id, episode_id=a.episode_id, scene_index=a.scene, shot_index=a.shot) out = Path(a.out); out.mkdir(parents=True, exist_ok=True) p = out / f"exit_probe_s{a.scene}_sh{a.shot}.json" p.write_text(json.dumps(got, ensure_ascii=False, indent=2), encoding="utf-8") v = got.get("verdict") or {} print(f"■ exit probe · 씬 {a.scene} 샷 {a.shot} · 섰다: {got.get('stopped_at')} " f"· 전송 {v.get('provider_send', 0)} · 캡처 {len(got.get('captured') or [])}") print(f" 기대 sha 실림 {v.get('expected_sha_in_request')} · 빠짐 {v.get('expected_sha_missing')} " f"· 그 밖의 inline {v.get('other_inline_in_request')}") if got.get("error"): print(f" 문: {got['error'][:200]}") print(f" 적었다: {p}") ok = got.get("stopped_at") == "urlopen_sentinel" and not v.get("expected_sha_missing") return 0 if ok else 1 if __name__ == "__main__": sys.path.insert(0, str(Path(__file__).resolve().parents[2])) raise SystemExit(main())