"""저장 에피소드로 **호출별 입력 표**를 낸다. ★유료 0.

Codex 가 소비·판별 활성화의 선행조건으로 건 것 (2026-08-31) — 그리고 첫 판을
넷으로 BLOCK 한 것을 반영한 두 번째 판이다.

## 이 표가 재는 것 · 안 재는 것

★**범위**: `grounding_a0` 부터 **첫 이미지 검색 직전**까지의 고증 관련
호출뿐이다. 「파이프라인 전체」가 아니다. **빠진 것**: `scene_*` · `beat_*` ·
`shot_*` · `entity_relation` · `entity_filter` · `entity_detail` · `entity_t2i`
· 그 뒤 전부.

★**SOT 는 bytes** 다. token-equivalent 는 bytes/2.5 **추정**이고 모델마다
tokenizer 가 달라 「N 토큰」으로 단정하면 안 된다 (Codex BLOCK-4).

★**산출 토큰은 안 센다** — 모르는 값이다. 입력만이다.

★`grounding_screen` 은 지금 `disabled` 다. **「현재 active」와 「screen 을
켜면 더해지는 것」을 따로 합계낸다** (Codex BLOCK-2).

★v2 는 실제 프로젝트에서 **한 번도 안 돌았다**(`grounding_a0` 체크포인트
0건). 그래서 A0 후보를 **가짜 3개**로 넣는다. 그 여파는 두 갈래로 다르다 —

  **하한**            결속 catalog overlay · `merges` 지문(지금 CP 에 그 칸이
                      없어 schema 가 안 붙는다). 실제로는 **더 크다**.
  **proxy·방향 미확정** classifier·screen 의 **모집단과 payload**. 지금은
                      `a0_candidates=[]` 라 `entity_description` 을 쓰는데,
                      실제 결속 뒤에는 **원문 인용**과 승격분이 들어가
                      bytes 가 **줄 수도 늘 수도** 있다.

## 어떻게 재나

`call_structured` 를 **가로채** 나가려던 payload 를 붙잡되 **보내지 않는다**.
손으로 프롬프트를 다시 짜면 「내 도구가 프로덕션과 다른 입력을 보낸다」가
되고, 그 표로 정한 설계는 실제와 무관해진다.

    python tools/grounding_audit/call_payload_table.py <project_id> <episode_id>
"""
from __future__ import annotations

import json
import sys
from pathlib import Path
from typing import Any, Dict, List

sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

#: ★bytes → token-equivalent **추정** 나눗수. 예산 근거로 쓰면 안 된다.
_BYTES_PER_TOKEN = 2.5

ACTIVE = "active"
PLANNED = "planned"


def seal_outbound() -> None:
    """★★**최외곽 전송 경계를 잠근다.** monkeypatch 가 빗나가도 안 나간다.

    실제로 겪었다 (2026-08-31): `assess_subjects` 가 **함수 안에서**
    `call_structured` 를 import 하는데 모듈 속성만 갈아 끼웠더니 안 잡히고
    **gemini 15콜이 실제로 나갔다**(Opik `op:audit` 15건 · prompt 12,707 ·
    completion 5,093 토큰). 그때 나는 「유료 0」이라고 보고했다.

    감시가 빗나갈 수 있다는 전제로, **HTTP 를 여는 자리 자체**를 세운다.
    monkeypatch 는 여전히 하되 이것이 마지막 문이다.

    ★**되돌릴 수 있어야 한다.** 이 문은 프로세스 **전역**이라, 시험 안에서
    한 번 열면 그 뒤 모든 시험의 소켓이 막힌다 — 실제로 내 시험이 다른 파일
    네 건을 깨뜨렸다(2026-08-31). 원래 함수를 보관하고 `unseal_outbound()` 로
    되돌린다. 도구 경로는 안 되돌린다(끝까지 잠긴 채로 끝난다).
    """
    import http.client
    import socket

    global _ORIG
    if _ORIG is None:
        _ORIG = (http.client.HTTPConnection.request,
                 http.client.HTTPSConnection.request,
                 socket.socket.connect)

    def _no(*_a, **_k):
        raise RuntimeError(
            "★감사 도구는 바깥으로 못 나간다 — 가로채기가 빗나갔다. "
            "이 자리에서 서는 것이 실제로 사 버리는 것보다 낫다")

    http.client.HTTPConnection.request = _no          # type: ignore[assignment]
    http.client.HTTPSConnection.request = _no         # type: ignore[assignment]
    socket.socket.connect = _no                       # type: ignore[assignment]


_ORIG = None


def unseal_outbound() -> None:
    """★시험 전용 — 문을 되돌린다. 도구 경로에서는 부르지 않는다."""
    import http.client
    import socket

    global _ORIG
    if _ORIG is None:
        return
    (http.client.HTTPConnection.request,
     http.client.HTTPSConnection.request,
     socket.socket.connect) = _ORIG
    _ORIG = None


def _s(x: Any) -> str:
    return x if isinstance(x, str) else json.dumps(x, ensure_ascii=False)


def _b(*parts: Any) -> int:
    return sum(len(_s(p).encode("utf-8")) for p in parts)


def _sha(*parts: Any) -> str:
    """payload 신원. ★재개 때 **같은 것을 다시 사지 않으려면** 이것이 열쇠다."""
    import hashlib

    h = hashlib.sha256()
    for p in parts:
        h.update(_s(p).encode("utf-8"))
        h.update(b"\x00")
    return h.hexdigest()[:16]


def _cp(root: Path, step: str) -> Dict[str, Any]:
    p = root / step / "manifest.json"
    return json.loads(p.read_text(encoding="utf-8")) if p.exists() else {}


class Capture:
    """`call_structured` 를 가로채 payload 만 모은다. ★나가지 않는다."""

    def __init__(self, ret: Any = None) -> None:
        self.calls: List[Dict[str, Any]] = []
        self._ret = ret if ret is not None else {}

    def __call__(self, *args, **kw) -> Dict[str, Any]:
        """★★**위치 인자도 받는다.** `era_research` 는
        `call_structured(step, system, user, schema, ...)` 로 **위치**로
        부른다 — kwargs 만 받는 fake 를 세웠다가 109번 전부 TypeError 로
        죽었고, 그것이 표에 **`+0`** 으로 찍혔다. 「못 쟀다」가 「없다」로
        읽히던 그 부류다.
        """
        pos = list(args) + [None] * 4
        parts = (kw.get("system_prompt") or pos[1] or "",
                 kw.get("user_prompt") or pos[2] or "",
                 kw.get("response_schema") or pos[3] or {})
        self.calls.append({
            "bytes": _b(*parts),
            # ★**신원**은 payload 다 — 단계 이름·batch 라벨이 아니다.
            #  라벨로 나누면 같은 요청을 두 번 산다.
            "sha": _sha(*parts),
            "model": str(kw.get("model") or ""),
        })
        return dict(self._ret)


class Rows:
    def __init__(self) -> None:
        self.rows: List[Dict[str, Any]] = []

    def add(self, name: str, n: int, one_bytes: int, carries: str,
            group: str = ACTIVE, sha: str = "") -> None:
        self.rows.append({"name": name, "n": n, "bytes": one_bytes,
                          "tot": one_bytes * n, "carries": carries,
                          "group": group, "shas": [sha] if sha else []})

    def add_calls(self, name: str, calls: List[Dict[str, Any]], carries: str,
                  group: str = ACTIVE) -> None:
        """★평균×N 이 아니라 **호출마다 실제 bytes 를 더한다** (Codex BLOCK-3)."""
        if not calls:
            return
        tot = sum(c["bytes"] for c in calls)
        self.rows.append({"name": name, "n": len(calls),
                          "bytes": tot // len(calls), "tot": tot,
                          "carries": carries, "group": group,
                          "shas": [c.get("sha", "") for c in calls]})


def main() -> int:
    if len(sys.argv) < 3:
        print(__doc__)
        return 2
    pid, epi = sys.argv[1], sys.argv[2]
    from app.core.config import settings

    # ★import 가 끝난 **뒤** 잠근다 — litellm 가격표 GET 이 import 때 난다.
    seal_outbound()

    root = Path(settings.projects_dir) / pid / "checkpoints" / "episodes" / epi
    if not root.exists():
        print(f"★없는 에피소드: {root}")
        return 2

    clean = (_cp(root, "text_cleanup").get("data") or {}).get("cleaned_text") or ""
    rules_cp = _cp(root, "visual_world_rules")
    rules = rules_cp.get("data") or {}
    era, region = str(rules.get("era") or ""), str(rules.get("region") or "")
    shots = (_cp(root, "shot_validator").get("data") or {}).get("scenes") or []
    merged = _cp(root, "entity_merge").get("data") or {}
    rules_json = json.dumps(rules, ensure_ascii=False)

    R = Rows()
    fake = [{"research_subject_id": f"r{i}", "owner_type": t,
             "surface_form": "가", "source_anchor": "a", "source_quote": "나",
             "planned_occurrences": 1, "why_candidate": "다"}
            for i, t in enumerate(("character", "location", "prop"))]

    # ── A0 (조립만 — production builder) ──
    from app.modules.pipeline import grounding_a0 as a0

    pk = a0.load_pack()
    _a0_parts = (pk["stems"][a0.SYSTEM_STEM]["content"],
                 a0.build_user_prompt(
                     clean, era=era, region=region,
                     shot_lines=[f"Scene {s.get('scene_index')}: "
                                 f"{sh.get('description','')}"
                                 for s in shots
                                 for sh in (s.get("shots") or [])]),
                 pk["stems"][a0.SCHEMA_STEM]["content"])
    R.add("grounding_a0", 1, _b(*_a0_parts), "원문 전문 + 샷 전부",
          sha=_sha(*_a0_parts))

    # ── entity_all ×3 · entity_extract ×3 (production 조립을 가로챈다) ──
    from app.modules.pipeline import entity_extractor_v4 as ex
    from app.modules.pipeline import entity_lister as el

    for t, key in (("character", "characters"), ("location", "locations"),
                   ("prop", "props")):
        cap = Capture()
        el.call_structured = cap
        el.list_entities_from_shots(shots, t, rules_json, a0_candidates=fake)
        R.add_calls(f"entity_all_{t}", cap.calls, "샷 description 전부")

        listed = [{"name": e.get("name", ""), "short_id": e.get("short_id", "")}
                  for e in (merged.get(key) or [])]
        if listed:
            cap = Capture()
            ex.call_structured = cap
            ex.extract_entities_by_type_with_list(
                clean, t, listed, rules_json, a0_candidates=fake)
            R.add_calls(f"entity_extract_{t}", cap.calls,
                        "★원문 전문 **또** + 이름 목록")

    # ── entity_merge (★첫 판에서 통째로 빠졌던 것 — Codex BLOCK-1) ──
    from app.core.steps import entity_steps as es

    cap = Capture({"remove": []})
    es.call_structured = cap
    step = es.EntityMergeStep(step_id="entity_merge", project_id=pid,
                              episode_id=epi, db=None, project_config={})
    step._load_prev_checkpoint = lambda sid: _cp(root, sid)  # type: ignore
    try:
        step._execute()
        R.add_calls("entity_merge", cap.calls, "세 갈래 이름+설명 목록")
    except Exception as exc:  # noqa: BLE001
        print(f"★entity_merge 를 못 태웠다 — {exc!r} (표에서 빠진다)")

    # ── grounding_plan classifier (표본 × 판정자, subject 는 묶음) ──
    from app.core.steps.grounding_steps import DEFAULT_SAMPLES
    from app.modules.pipeline import grounding_carry as carry
    from app.modules.pipeline import grounding_classifier as gc

    built = carry.build_subjects(merged, project_id=pid, episode_id=epi,
                                 source_step="entity_merge",
                                 a0_candidates=[])
    subjects = built["subjects"]
    n_cls = DEFAULT_SAMPLES * len(gc.DEFAULT_JUDGES)
    if subjects:
        gpk = gc.load_pack()
        R.add("grounding_plan(classifier)", n_cls,
              _b(gpk["stems"][gc.SYSTEM_STEM]["content"],
                 gc.build_user_prompt(subjects, era=era, region=region),
                 gpk["stems"][gc.SCHEMA_STEM]["content"]),
              f"대상 {len(subjects)}개 **묶음** — 대상 수와 무관한 고정 {n_cls}")

    # ── grounding_research (★active 인데 빠졌던 것 — Codex BLOCK-2) ──
    #   BATCH_SIZE=1 이라 route=research 대상마다 한 번이다.
    from app.modules.pipeline import grounding_claims_search as gcs

    n_research = sum(1 for d in (built.get("unbound") or []) if False) or 0
    # ★저장 CP 에 `grounding_plan` 이 없으면 **몇 개가 research 인지 모른다.**
    #  짐작으로 채우지 않는다 — 「모른다」로 적는다.
    plan_cp = _cp(root, "grounding_plan").get("data") or {}
    decided = plan_cp.get("decided") or []
    n_research = sum(1 for d in decided if d.get("route") == "research")
    # ★★R 을 **0 으로 합산하지 않는다** (Codex). 모르는 것을 0 으로 더하면
    #  합계가 거짓이 된다. 대신 **상한**을 낸다 — 모든 subject 가 research 일
    #  때의 bytes 를 프로덕션 조립으로 실제로 만들어 본다.
    research_max = 0
    research_one = 0
    if subjects:
        rpk = gcs.load_pack()
        _rsys = next((v["content"] for k, v in rpk["stems"].items()
                      if "sys" in k), "")
        _rsch = next((v["content"] for k, v in rpk["stems"].items()
                      if "schema" in k), {})
        for sub in subjects:
            research_max += _b(_rsys, gcs.build_user_prompt(
                [sub], era=era, region=region), _rsch)
        # ★평균은 **합계를 개수로** 나눈다. 앞 판은 루프 안에서 덮어쓴
        #  마지막 값을 109 로 나눠 「1회 68 bytes」라는 헛것을 냈다.
        research_one = research_max // max(1, len(subjects))

    # ── grounding_screen (★예정 · 지금 disabled) ──
    from app.core.world_context import build_world_facts_block
    from app.modules.pipeline import era_research as era_mod
    from app.modules.pipeline import grounding_screen as gs

    world = build_world_facts_block(rules_cp)
    if subjects:
        # ★`assess_subjects` 는 **함수 안에서** `llm_client.call_structured` 를
        #  import 한다 — 모듈 속성만 갈아 끼우면 안 잡히고 **실제로 나간다**.
        #  (처음에 그렇게 했다가 2분 timeout 이 났다.)
        from app.modules.llm import llm_client as _llm

        cap = Capture({"subjects": []})
        _orig = _llm.call_structured
        _llm.call_structured = cap
        failed = 0
        try:
            for sub in subjects:
                try:
                    era_mod.assess_subjects(
                        step_tag="audit", subject_text=gs.subject_text_of(sub),
                        world_facts_block=world)
                except Exception:  # noqa: BLE001
                    failed += 1
        finally:
            _llm.call_structured = _orig
        if failed:
            # ★**못 잰 것을 0 으로 안 적는다.**
            print(f"★screen 을 {failed}/{len(subjects)}개 못 쟀다 — "
                  f"아래 소계는 **하한**이다")
        R.add_calls("grounding_screen (예정·지금 disabled)", cap.calls,
                    "대상 하나 — ★대상마다 한 번", group=PLANNED)

    # ── 구조화 산출 (★다른 도구가 읽는다 — 표를 다시 파싱하지 않게) ──
    if "--json" in sys.argv:
        out = Path(sys.argv[sys.argv.index("--json") + 1])
        out.write_text(json.dumps(
            {"project": pid, "episode": epi, "chars": len(clean),
             "scenes": len(shots), "rows": R.rows},
            ensure_ascii=False, indent=1), encoding="utf-8")
        print(f"■ 적었다: {out}")

    # ── 표 ──
    print(f"■ {pid[:8]}/{epi[:8]} — 원문 {len(clean):,}자 · 씬 {len(shots)} · "
          f"샷 {sum(len(s.get('shots') or []) for s in shots)} · "
          f"엔티티 {sum(len(merged.get(k) or []) for k in ('characters','locations','props'))}")
    print("  ★범위 = A0 → 첫 이미지 검색 직전의 고증 관련 호출만. "
          "scene/beat/shot/entity_relation/filter/detail/t2i 는 **빠졌다**.")
    print("  ★SOT 는 bytes. 오른쪽 token-eq 는 bytes/2.5 **추정**이다.")
    print()
    hdr = f"{'호출':34} {'회수':>4} {'1회 bytes':>11} {'합계 bytes':>12} {'합계 token-eq':>13}"
    for grp, label in ((ACTIVE, "현재 active"), (PLANNED, "screen 을 켜면 더해짐")):
        sel = [r for r in R.rows if r["group"] == grp]
        if not sel:
            continue
        print(f"── {label}")
        print(hdr)
        print("─" * 78)
        for r in sel:
            print(f"{r['name']:34} {r['n']:>4} {r['bytes']:>11,} "
                  f"{r['tot']:>12,} {int(r['tot']/_BYTES_PER_TOKEN):>13,}")
            if r["n"] == 0 or "모른다" in r["carries"]:
                print(f"{'':34}   ↳ {r['carries']}")
        tc, tb = sum(r["n"] for r in sel), sum(r["tot"] for r in sel)
        print("─" * 78)
        print(f"{'소계':34} {tc:>4} {'':>11} {tb:>12,} "
              f"{int(tb/_BYTES_PER_TOKEN):>13,}")
        print()
    ac = [r for r in R.rows if r["group"] == ACTIVE]
    pl = [r for r in R.rows if r["group"] == PLANNED]
    kc, kb = sum(r["n"] for r in ac), sum(r["tot"] for r in ac)
    n_sub = len(subjects)
    print(f"★알려진 active           호출 {kc} · {kb:,} bytes")
    print(f"★+ grounding_research    호출 **R (모름, 0 ≤ R ≤ {n_sub})** · "
          f"1회 {research_one:,} bytes · R={n_sub} 이면 +{research_max:,}")
    print(f"  → active 합계 범위       호출 {kc}~{kc + n_sub} · "
          f"{kb:,}~{kb + research_max:,} bytes")
    print(f"★screen 을 켜면 더해짐   호출 +{sum(r['n'] for r in pl)} · "
          f"+{sum(r['tot'] for r in pl):,} bytes")
    print()
    _output_evidence(root, len(subjects))
    print("★한계 — 어디가 하한이고 어디가 방향조차 모르는지 (Codex 정정):")
    print("  · **하한**: 결속 catalog overlay(가짜 후보 3개) · "
          "merge 의 provenance 지문(지금 CP 에 그 칸이 없어 `merges` schema 가 안 붙는다)")
    print("  · **proxy · 방향 미확정**: classifier·screen 의 **모집단과 payload**. "
          "지금은 `a0_candidates=[]` 라 `entity_description` 을 쓰는데, 실제 A0 "
          "결속 뒤에는 **원문 인용**과 승격분이 들어가 bytes 가 **줄 수도 늘 수도** 있다")
    return 0


def _output_evidence(root: Path, n_subjects: int) -> None:
    """★C 안의 **잘림 위험**을 재는 유일한 근거 — 지금 한 호출이 몇 행·몇
    bytes 를 **실제로** 내고 있나.

    C 안(구간마다 한 번 읽고 union 을 낸다)의 미지수는 구간 크기 S 다.
    S 를 짐작으로 정하면 안 되고, 「이 파이프라인이 **이미 성공적으로 내고
    있는 산출 크기**」를 닻으로 삼는 것이 가장 가까운 근거다.

    ★이것이 증명하지 **않는** 것 — 한 응답에 owner 다섯 갈래와 flag 를 함께
    담았을 때의 품질. 크기가 같아도 판단이 섞여 나빠질 수 있다. 그건 재려면
    실제로 사야 한다.
    """
    import json as _j

    print()
    print("■ C 안의 구간 크기 S 를 정할 **닻** — 지금 한 호출이 내는 산출")
    rows_max = bytes_max = 0
    for t, k in (("character", "characters"), ("location", "locations"),
                 ("prop", "props")):
        a = (_cp(root, f"entity_all_{t}").get("data") or {}).get(k) or []
        d = (_cp(root, f"entity_extract_{t}").get("data") or {}).get(k) or []
        ab = len(_j.dumps(a, ensure_ascii=False).encode("utf-8"))
        db = len(_j.dumps(d, ensure_ascii=False).encode("utf-8"))
        rows_max = max(rows_max, len(a))
        bytes_max = max(bytes_max, db)
        print(f"   entity_all_{t:10} {len(a):>4}행 {ab:>8,} B  →  "
              f"entity_extract {len(d):>4}행 {db:>8,} B")
    print(f"   ★지금 **성공하고 있는** 한 응답의 최대: "
          f"{rows_max}행 · {bytes_max:,} B")
    print(f"   ★C 를 **에피소드 통째 1콜**로 하면 대략 {n_subjects}행 · "
          f"{bytes_max * 3:,} B 언저리 — 지금 최대의 **약 3배**다.")
    # ★★앞 판은 여기서 「출력 3배니까 구간 3개」라고 적었다. **틀렸다** —
    #  엔티티는 구간에 **겹쳐 나타난다**(열 씬에 나오는 인물은 그 씬을 품은
    #  모든 구간에서 다시 나온다). 실측하면 S=3 이 행 수를 1/3 로 안 줄인다.
    print("   ★「출력이 3배니 구간 3개」는 **틀린 셈**이다 — 엔티티는 구간에 "
          "**겹쳐 나타난다**.")
    print("     실제 구간당 행 수는 `tools/grounding_audit/segment_bound.py` "
          "가 `scene_director` 로 잰다.")
    print("   ★이 수가 말하지 **않는 것**: 한 응답에 다섯 갈래와 flag 를 함께 "
          "담았을 때의 **품질**.")
    print("     크기가 같아도 판단이 섞여 나빠질 수 있다 — 재려면 실제로 사야 한다.")


if __name__ == "__main__":
    raise SystemExit(main())