#!/usr/bin/env python
"""GROUNDING-V2 §2-3c — 지정 대조군 아홉 축 판정. ★**유료**(Sol 호출).

계획서 §10 이 정본이고 이 도구는 그 표를 코드로 읽는다:
    저장 양성 4 → 전부 `research`
    저장 음성 3 → 전부 non-research
    합성 음성 2 → 전부 non-research

★**하나라도 어긋나면 미확정이다.** 일부만 맞은 것을 통과로 세지 않는다.

★**한 판으로 재지 않는다.** 실측에서 같은 입력·같은 팩이 판마다 달랐다
 (`0aea12c2/P03` research 4 · skip 1 · `fixture:hanbok` 이 `discriminability`
 yes↔uncertain, `confidence` 0.82↔0.62 로 흔들림). 판정기를 결정적으로 만들 수도
 없다 — `gpt`(Sol) 는 `_NO_TEMPERATURE_ALIASES` 라 temperature 를 못 내린다.

★그래서 `--repeats` 회 재되 **갈린 축은 통과도 실패도 아닌 「미확정」**이다.
 앞 판은 `research/skip/skip` 을 research 로 접어 **양성 축을 통과**시켰는데,
 프로덕션은 한 번 접어 확정하므로 실제로는 어느 쪽이든 나올 수 있다 —
 그것은 **미확정을 합격으로 바꾼 것**이었다.

★완료를 쓸 때도 「안정성이 증명됐다」가 아니라 **「N회에서 갈림 0건 관찰」**이라고
 쓴다. N회 같았다는 것이 항상 같다는 뜻은 아니다.
★fixture 의 `content_hash` 를 매번 다시 계산해 확인한다 — 판정이 원하는 대로
 안 나왔을 때 문안을 고쳐 다시 돌리면 그건 측정이 아니다.

    .venv/bin/python tools/prompt_measure/grounding_controls_acceptance.py \
        --projects-root <경로> --i-know-this-costs-money

종료코드: 0 아홉 축 전부 통과 · 1 어긋남/미확정 · 2 잴 것이 없음
"""
from __future__ import annotations

import argparse
import contextlib
import hashlib
import json
import sys
from pathlib import Path

_REPO = Path(__file__).resolve().parents[3]
sys.path.insert(0, str(_REPO / "backend"))

from app.core.steps.grounding_steps import check_provenance  # noqa: E402
from app.modules.pipeline.grounding_classifier import (  # noqa: E402
    DEFAULT_JUDGES, classify, classify_samples,
)
from app.modules.pipeline.grounding_planner import (  # noqa: E402
    decide_route, decide_route_from_samples,
)
from app.modules.pipeline.grounding_shadow import (  # noqa: E402
    ShadowSourceError, build_subjects_from_saved_episode,
)
from app.modules.pipeline.grounding_subject import build_subject  # noqa: E402

_FIXTURE = _REPO / "backend" / "tests" / "fixtures" / "grounding" / "controls.json"
#: ★리그레션 대조군. §10 아홉 축과 겹치지 않는 대상들이다.
#: ★한때 튜닝 독립성 증명용이었으나, 이 축들의 **갈림 통계**를 보고 문안을
#:  고친 뒤로는 아니다 — 낱말이 안 새도 **통계를 봤으면 튜닝**이다(Codex).
_HELDOUT = (_REPO / "backend" / "tests" / "fixtures" / "grounding"
            / "heldout_controls.json")

#: ★`research` 만 「조사한다」다. `design`·`skip` 은 조사를 안 사고,
#: `unresolved` 는 **통과도 실패도 아니다** — 별도로 센다.
_RESEARCH = "research"


def load_fixture() -> dict:
    """★content_hash 를 **다시 계산해** 확인한다. 사후 수정 경로를 막는다."""
    data = json.loads(_FIXTURE.read_text(encoding="utf-8"))
    stated = data.pop("content_hash", None)
    actual = hashlib.sha256(
        json.dumps(data, sort_keys=True, ensure_ascii=False).encode()).hexdigest()[:16]
    if stated != actual:
        raise SystemExit(
            f"fixture content_hash 가 안 맞는다 (적힌 값 {stated}, 실제 {actual}) — "
            "돌린 뒤에 문안을 고쳤거나 해시를 안 갱신했다. 둘 다 측정을 무효로 만든다")
    data["content_hash"] = actual
    return data


_CACHE: dict = {}
_REPLAY: dict | None = None
#: ★**끊긴 판을 이어 산다** — 이미 산 칸은 다시 안 사고 빠진 칸만 산다.
#: `_REPLAY`(바깥 호출 0) 와 다르다: 여기는 **빠진 것을 실제로 산다**.
_RESUME: dict | None = None
#: ★산 것을 **판정보다 먼저** 흘려 쓸 곳. 없으면 안 쓴다.
_FLUSH_TO: Path | None = None
#: ★주행 예산을 여는 스택. ★**예외로 나가도** 내려가야 해서 전역이다.
_RUN_STACK: contextlib.ExitStack | None = None


def _flush(where: str = "") -> None:
    """★★**산 것을 판정보다 먼저 저장한다.**

    끝에서 한 번만 쓰면 중간에 서는 순간 **산 검색을 통째로 잃고 다시 산다.**
    실제로 유료 산출 23개를 그렇게 날린 적이 있다.
    """
    if _FLUSH_TO is None or _REPLAY is not None:
        return
    tmp = _FLUSH_TO.with_suffix(_FLUSH_TO.suffix + ".part")
    tmp.write_text(json.dumps(_CACHE, ensure_ascii=False, sort_keys=True,
                              indent=1), encoding="utf-8")
    tmp.replace(_FLUSH_TO)
    if where:
        print(f"    · 산 것 저장({len(_CACHE)}건) ← {where}")


def _save_records(args) -> None:
    """마지막 저장. ★호출 하나하나는 `_flush` 가 이미 썼다."""
    if getattr(args, "save_records", None) is None or _REPLAY is not None:
        return
    _flush()
    print(f"판정 기록 {len(_CACHE)}건 저장 → {args.save_records}")
    print("  이후 접기 수정은  --replay 로 **무료**로 잰다")


def _identity_of(kind: str, **coords) -> dict:
    """★이 호출이 **무엇이었나**. 재생이 옛 판을 새 판으로 못 쓰게 막는다."""
    return {"kind": kind, **{k: str(v) for k, v in sorted(coords.items())}}


def cached_call(cache_key: str, produce, *, identity: dict | None = None):
    """유료 호출 하나를 기록하거나 재생한다.

    ★재생은 **key 만 보지 않는다.** 같은 이름이라도 팩·모델·요청 계약·payload
    가 다르면 **다른 판**이다 — 그것을 재사용하면 옛 응답으로 새 계약을
    통과시킨다 (Codex).
    """
    if _REPLAY is not None:
        if cache_key not in _REPLAY:
            raise SystemExit(
                f"재생 파일에 '{cache_key}' 가 없다 — 같은 판이 아니다. "
                f"있는 것: {sorted(_REPLAY)[:6]}")
        got = _REPLAY[cache_key]
        if identity is not None and isinstance(got, dict):
            was = got.get("__identity__")
            if was != identity:
                raise SystemExit(
                    f"재생 신원이 다르다 — '{cache_key}'\n"
                    f"  저장된 것: {was}\n  지금 판: {identity}\n"
                    "★같은 이름이지만 **다른 판**이다. 옛 응답으로 새 계약을 "
                    "통과시키지 않는다.")
        return got.get("__value__", got) if isinstance(got, dict) \
            and "__value__" in got else got
    have = _resumed(cache_key, identity)
    if have is not None:
        _CACHE[cache_key] = ({"__identity__": identity, "__value__": have}
                             if identity is not None else have)
        return have
    got = produce()
    _CACHE[cache_key] = ({"__identity__": identity, "__value__": got}
                         if identity is not None else got)
    _flush(cache_key)
    return got


def _resumed(cache_key: str, identity: dict | None):
    """★끊긴 판이 **이미 산 것**이면 그것을 돌려준다. 없으면 ``None``.

    ★신원이 다르면 **없는 것으로 친다** — 옛 판을 새 계약에 이어 붙이지
    않는다. 그건 재개가 아니라 오염이다.
    """
    if _RESUME is None or cache_key not in _RESUME:
        return None
    row = _RESUME[cache_key]
    if isinstance(row, dict) and "__value__" in row:
        if identity is not None and row.get("__identity__") != identity:
            return None
        return row["__value__"]
    return None if identity is not None else row


def cached_classify(cache_key: str, **kw) -> dict:
    """★유료 판정을 **한 번만** 태우고, 이후 접는 규칙 수정은 **무료로** 잰다.

    planner 의 접는 규칙을 고칠 때마다 유료로 다시 구웠다 — 세 번 그랬다.
    판정 기록(축 값)은 그대로 두고 **접는 법만** 바꿔 재는 길이 있어야 한다.

    ★재생은 **판정만** 되돌린다. 접기·채점·지문 확인은 매번 새로 돈다.

    ★★그리고 재생은 **key 만 보지 않는다** — 검색과 같은 규칙이다. 팩·계약·
    판정자·표본 수·세계 맥락 중 하나라도 다르면 **다른 판**이고, 그 응답으로
    새 계약을 통과시키면 안 된다. 전에 여기만 신원이 없었다.
    """
    from app.modules.pipeline import grounding_classifier as _gc

    # ★신원의 뿌리는 **실제로 나갈 payload** 다 — subject id 목록이 아니다.
    #  프로덕션과 **같은 함수**(`planned_payload`)로 무료로 낸다. 지시문 한 줄만
    #  바뀌어도 신원이 움직여야 옛 응답이 새 계약을 통과하지 못한다 (Codex).
    ident = _identity_of(
        "classify",
        pack=_gc.PROMPT_PACK_VERSION,
        contract=_gc.CLASSIFIER_CONTRACT_VERSION,
        judges=",".join(kw.get("judges") or _gc.DEFAULT_JUDGES),
        samples=kw.get("samples"),
        strict=kw.get("strict_single_attempt"),
        payload=_gc.planned_payload(
            kw.get("subjects") or [],
            era=kw.get("era") or "", region=kw.get("region") or ""),
    )
    # ★★한 판(판정자 2 × 표본 3 = 6 논리 호출)이 다 끝난 뒤에만 저장하면,
    #  네 번째에서 끊길 때 **앞서 산 셋까지 다시 산다**. 호출 하나하나를
    #  먼저 흘려 쓴다 (Codex).
    def _slot(alias, n) -> str:
        # ★칸 이름은 **(판정자, 표본 번호)** 다. 도착 순서로 이름 붙이면
        #  재개가 「몇 번째로 왔나」에 기대게 되고, 순서가 달라지면 엉뚱한
        #  칸을 재사용한다 (Codex).
        return f"{cache_key}#{alias}#{n}"

    def _keep(one):
        key = _slot(one["judge_alias"], one["sample_index"])
        _CACHE[key] = {"__identity__": {**ident, "칸": key.split("#", 1)[1]},
                       "__value__": one}
        _flush(key)

    def _reuse(alias, n):
        key = _slot(alias, n)
        ident_slot = {**ident, "칸": key.split("#", 1)[1]}
        got = _resumed(key, ident_slot)
        if got is not None:
            # ★★**이어 받은 칸을 새 기록에도 남긴다.** 안 남기면 이 판이 또
            #  끊겼을 때 앞판에서 산 것이 사라지고, 다음 재개가 그것을 다시
            #  산다 — 재개하려고 만든 것이 재개를 못 하게 된다.
            _CACHE[key] = {"__identity__": ident_slot, "__value__": got}
            # ★**바로** 쓴다. 「나갈 때 한 번」에 기대면 다음 호출에서 터질 때
            #  이어 받은 것이 새 파일에 없다.
            _flush(f"{key} (이어받음)")
        return got

    return cached_call(cache_key,
                       lambda: classify_samples(on_sample=_keep, reuse=_reuse,
                                                **kw),
                       identity=ident)


def _record_run_fp(runs: dict, key: str, fp) -> None:
    # ★칸을 골라 복사하면 gate 가 읽는 칸이 조용히 빠진다 — 실제로 그랬다.
    #  지문을 **통째로** 남기고 gate 가 필요한 것을 고르게 한다.
    runs.setdefault(key, []).append(dict(fp or {}))


def _fold_records(records: list) -> tuple[str, str, bool]:
    """★**production 용으로 정의한 접기**를 쓴다 — `decide_route_from_samples`.

    ★production 호출부는 ``GroundingPlanStep`` 이다(§2-3.5 에서 배선했다).
    도구와 프로덕션이 **같은 접기 함수**를 쓴다.
    """
    d = decide_route_from_samples(records)
    votes = " ".join(f"{k}×{v}" for k, v in
                     sorted(d["votes"].items(), key=lambda kv: -kv[1]))
    if d.get("axis_undecided"):
        votes += f" [축 미결: {','.join(d['axis_undecided'])}]"
    elif d["unstable"]:
        # ★route 투표는 갈렸지만 **축 다수는 섰다** — 접은 판정이 정본이다.
        votes += " [축은 접힘]"
    # ★「못 정했다」는 **축 다수 실패**다. route 투표 갈림이 아니다 —
    #  판정을 축 접기로 바꾼 뒤로 route 투표는 진단일 뿐이다.
    return d["route"], votes, bool(d.get("undecided"))


def sourced_verdict(subject: dict, *, era: str, region: str,
                    cache_key: str, model: str) -> tuple[str, str]:
    """★★★**출처로 최종 판정을 낸다** — 계약 §2 의 마지막 칸.

    `decide_route` 가 `research` 라고 한 것은 「조사한다」이지 「시대 차이가
    있다」가 아니다. 진짜 답은 **검색해서 얻은 claim** 으로 `decide_delta` 가
    낸다:

    ```
    delta=yes        → research   (출처가 시대 차이를 지지한다)
    delta=no         → skip       (출처가 「차이 없음」을 지지한다)
    delta=unresolved → 미확정      (못 찾음·충돌·출처 없음·시간 초과)
    ```

    ★도구가 따로 세지 않는다 — **프로덕션 함수를 그대로 부른다**
    (`search_claims` → `sanitize_and_decide` — 저장 경로가 쓰는 바로 그것).
    도구가 자기 규칙을 쓰면 재는 것과 도는 것이 갈린다.

    Returns:
        ``(route, 근거 한 줄)``. `route` 는 ``research``·``skip``·``unresolved``.
    """
    from app.core.openai_keys import openai_client
    from app.modules.pipeline import grounding_claims_search as gcs
    from app.modules.pipeline.grounding_claims import sanitize_and_decide

    sid = subject["research_subject_id"]
    # ★재생 신원 — 팩·모델·요청 계약·payload 가 다르면 **다른 판**이다.
    ident = _identity_of(
        "search", model=model, pack=gcs.PROMPT_PACK_VERSION,
        request=json.dumps(gcs.request_contract(), sort_keys=True),
        payload=";".join(gcs.planned_payloads([subject], 1, era=era,
                                              region=region)))
    out = cached_call(cache_key, lambda: gcs.search_claims(
        openai_client(max_retries=gcs.SDK_RETRIES), [subject],
        model=model, era=era, region=region, batch_size=1),
        identity=ident)
    rows = out.get("batches") or []
    if not rows:
        return "unresolved", "검색 행이 없다"
    row = rows[0]
    if row.get("limit_kind") or row.get("error"):
        return "unresolved", f"상한/전송: {row.get('limit_kind') or row.get('error')}"
    mine = [r for r in ((row.get("parsed") or {}).get("results") or [])
            if isinstance(r, dict) and str(r.get("research_subject_id")) == sid]
    if len(mine) != 1:
        return "unresolved", f"그 대상 줄이 {len(mine)}개다"

    # ★★**검증·오염 처리·판정을 도구가 다시 만들지 않는다.** 전에는 검증
    #  못 거친 줄을 조용히 버리고 남은 것으로 yes/no 를 냈다 — production 은
    #  하나라도 깨지면 통째로 unresolved 다. 그러면 도구가 「맞음」이라고 센
    #  것을 production 은 미확정으로 쓴다 (Codex).
    fin = sanitize_and_decide(mine[0].get("claims") or [],
                              mine[0].get("gaps") or [],
                              research_subject_id=sid)
    delta, why, claims = fin["delta"], fin["why"], fin["claims"]
    if delta == "yes":
        return "research", f"출처가 시대 차이를 지지한다 (claim {len(claims)})"
    if delta == "no":
        return "skip", "출처가 「차이 없음」을 지지한다"
    return "unresolved", str((why or {}).get("why") or "미확정")


def _final_route(pre_route: str, subject: dict, *, era: str, region: str,
                 cache_key: str, model: str) -> tuple[str, str]:
    """검색 전 route → **최종 route**. ★`research` 만 실제로 산다.

    `design`·`skip` 은 검색을 안 한다 — 계약이 그렇게 갈라 놓았다.
    """
    if pre_route != "research":
        return pre_route, "검색 전 단계에서 끝남"
    return sourced_verdict(subject, era=era, region=region,
                           cache_key=cache_key, model=model)


def _cost_cap(fx: dict, args) -> dict:
    """★**이 slice 가 실제로 살 수 있는 최대**를 코드 경로대로 센다.

    ★출력 문구만 상한이면 **승인 문이 아니다.** 이 수를 그대로
    `research_run_scope(cap=…)` 에 넣어 검색을 물리적으로 막는다 (Codex).

    - 분류: (저장 에피소드 그룹 + 합성 축) × 판정자 × 반복
    - A0  : **저장 에피소드에서만** 돈다. 합성은 `build_subject` + 고정 인용이라
            A0 를 안 부른다 — 전에 여기에 합성 둘을 더해 5라고 적었다
    - 검색: pre-route 가 `research` 인 축만, 축당 1회. 최대 = **아홉 축 전부**
    """
    eps = sorted({c["episode_prefix"] for k in ("positive", "negative")
                  for c in fx["stored"][k]})
    syn = len(fx["synthetic_negative"])
    axes = sum(len(fx["stored"][k]) for k in ("positive", "negative")) + syn
    groups = len(eps) + syn
    return {
        "episodes": len(eps), "synthetic": syn, "groups": groups, "axes": axes,
        "classify": groups * len(DEFAULT_JUDGES) * args.repeats,
        "a0": len(eps) if args.with_a0 else 0,
        "search": axes,
    }


def _cap_lines(cap: dict, args) -> list:
    """★**논리와 물리를 갈라 적는다.** 「물리 18」은 검색만이었다 (Codex).

    ★★그리고 「분류·A0 는 strict 라 논리=물리」도 **틀렸다**(2026-08-30 실측).
    `strict_single_attempt` 가 봉인하는 것은 **tier fallback 과 num_retries**
    뿐이고, **OpenAI 키 슬롯 전환**은 그와 별개의 loop 다
    (`llm_client._router_completion` 의 `for _ in range(slot_count())`).
    실제 주행 로그에 그대로 찍혔다:

        OpenAI 키 전환: primary → secondary (사유=429 · 지점=router.completion[gpt])

    그래서 **세 갈래 전부** 슬롯 수만큼 곱한다.
    """
    from app.core.openai_keys import slot_count

    slots = max(1, slot_count())
    total = cap["classify"] + cap["a0"] + cap["search"]
    phys_search = cap["search"] * slots
    phys_all = total * slots
    return [
        f"★논리 상한 — 분류 {cap['groups']}그룹"
        f"(저장 에피소드 {cap['episodes']} + 합성 {cap['synthetic']})"
        f" × 판정자 {len(DEFAULT_JUDGES)}({','.join(DEFAULT_JUDGES)})"
        f" × {args.repeats}회 = **{cap['classify']}**",
        f"  + A0 **{cap['a0']}**(저장 에피소드에서만 — 합성은 A0 를 안 부른다)",
        f"  + 검색 최대 **{cap['search']}**"
        f"(pre-route=research 인 축만 · 축당 1회 · batch 1)",
        f"  = 논리 최대 **{total}**",
        f"★물리 상한 — 전체 **≤{phys_all}** (= 논리 {total} × 슬롯 {slots})",
        f"  ★세 갈래 **전부** 곱한다. `strict_single_attempt` 는 tier fallback 과"
        " num_retries 만 봉인하고, **키 슬롯 전환은 별개 loop** 다"
        " (실측: 429 로 primary → secondary 전환이 찍혔다).",
        f"  ★예산으로 **실제로 막는 것은 검색 ≤{phys_search}** 뿐이다."
        " 분류·A0 는 loop 구조로만 묶인다 — 예산 문이 아니다.",
        "  ★측정 범위 — 분류는 저장 에피소드의 **후보 전체**를 보지만, 검색은",
        "    사전에 잠근 **아홉 축만** 산다. full-episode 조사 E2E 가 아니다.",
    ]


def _align(subjects, records):
    """subject 순서에 맞춰 record 를 돌려준다 — classify 는 id 로 대응한다."""
    by_id = {r.get("research_subject_id"): r for r in records}
    return [by_id.get(s["research_subject_id"], {"classifier_missing": True})
            for s in subjects]


def _episode_dir(root: Path, project_prefix: str, episode_prefix: str) -> Path | None:
    """대조군 좌표(8자 접두어) → 실제 에피소드 디렉토리.

    ★`entity_merge` 로 본다 — production 의 `grounding_plan` 이 읽는 자리다.
    `entity_filter` 로 보면 **거기서 걸러진 것을 못 보는** 모집단이 된다.
    """
    for d in sorted(root.glob(f"{project_prefix}*/checkpoints/episodes/{episode_prefix}*")):
        if (d / "entity_merge" / "manifest.json").exists():
            return d
    return None


def real_ids(episode_dir: Path) -> tuple[str, str]:
    """디렉토리에서 **진짜 UUID** 를 읽는다.

    ★8자 접두어로 subject id 를 발급하면 production 과 **다른 id** 가 나온다 —
    `mint_subject_id` 가 project/episode id 를 해시에 넣기 때문이다. 좌표는
    사람이 읽는 접두어지만, 발급은 실제 범위로 해야 같은 것을 잰 것이 된다.
    """
    # <project>/checkpoints/episodes/<episode> — parents[0]=episodes,
    # parents[1]=checkpoints, parents[2]=<project>
    return episode_dir.parents[2].name, episode_dir.name


#: §10 이 정한 축 수. 이보다 적거나 많으면 fixture 가 틀린 것이다.
EXPECTED_AXES = 9


MARK_OK, MARK_UNDECIDED, MARK_MISMATCH = "맞음", "미확정", "★어긋남"


def score_row(want: str, got: str) -> str:
    """한 축의 채점. ★규칙은 **여기 하나**다.

    출력 loop 와 exit 판정이 각자 세면 두 곳이 갈린다 — 실제로 갈릴 뻔했다
    (`decide_exit` 이 `got != want` 로 다시 세서 합성 축의 `skip` 을
    어긋남으로 셌다).
    """
    if got.startswith("흔들림:"):
        # ★반복 사이에 답이 갈렸다. 통과도 실패도 아니다 —
        #  프로덕션은 1콜이라 어느 쪽이든 나올 수 있다.
        return MARK_UNDECIDED
    if got not in ("research", "design", "skip"):
        return MARK_UNDECIDED       # `unresolved` · 읽기 실패 등
    return MARK_OK if (got == _RESEARCH) == (want == _RESEARCH) else MARK_MISMATCH


#: 저장 축이 production 과 같은 입력으로 재졌다고 말할 수 있는 근거.
#: ``fixture`` 는 합성 축 — 원문이 없으니 문안이 정본이다(계획 §10).
PRODUCTION_EQUIVALENT_BASIS = frozenset({"manuscript", "fixture"})


def decide_exit(rows, provenance_bad, quote_sources=None, *, a0_ran=None):
    """★판정을 **한 함수**로 뺐다 — 소스 문자열이 아니라 동작으로 재려고.

    돌려주는 것: ``(exit_code, 사유 목록)``. ``0`` 은 통과뿐이다.

    가르는 순서:

    1. **A0 를 돌렸는가** — 안 돌린 판은 production 과 다른 입력이다.
       ★축마다 `manuscript` 를 요구하지는 **않는다.** production 도 A0 후보가
       없는 대상은 `entity_description` 으로 가므로, 그것을 요구하면
       프로덕션보다 엄격해지고 A0 가 판마다 달라 gate 자체가 흔들린다.
    2. 축 수가 아홉인가
    3. 같은 것을 N회 잰 것인가 (지문)
    4. 어긋남·미확정이 **둘 다 0** 인가

    Args:
        rows: ``(축, 라벨, 기대, 접은결과, 투표, 근거)``
        quote_sources: 진단 출력용.
        a0_ran: A0 를 돌렸는가. ``False`` 면 production 과 다른 입력이다.
    """
    reasons = []
    # ★**A0 가 돌았는가**를 본다 — 축마다 원문 인용이 붙었는가가 아니다.
    #
    #  왜 고쳤나: production 도 A0 후보가 없는 대상은 `entity_description` 으로
    #  간다. 축마다 `manuscript` 를 요구하면 **프로덕션보다 엄격**해서, A0 가
    #  그 대상을 안 건진 판은 영영 통과 못 한다. 게다가 A0 도 LLM 이라 판마다
    #  건지는 것이 달라 **gate 자체가 흔들린다**(실측: 같은 원고에서 원문 근거
    #  축이 판마다 2~5개로 오갔다).
    #
    #  ★대신 **A0 를 아예 안 돌린 판**은 막는다. 그건 production 과 다른 입력이다.
    if a0_ran is False:
        reasons.append(
            "A0 를 안 돌렸다 — production 과 같은 입력이 아니다. "
            "`--with-a0` 로 다시 재라")
    if len(rows) != EXPECTED_AXES:
        reasons.append(f"축이 {EXPECTED_AXES}개가 아니다 ({len(rows)}) — fixture 확인")
    reasons.extend(provenance_bad)
    bad = [r[1] for r in rows if score_row(r[2], r[3]) != MARK_OK]
    if bad:
        reasons.append(
            f"어긋나거나 미확정인 축 {len(bad)}개 {bad} — "
            "일부만 맞은 것을 통과로 안 센다")
    return (1 if reasons else 0), reasons


def _a0_from_records(ep_prefix: str):
    """저장 기록에 A0 산출이 있으면 그것을 돌려준다. 없으면 ``None``.

    ★preflight 가 A0 **없이** 지은 subject 로 id 를 찍으면, `--with-a0` 주행이
    실제로 사는 대상과 **다른 id** 를 보고하게 된다(실측: 아홉 중 둘).
    """
    src = _RESUME if _RESUME is not None else _REPLAY
    if not src:
        return None
    row = src.get(f"a0:{ep_prefix}")
    val = row.get("__value__") if isinstance(row, dict) and "__value__" in row \
        else row
    if isinstance(val, dict):
        return val.get("candidates")
    return val if isinstance(val, list) else None


def preflight(args, fx: dict) -> int:
    """★★★**무료 preflight** — 유료 승인을 내리기 전에 보는 표 (Codex).

    무엇을 보이나:

    1. **잠긴 아홉 축의 정확한 id** — 저장 축은 실제 CP 를 읽어
       `research_subject_id` 까지 낸다(LLM 0회). 못 붙는 축이 있으면 **여기서**
       드러난다. 유료로 30회 태운 뒤에 「후보에 없음」을 보면 늦다
    2. **논리 상한과 물리 상한** — 물리는 슬롯 failover 때문에 논리보다 크다
    3. 이 slice 가 **무엇이 아닌지** — full-episode 조사 E2E 가 아니다

    ★바깥 호출 0. 읽기만 한다.
    """
    root = args.projects_root
    cap = _cost_cap(fx, args)
    print()
    for line in _cap_lines(cap, args):
        print(line)

    _src = "A0 기록 사용" if (_RESUME or _REPLAY) else "A0 **전** 상태"
    print(f"\n잠긴 아홉 축 — fixture {fx['content_hash']} · {_src}")
    if not (_RESUME or _REPLAY) and args.with_a0:
        print("  ★A0 가 subject id 를 바꾼다. 지금 id 는 **A0 전** 것이라")
        print("    `--with-a0` 주행이 실제로 살 id 와 다를 수 있다 —")
        print("    정확히 보려면 `--resume <기록>` 이나 `--replay <기록>` 을 같이 준다.")
    missing: list = []
    n = 0
    for polarity in ("positive", "negative"):
        want = _RESEARCH if polarity == "positive" else "non-research"
        for spec in fx["stored"][polarity]:
            n += 1
            ep_prefix, short = spec["episode_prefix"], spec["short_id"]
            ep = _episode_dir(root, fx["project_id_prefix"], ep_prefix)
            sid = "★에피소드를 못 찾음"
            if ep is not None:
                try:
                    _pid, _eid = real_ids(ep)
                    # ★★**A0 가 subject id 를 바꾼다.** A0 없이 지은 subject 로
                    #  id 를 찍어 놓고 「이것이 이번 주행이 살 아홉 축」이라고
                    #  적으면 **거짓**이다 — 실측에서 아홉 중 둘이 달랐다
                    #  (`fb7a883f/P01`·`P03`). 기록이 있으면 그 A0 산출을
                    #  써서 **주행과 같은 subject** 를 짓는다.
                    _cands = _a0_from_records(ep_prefix)
                    built = build_subjects_from_saved_episode(
                        ep, project_id=_pid, episode_id=_eid,
                        a0_candidates=_cands)
                    hit = next(
                        (x for x in built["subjects"]
                         if (x.get("provenance") or {}).get("short_id") == short),
                        None)
                    sid = hit["research_subject_id"] if hit else "★후보에 못 붙음"
                    if _cands is None and args.with_a0:
                        sid += "  (A0 전 — 주행에서 달라질 수 있다)"
                except ShadowSourceError as exc:
                    sid = f"★CP 읽기 실패: {exc}"
            if sid.startswith("★"):
                missing.append(f"{ep_prefix}/{short} — {sid}")
            print(f"  {n}. {want:13} {ep_prefix}/{short:4} {spec['name'][:22]:24} "
                  f"{sid}")

    for spec in fx["synthetic_negative"]:
        n += 1
        subj = build_subject(
            project_id="fixture", episode_id=spec["fixture_id"],
            source_anchor=spec["fixture_id"], surface_form=spec["surface_form"],
            owner_type=spec["owner_type"],
            provenance={"fixture_id": spec["fixture_id"]})
        print(f"  {n}. {'non-research':13} {spec['fixture_id']:17} "
              f"{spec['surface_form'][:22]:24} {subj['research_subject_id']}")

    print(f"\n합계 {n}축 · 못 붙은 축 {len(missing)}")
    if missing:
        print("★유료로 태우기 전에 이것부터 — 붙지 않는 축은 미확정으로 끝난다:")
        for m in missing:
            print(f"  · {m}")
        return 2
    print("★바깥 호출 0 — 여기까지 전부 읽기만 했다")
    return 0


def load_heldout() -> dict:
    """held-out 정본. ★``content_hash`` 를 **다시 계산해** 대조한다.

    돌리기 전에 잠근 것을 결과 보고 고치면 측정이 아니다.
    """
    import hashlib

    spec = json.loads(_HELDOUT.read_text(encoding="utf-8"))
    want = spec.pop("content_hash", None)
    body = json.dumps(spec, ensure_ascii=False, sort_keys=True,
                      separators=(",", ":"))
    got = hashlib.sha256(body.encode()).hexdigest()[:16]
    if want != got:
        print(f"★held-out 이 잠긴 내용과 다르다 (기록 {want} · 계산 {got})")
        raise SystemExit(2)
    spec["content_hash"] = got
    return spec


def run_heldout(args) -> int:
    """held-out 축을 **프로덕션과 같은 접기**로 잰다.

    ★`why` 는 **모델에 안 들어간다** — 그건 사람이 읽는 사유다. 넣으면
    정답을 알려 주는 것이라 held-out 이 아니게 된다.
    """
    spec = load_heldout()
    print(f"held-out {len(spec['cases'])}축 · hash {spec['content_hash']} "
          f"· {args.repeats}회 반복")
    runs: dict[str, list[dict]] = {}
    rows: list[tuple[str, str, str, str, str, str]] = []
    for c in spec["cases"]:
        subj = build_subject(
            project_id="heldout", episode_id=c["fixture_id"] if "fixture_id" in c
            else c["id"], source_anchor=c["id"],
            surface_form=c["surface_form"], owner_type=c["owner_type"],
            provenance={"held_out_id": c["id"]})
        subj["source_quote"] = c["source_quote"]
        smp = cached_classify(
            f"heldout:{c['id']}",
            subjects=[subj], samples=args.repeats, era=c["era"], region=c["region"],
            strict_single_attempt=True)
        for fp in smp["runs"]:
            _record_run_fp(runs, c["id"], fp)
        got = smp["by_subject"].get(subj["research_subject_id"], [])
        route, votes, unstable = _fold_records(got) if got else ("표본 없음", "", True)
        rows.append(("heldout", c["id"], c["expect"],
                     f"흔들림:{route}" if unstable else route, votes, "fixture"))

    ok = bad = und = 0
    print()
    for axis, label, want, got, votes, basis in rows:
        mark = score_row(want, got)
        ok += mark == MARK_OK
        bad += mark == MARK_MISMATCH
        und += mark == MARK_UNDECIDED
        print(f"  {mark:8} {axis:8} 기대={want:13} 실제={got:12} [{votes}] {label}")
    print(f"\n맞음 {ok} · 어긋남 {bad} · 미확정 {und}  (총 {len(rows)})")
    # ★**갈림을 따로 센다.** 「미확정 0」과 「갈림 0」은 다른 말이다 —
    #  축 다수로 접은 route split 이 있으면 그것을 안 적고 「갈림 0건」이라
    #  쓰면 안 된다(Codex 지적: 실제로 두 판에 각각 2건이 있었다).
    _sf = sum(1 for r in rows if "[축은 접힘]" in r[4])
    _au = sum(1 for r in rows if "축 미결:" in r[4])
    print(f"축 다수 실패 {_au}건 · 표본 갈림을 다수로 접은 것 {_sf}건")
    if _sf:
        print("  ★「갈림 0건」이라고 쓰면 안 된다 — 위 수를 그대로 적는다")

    provenance_bad: list[str] = []
    for key, entries in runs.items():
        # ★**프로덕션 gate 를 그대로 부른다.** 규칙을 여기 다시 적었더니
        #  본판만 고치고 held-out 을 안 고쳐 여섯 축이 통째로 막혔다.
        for b in check_provenance(entries,
                                  expected_judges=list(DEFAULT_JUDGES),
                                  expected_samples=args.repeats):
            provenance_bad.append(f"{key}: {b}")
    code, reasons = decide_exit(rows, provenance_bad)
    if len(rows) != len(spec["cases"]):
        reasons.append("축 수가 정본과 다르다")
        code = 1
    # ★held-out 은 축 수가 9 가 아니다 — 그 사유는 뺀다
    reasons = [r for r in reasons if "축이 9개가 아니다" not in r
               and not r.startswith("축이 ")]
    code = 1 if reasons else 0
    if code:
        print("\n★held-out 통과가 아니다:")
        for r in reasons:
            print(f"  · {r}")
        return code
    print("\n★리그레션 대조군 통과 — **일반화 증명이 아니다**(이 축들을 보고\n     문안을 고쳤다). 일반화는 한 번도 안 본 새 평가축의 첫 실행으로만 낸다")
    return 0


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--projects-root", type=Path, default=_REPO / "projects")
    ap.add_argument("--i-know-this-costs-money", action="store_true")
    ap.add_argument(
        "--held-out", action="store_true",
        help=("★§10 대신 **리그레션 대조군**으로 잰다 — §10 과 겹치지 않는 "
              "대상들이다. ★튜닝 독립성 증명이 **아니다**. "
              "합성 축뿐이라 저장 에피소드를 안 읽고 A0 도 안 부른다. "
              "★이것은 §2-3c 통과 조건이 아니다."))
    ap.add_argument(
        "--with-a0", action="store_true",
        help=("★A0 를 돌려 **원문 인용**을 근거로 잰다 — production 과 같은 입력. "
              "안 주면 저장 CP 의 LLM 묘사로 재고, 그 판은 **미확정**이다. "
              "에피소드마다 Sol 1회를 더 산다(검색은 0)."))
    ap.add_argument("--save-records", type=Path, default=None,
                    help="유료 판정 기록을 저장한다 — 이후 접기 수정은 무료로 잰다")
    ap.add_argument("--replay", type=Path, default=None,
                    help="저장한 판정 기록으로 **바깥 호출 없이** 다시 채점한다")
    ap.add_argument(
        "--resume", type=Path, default=None,
        help=("★끊긴 판을 **이어 산다** — 이미 산 칸은 안 사고 빠진 칸만 산다. "
              "`--replay` 와 다르다(그건 바깥 호출 0). 신원이 다른 칸은 "
              "**없는 것으로 치고 다시 산다**."))
    ap.add_argument(
        "--search-model", default="",
        help=("★§4a 조사에 쓸 모델. 안 주면 설정의 기본값. 검색은 **route 가 "
              "`research` 인 축에서만** 나간다 — 축마다 최대 1회."))
    ap.add_argument(
        "--preflight", action="store_true",
        help=("★**무료**. 잠긴 아홉 축의 실제 `research_subject_id` 와 "
              "논리·물리 상한을 보인다. 바깥 호출 0."))
    ap.add_argument("--repeats", type=int, default=3,
                    help="★같은 입력을 몇 번 재는가 (기본 3). 판정이 경계에서 흔들린다")
    args = ap.parse_args()
    if not args.search_model:
        from app.core.config import settings
        args.search_model = settings.openai_model
    # ★★**저장 대상을 모든 분기보다 먼저 잡는다.** held-out 은 여기 아래
    #  분기에서 바로 return 하는데, `_FLUSH_TO` 를 그 뒤에 잡아 뒀더니 새
    #  프로세스의 held-out 유료 주행은 `_flush` 가 통째로 no-op 이었다 —
    #  중단이든 완주든 산 것을 잃는다 (Codex).
    global _FLUSH_TO
    _FLUSH_TO = args.save_records

    global _RESUME
    if args.resume is not None:
        _RESUME = json.loads(args.resume.read_text(encoding="utf-8"))
        print(f"재개: {args.resume} — 이미 산 칸 {len(_RESUME)}개는 다시 안 산다")
        if args.save_records is None:
            raise SystemExit(
                "--resume 은 --save-records 와 같이 준다 — 이어 산 것도 "
                "남겨야 다음 번에 또 이어 살 수 있다")

    global _REPLAY
    if args.replay is not None:
        _REPLAY = json.loads(args.replay.read_text(encoding='utf-8'))
        print(f'재생: {args.replay} — 바깥 호출 0건')

    fx = load_fixture()
    print(f"fixture content_hash = {fx['content_hash']}  (아홉 축 · {args.repeats}회 반복)")

    if args.preflight:
        if not args.projects_root.is_dir():
            print(f"projects 디렉토리가 없다: {args.projects_root}")
            return 2
        return preflight(args, fx)

    if args.held_out:
        # ★held-out 은 합성 축뿐이라 저장 에피소드·A0 가 필요 없다.
        if not (args.i_know_this_costs_money or _REPLAY is not None):
            print("held-out 도 에피소드마다 Sol 을 부른다 (검색은 0).")
            print("정말 돌리려면 --i-know-this-costs-money 를 주세요.")
            return 2
        rc = run_heldout(args)
        _save_records(args)
        return rc

    if not (args.i_know_this_costs_money or _REPLAY is not None):
        # ★★상한을 **세어서** 적는다. 「Sol 을 부른다」로만 적었더니 판정자가
        #  둘인 것도, 검색이 나가는 것도 빠져 실제와 달랐다 (Codex).
        cap = _cost_cap(fx, args)
        for line in _cap_lines(cap, args):
            print(line)
        print("정말 돌리려면 --i-know-this-costs-money 를 주세요.")
        return 1

    root = args.projects_root
    if not root.is_dir():
        print(f"projects 디렉토리가 없다: {root}")
        return 2

    # ★★**세어 놓은 상한을 그대로 문으로 건다.** 출력 문구만 상한이면 승인
    #  문이 아니다 — 검색 호출은 `reserve_current_research_call` 자리에서
    #  물리적으로 막혀야 한다 (Codex).
    cap = _cost_cap(fx, args)
    for line in _cap_lines(cap, args):
        print(line)
    _budget = None
    global _RUN_STACK
    _RUN_STACK = contextlib.ExitStack()
    _stack = _RUN_STACK
    if _REPLAY is None:
        from app.core.openai_keys import slot_count
        from app.core.research_call_budget import research_run_scope
        # ★예산은 **물리 전송**을 센다(슬롯 failover 가 논리 1을 물리 2로
        #  만든다). 논리 상한을 그대로 물리 상한으로 두면 정상 failover 가
        #  막히므로 슬롯 수를 곱한다 — 그래도 「무한」이 아니다.
        _slots = max(1, slot_count())
        _phys = cap["search"] * _slots
        # ★★`install_budget` 만 부르면 **중간에 예외가 나는 순간 예산이 그대로
        #  남는다.** production 이 쓰는 `research_run_scope` 를 그대로 쓴다 —
        #  들어올 때 걸려 있던 것을 되돌리는 것까지 그 함수가 한다.
        _budget = _stack.enter_context(research_run_scope(cap=_phys))
        print(f"  ★검색 예산 설치 — 논리 {cap['search']} · "
              f"물리 {_phys}(= {cap['search']} × 슬롯 {_slots})에서 **막는다**")
    # ★반복마다 **무엇을 태웠는지** 남긴다 — payload hash 가 다르면 같은 입력이
    #  아니고, physical model 이 다르면 같은 판정자가 아니다. 그것을 모르면
    #  「흔들린다」가 모델 분산인지 입력 차이인지 못 가른다.
    runs: dict[str, list[dict]] = {}
    rows: list[tuple[str, str, str, str, str]] = []   # (축, 라벨, 기대, 접은결과, 투표)
    #: ★에피소드별로 **무엇을 근거로 쟀는지**. 보고에 그대로 옮긴다.
    quote_sources: dict[str, dict[str, int]] = {}

    # ── 저장 대조군 — ★**에피소드당 1콜**로 묶는다 ────────────────────
    #
    # ★production/shadow 는 에피소드의 **후보 전체를 1회** classify 한다.
    #  축 하나씩 부르면 호출 모양이 달라져 **다른 것을 재게 된다** — 한 후보만
    #  든 프롬프트와 전체가 든 프롬프트는 모델이 보는 맥락이 다르다.
    want_by_ep: dict[str, dict[str, str]] = {}
    label_by_ep: dict[str, dict[str, str]] = {}
    for polarity in ("positive", "negative"):
        want = _RESEARCH if polarity == "positive" else "non-research"
        for spec in fx["stored"][polarity]:
            ep = spec["episode_prefix"]
            want_by_ep.setdefault(ep, {})[spec["short_id"]] = want
            label_by_ep.setdefault(ep, {})[spec["short_id"]] = (
                f"{ep}/{spec['short_id']} {spec['name']}")

    for ep_prefix, wants in want_by_ep.items():
        labels = label_by_ep[ep_prefix]
        ep = _episode_dir(root, fx["project_id_prefix"], ep_prefix)
        if ep is None:
            for sid, want in wants.items():
                rows.append(("stored", labels[sid], want, "에피소드를 못 찾음", "", "없음"))
            continue
        a0_cands = None
        unbound_by_sid: dict = {}
        if args.with_a0:
            # ★production 과 **같은 입력**으로 재려면 A0 를 실제로 돌려야 한다.
            #  저장 CP 에는 A0 산출이 없다(legacy 주행이었다).
            from app.modules.pipeline.grounding_a0 import collect_candidates

            _rules = json.loads((ep / "visual_world_rules" / "manifest.json")
                                .read_text(encoding="utf-8"))["data"]
            _text_cp = json.loads((ep / "text_cleanup" / "manifest.json")
                                  .read_text(encoding="utf-8"))["data"]
            _pid, _eid = real_ids(ep)
            # ★A0 도 유료다 — 캐시에 넣어야 `--replay` 가 **정말** 무료가 된다.
            #  안 넣었더니 재생인데 A0 만 다시 굽고 있었다.
            from app.modules.pipeline import grounding_a0 as _a0
            _txt = _text_cp.get("cleaned_text") or ""
            # ★A0 도 재생 신원을 건다 — 팩·계약·세계 맥락·원고가 바뀌면
            #  **다른 판**이다. 여기만 key 만 보고 있었다 (Codex).
            _a0_ident = _identity_of(
                "a0", pack=_a0.PROMPT_PACK_VERSION,
                contract=_a0.A0_CONTRACT_VERSION,
                # ★팩 **버전**만 접으면 같은 버전 안에서 지문 bytes 가 바뀐
                #  것을 못 잡는다. 프로덕션과 같은 함수로 실제 payload 를 낸다.
                payload=_a0.planned_payload(
                    _txt, era=str(_rules.get("era") or ""),
                    region=str(_rules.get("region") or "")),
                # ★범위가 다르면 subject id 가 달라진다 — 같은 후보가 아니다.
                scope=f"{_pid}/{_eid}", strict="1",
                # ★A0 는 판정자를 명시로 안 받고 manifest 기본값을 탄다.
                #  그 기본값이 바뀌면 **다른 모델이 건진 후보**다.
                model=_a0.requested_model())
            # ★`["candidates"]` 만 저장하면 **무엇으로 건졌는지**가 사라진다.
            #  full 결과(지문 포함)를 남기고 여기서 꺼내 쓴다.
            a0_cands = cached_call(
                f"a0:{ep_prefix}",
                lambda: collect_candidates(
                    _txt,
                    project_id=_pid, episode_id=_eid,
                    era=str(_rules.get("era") or ""),
                    region=str(_rules.get("region") or ""),
                    strict_single_attempt=True),
                identity=_a0_ident)["candidates"]
        try:
            _pid, _eid = real_ids(ep)
            _built = build_subjects_from_saved_episode(
                ep, project_id=_pid, episode_id=_eid,
                a0_candidates=a0_cands)
            subjects = _built["subjects"]
            # ★못 붙인 것은 분류기에 안 들어간다 — 그래도 **판정을 낸 행**이다.
            #  안 세면 「후보에 없음」으로 찍혀 「원고에 없다」로 오독된다.
            unbound_by_sid = {u.get("_short_id"): u for u in _built["unbound"]
                              if u.get("_short_id")}
            # ★**무엇을 근거로 쟀는지** 적는다. 저장 CP 만 있는 legacy
            #  에피소드는 A0 산출이 없어 전부 `entity_description` 이다 —
            #  production 은 A0 가 건진 **원문 문장**을 넘긴다. 같은 대상을
            #  다른 근거로 판정해 놓고 「production 을 쟀다」로 쓰면 안 된다.
            quote_sources.setdefault(ep_prefix, _built["quote_sources"])
        except ShadowSourceError as exc:
            for sid, want in wants.items():
                rows.append(("stored", labels[sid], want, f"CP 읽기 실패: {exc}", "", "없음"))
            continue
        rules = json.loads((ep / "visual_world_rules" / "manifest.json")
                           .read_text(encoding="utf-8"))["data"]
        smp = cached_classify(
            f"stored:{ep_prefix}",
            subjects=subjects, samples=args.repeats,
            era=str(rules.get("era") or ""), region=str(rules.get("region") or ""),
            strict_single_attempt=True)
        for fp in smp["runs"]:
            _record_run_fp(runs, ep_prefix, fp)
        # ★이름이 아니라 short_id 로 채점한다 — 이름은 에피소드마다 달라진다.
        recs_by_sid: dict[str, list] = {}
        # ★**축마다** 무엇을 근거로 쟀는지. 에피소드 단위로 「하나라도 원문」을
        #  보면 **채점하는 축이 상상 묘사로 재졌는데도 통과**한다.
        basis_by_sid: dict[str, str] = {}
        for subj in subjects:
            sid = (subj.get("provenance") or {}).get("short_id")
            if sid:
                recs_by_sid[sid] = smp["by_subject"].get(
                    subj["research_subject_id"], [])
                basis_by_sid[sid] = subj.get("quote_source") or "없음"
        for sid, want in wants.items():
            if sid in unbound_by_sid:
                u = unbound_by_sid[sid]
                rows.append(("stored", labels[sid], want, "unresolved",
                             u.get("route_override_reason", ""), "미결속"))
                continue
            got = recs_by_sid.get(sid)
            if not got:
                rows.append(("stored", labels[sid], want, "후보에 없음", "", "없음"))
                continue
            route, votes, unstable = _fold_records(got)
            if unstable:
                rows.append(("stored", labels[sid], want, f"흔들림:{route}",
                             votes, basis_by_sid.get(sid, "없음")))
                continue
            # ★★★**여기서 끝나면 옛 계약을 재는 것이다** (Codex).
            #  `decide_route` 의 `research` 는 「조사한다」이지 「시대 차이가
            #  있다」가 아니다 — 최종 답은 **출처**가 낸다.
            _subj = next((x for x in subjects
                          if (x.get("provenance") or {}).get("short_id") == sid
                          or x.get("research_subject_id") == sid), None)
            if _subj is None:
                rows.append(("stored", labels[sid], want, "subject 를 못 찾음",
                             votes, basis_by_sid.get(sid, "없음")))
                continue
            route, why = _final_route(
                route, _subj,
                era=str(rules.get("era") or ""),
                region=str(rules.get("region") or ""),
                cache_key=f"sourced:{ep_prefix}:{sid}",
                model=args.search_model)
            rows.append(("stored", labels[sid], want, route,
                         f"{votes} → {why}", basis_by_sid.get(sid, "없음")))

    # ── 합성 음성 — era 가 서로 달라 각 1콜 ──────────────────────────
    for spec in fx["synthetic_negative"]:
        subj = build_subject(
            project_id="fixture", episode_id=spec["fixture_id"],
            source_anchor=spec["fixture_id"], surface_form=spec["surface_form"],
            owner_type=spec["owner_type"],
            provenance={"fixture_id": spec["fixture_id"]})
        subj["source_quote"] = spec["source_quote"]
        smp = cached_classify(
            f"fixture:{spec['fixture_id']}",
            subjects=[subj], samples=args.repeats, era=spec["era"], region=spec["region"],
            strict_single_attempt=True)
        for fp in smp["runs"]:
            _record_run_fp(runs, spec["fixture_id"], fp)
        got = smp["by_subject"].get(subj["research_subject_id"], [])
        route, votes, unstable = _fold_records(got) if got else ("표본 없음", "", True)
        if unstable:
            rows.append(("synthetic", spec["fixture_id"], "non-research",
                         f"흔들림:{route}", votes, "fixture"))
            continue
        # ★★합성 둘도 **저장 축과 똑같이** 최종 판정을 탄다. 여기만 검색 전
        #  route 를 그대로 쓰면 「검색 ≤9」라고 써 놓고 실제로는 7만 사는
        #  **다른 기구**가 된다 (Codex).
        route, why = _final_route(
            route, subj, era=spec["era"], region=spec["region"],
            cache_key=f"sourced:fixture:{spec['fixture_id']}",
            model=args.search_model)
        rows.append(("synthetic", spec["fixture_id"], "non-research",
                     route, f"{votes} → {why}", "fixture"))

    # ── 채점 ─────────────────────────────────────────────────────────
    ok = mismatch = undecided = 0
    print()
    for axis, label, want, got, votes, basis in rows:
        mark = score_row(want, got)
        if mark == MARK_OK:
            ok += 1
        elif mark == MARK_UNDECIDED:
            undecided += 1
        else:
            mismatch += 1
        print(f"  {mark:8} {axis:9} 기대={want:13} 실제={got:12} "
              f"[{votes}] 근거={basis:19} {label}")

    print(f"\n맞음 {ok} · 어긋남 {mismatch} · 미확정 {undecided}  (총 {len(rows)})")
    # ★**갈림을 따로 센다.** 「미확정 0」과 「갈림 0」은 다른 말이다 —
    #  축 다수로 접은 route split 이 있으면 그것을 안 적고 「갈림 0건」이라
    #  쓰면 안 된다(Codex 지적: 실제로 두 판에 각각 2건이 있었다).
    _sf = sum(1 for r in rows if "[축은 접힘]" in r[4])
    _au = sum(1 for r in rows if "축 미결:" in r[4])
    print(f"축 다수 실패 {_au}건 · 표본 갈림을 다수로 접은 것 {_sf}건")
    if _sf:
        print("  ★「갈림 0건」이라고 쓰면 안 된다 — 위 수를 그대로 적는다")
    # ★기록만 하지 않고 **gate 한다** — 값이 비었거나 판마다 다르면 「같은 것을
    #  N회 쟀다」가 성립하지 않는다. 그러면 흔들림 수치 자체가 못 믿을 것이 된다.
    provenance_bad: list[str] = []
    for key, entries in runs.items():
        by_judge: dict[str, list] = {}
        for e in entries:
            by_judge.setdefault(str(e.get("requested_judge_alias")), []).append(e)
        print(f"  {key[:12]:14} 부탁한 판정자={sorted(by_judge)}")
        for alias, grp in sorted(by_judge.items()):
            print(f"    {alias:14} 실제="
                  f"{sorted({str(e.get('judge_physical_model')) for e in grp})}"
                  f" 팩={sorted({str(e.get('prompt_version')) for e in grp})}")
        # ★**프로덕션 gate 를 그대로 부른다.** 규칙을 도구에 다시 적었더니
        #  본판만 고치고 held-out 을 안 고쳐 여섯 축이 통째로 막혔다.
        for b in check_provenance(entries,
                                  expected_judges=list(DEFAULT_JUDGES),
                                  expected_samples=args.repeats):
            provenance_bad.append(f"{key}: {b}")
    # ★**무엇을 근거로 쟀는지**를 반드시 찍는다. production 은 A0 가 건진
    #  원문 문장을 분류기에 넘긴다. 저장 CP 만 있는 legacy 에피소드는 A0
    #  산출이 없어 전부 상상 묘사다 — 같은 대상을 **다른 근거로** 판정해
    #  놓고 「production 을 쟀다」로 쓰면 안 된다.
    if quote_sources:
        print("\n판정 근거:")
        for ep, srcs in sorted(quote_sources.items()):
            print(f"  {ep[:12]:14} " + " ".join(
                f"{k}={v}" for k, v in sorted(srcs.items())))

    if _budget is not None:
        _snap = _budget.snapshot()
        print(f"\n검색 예산 — 논리 상한 {cap['search']} · 물리 상한 "
              f"{_snap['cap']} · 실제 전송 {_snap['used']} · 거절 "
              f"{_snap['denied']}")
    _stack.close()

    code, reasons = decide_exit(rows, provenance_bad, quote_sources,
                                a0_ran=bool(args.with_a0))
    # ★**떨어져도** 기록을 남긴다 — 떨어진 판이야말로 접기를 고쳐 다시 잴 판이다.
    _save_records(args)
    if code:
        print("\n★통과가 아니다:")
        for r in reasons:
            print(f"  · {r}")
        return code
    print("\n통과 — 아홉 축 전부 맞음")
    return 0


def cli() -> int:
    """★겉옷 하나 — **어떻게 나가든** 주행 예산과 산 기록을 정리한다.

    맨손 `install_budget` 만 하고 정상 종료에서만 내리면, 중간에 예외가 나는
    순간 예산이 그 스레드에 그대로 남는다. 그리고 **산 것도 흘려 쓴다** —
    끝에서 한 번만 쓰면 중간에 서는 순간 통째로 잃는다.
    """
    try:
        return main()
    finally:
        if _RUN_STACK is not None:
            _RUN_STACK.close()
        _flush()


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