"""C(c) **B-only** 진단 주행기 — 구간 N + merge 1. ★기본은 **안 산다**.

Codex 가 최소 실험으로 정했다 (2026-08-31): arm A 는 **안 산다.** arm A 의
후단 payload 는 앞 응답에서 동적으로 생기는데 지금 표는 미리 깔아 둔 것으로
지은 **대역**이라, 그것으로 A 를 사면 「현행 구조」 비교가 아니다. B 가
구조적으로 되는지부터 보고, 구조 gate 가 떨어지면 **A 를 안 사고 끝낸다.**

## 이 도구가 지키는 것 — 보내기 **전에**

    ① 슬롯 수 확인   승인 때 센 수와 다르면 **provider 0회로 선다**.
                     `num_retries=0` 은 Router 재시도만 닫는다 — 키 슬롯
                     loop(`llm_client.py:427-439`)는 그 밖에 있어서 안 닫힌다.
    ② 논리 상한      보내기 전에 센다. 보낸 뒤 세면 이미 산 것이다.
    ③ 신원           payload + **모델 alias + 물리 모델**. alias 만 넣으면
                     설정의 실제 모델만 바뀌었을 때 옛 산출이 재사용된다.
    ④ 장부           호출마다 **바로** 적는다. 한 판 끝나고 적으면 후반이
                     끊길 때 앞서 산 것까지 다시 산다.
    ⑤ 재사용         장부에 같은 신원이 있으면 **안 산다**. `bought` 와
                     `reused` 를 따로 센다 — 뭉치면 재개한 판이 「N회 샀다」
                     로 보고된다.

## 무엇을 사고 무엇을 안 사나

    산다      구간 판독 N + 동일성 판정 1 = **논리 N+1**
    안 산다   검색 · 이미지 생성 · 이미지 검색 · arm A · classifier

    python tools/grounding_audit/cc_runner.py --dry  <journal.json>
    python tools/grounding_audit/cc_runner.py --live <journal.json>   # ★유료
"""
from __future__ import annotations

import hashlib
import json
import sys
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Sequence

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "tests"))

#: ★승인받은 수. 여기를 고치면 승인 범위가 바뀐다 — 코드가 스스로 못 넓힌다.
APPROVED_SLOTS = 2
#: ★**손으로 적은 수**다 — fixture 에서 끌어오지 않는다. 끌어오면 원고가
#:  길어질 때 승인 범위가 **조용히** 넓어진다. 시험이 「구간 수 + merge 1」과
#:  같은지 대조하므로, 원고가 바뀌면 여기서 서고 사람이 다시 정한다.
APPROVED_LOGICAL = 3
MODEL_ALIAS = "gpt"

#: ★요청 계약 — **여기가 한 곳**이다. 앞 판은 이 값이 preflight **문구에만**
#:  있고 주행기가 안 썼다. 문구는 문이 아니다.
PINNED: Dict[str, Any] = {"num_retries": 0, "enable_fallback": False}

#: ★**dispatch 예산** — 우리가 세는 것은 `call_structured` 를 부른 횟수다.
#:  실제 물리 시도는 그 안에서 키 슬롯 loop 로 최대 슬롯 수만큼 갈 수 있다
#:  (`llm_client.py:427-439`, Router **밖**이라 `num_retries=0` 이 안 닫는다).
#:  그래서 **물리 상한 = dispatch 예산 × 슬롯**이고, 우리가 셀 수 있는 것은
#:  왼쪽뿐이다. 실제치는 Opik + provider 로그로 본다.
def dispatch_budget_for(cap: int) -> int:
    return cap


def physical_upper_bound(dispatch_budget: int, slots: int) -> int:
    return dispatch_budget * slots


class CapExceeded(RuntimeError):
    pass


class SlotMismatch(RuntimeError):
    pass


class LockDrift(RuntimeError):
    pass


class NeedsDecision(RuntimeError):
    """앞 판에 **불확실하게 실패한** 호출이 있다. ★자동으로 다시 안 산다."""


class TraceUnavailable(RuntimeError):
    """기록을 못 남기는 상태다. ★기록이 목적인 주행은 여기서 선다."""


def experiment_lock(world_facts: str, slots: int, cap: int,
                    budget: int) -> Dict[str, Any]:
    """이 실험의 **좌표 한 벌**. ★하나라도 어긋나면 provider 0회로 선다.

    앞 판은 `PINNED` 가 preflight **문구에만** 있고 주행기가 안 썼다. 문구는
    문이 아니다 — 잠그려면 값을 한 곳에 모아 **대조**해야 한다.
    """
    from app.modules.pipeline import grounding_chunk as gc

    pack = gc.pack_dir()
    h = hashlib.sha256()
    for f in sorted(pack.iterdir()):
        if f.is_file():
            h.update(f.name.encode("utf-8"))
            h.update(f.read_bytes())
    from app.core.config import settings

    return {
        "pack_version": gc.CHUNK_PACK_VERSION,
        "pack_hash": h.hexdigest()[:16],
        # ★기록 기능이 꺼져 있으면 **장부에 구매 줄을 적기 전에** 선다.
        "opik_trace_v2": bool(getattr(settings, "opik_trace_v2_enabled",
                                      False)),
        "world_hash": hashlib.sha256(
            world_facts.encode("utf-8")).hexdigest()[:16],
        "model_alias": MODEL_ALIAS,
        "model_physical": physical_model(),
        "request_contract": dict(PINNED),
        "slots": slots,
        "logical_cap": cap,
        "dispatch_budget": budget,
        "physical_upper_bound": physical_upper_bound(budget, slots),
    }


def identity(system: str, user: str, schema: Any,
             alias: str, physical: str,
             contract: Optional[Dict[str, Any]] = None) -> str:
    """호출 **신원**. ★단계 이름·순번이 아니라 나가는 것 그 자체다.

    ★`contract`(재시도·대체 tier)도 접는다. 안 접으면 **다른 계약으로 산 것**을
    같은 신원으로 재사용한다 — 재시도를 켠 판과 끈 판이 같은 것이 된다.
    """
    # ★★신원 계산은 **production 모듈 한 곳**이 한다 (Codex 2026-08-31).
    #  도구가 제 것을 따로 만들면 「도구가 자기를 검사하는」 축이 되고,
    #  실제 스텝·재개·config 소비자와 갈린다.
    from app.modules.pipeline.grounding_chunk import acquisition_identity

    return acquisition_identity(
        {"system": system, "parts": [{"type": "text", "text": user}],
         "schema": schema},
        model_alias=alias, model_physical=physical,
        request_contract=contract or PINNED)


def stamp_of(payload: Dict[str, Any], phys: str,
             contract: Optional[Dict[str, Any]]) -> str:
    """해석 지문. ★**계약을 인자로 받는다** — 안에서 `PINNED` 을 안 읽는다.

    ★`run` 과 `replay` 가 각자 적었더니, 조회는 **장부에 적힌 계약**으로 하고
    지문만 **지금 `PINNED`** 으로 찍었다 (Codex NON-BLOCK, 2026-08-31).
    지금은 두 값이 같아서 안 드러나지만, 계약을 바꾼 뒤 옛 장부를 다시 읽으면
    **다른 계약으로 산 응답에 새 계약 지문이 찍힌다** — 하류 무효화가 엉뚱한
    것을 살려 둔다. 계산하는 자리를 하나로 두고 계약을 넘겨받는다.
    """
    from app.modules.pipeline import grounding_chunk as gc

    return gc.processing_stamp(
        gc.acquisition_identity(payload, model_alias=MODEL_ALIAS,
                                model_physical=phys,
                                request_contract=contract or PINNED))


class Journal:
    """호출 장부. ★적을 때마다 **파일까지** 내려쓴다."""

    def __init__(self, path: Path) -> None:
        self.path = path
        self.entries: Dict[str, Dict[str, Any]] = {}
        self._lock: Optional[Dict[str, Any]] = None
        self._nonce: Optional[str] = None
        if path.exists():
            d = json.loads(path.read_text(encoding="utf-8")) or {}
            self._lock = d.get("lock")
            self._nonce = d.get("nonce")
            for e in d.get("calls") or []:
                self.entries[str(e.get("identity"))] = e

    def nonce(self) -> str:
        """이 장부의 **한 번뿐인 표식**. ★재개하면 같고, 새 장부면 다르다.

        잠금만으로 주행 id 를 만들면 같은 잠금의 **옛 판** trace 까지 이번
        판으로 세어진다. 장부에 한 번 적어 두면 재개는 이어지고 새 판은 갈린다.
        """
        if self._nonce is None:
            import uuid

            self._nonce = uuid.uuid4().hex[:12]
            self._flush()
        return self._nonce

    def lock(self) -> Optional[Dict[str, Any]]:
        return self._lock

    def set_lock(self, lock: Dict[str, Any]) -> None:
        self._lock = lock
        self._flush()

    def dispatch_count(self) -> int:
        """★`call_structured` 를 **부른 횟수**. 성공만 세면 실패한 것이 샌다.

        ★이것은 **물리 시도 수가 아니다** — 그 안에서 키 슬롯 loop 가 최대
        슬롯 수만큼 갈 수 있다. 물리 실제치는 Opik+provider 로그로 본다.
        """
        return sum(1 for e in self.entries.values()
                   if e.get("status") in ("ok", "uncertain"))

    def has(self, ident: str) -> bool:
        return ident in self.entries

    def get(self, ident: str) -> Dict[str, Any]:
        return self.entries[ident]

    def put(self, ident: str, rec: Dict[str, Any]) -> None:
        self.entries[ident] = {"identity": ident, **rec}
        self._flush()

    def _flush(self) -> None:
        tmp = self.path.with_suffix(self.path.suffix + ".tmp")
        tmp.write_text(json.dumps(
            {"lock": self._lock, "nonce": self._nonce,
             "calls": list(self.entries.values())},
            ensure_ascii=False, indent=1), encoding="utf-8")
        tmp.replace(self.path)              # ★반쯤 쓰인 장부를 안 남긴다


def check_slots(approved: int = APPROVED_SLOTS) -> int:
    """슬롯 수를 **보내기 전에** 확인한다. 다르면 선다."""
    from app.core import openai_keys

    n = int(openai_keys.slot_count())
    if n != approved:
        raise SlotMismatch(
            f"키 슬롯이 {n}개인데 승인은 {approved}개 기준이다 — "
            f"물리 상한이 {n / max(approved, 1):.0f}배로 달라진다. "
            "provider 를 한 번도 안 부르고 선다")
    return n


def physical_model(alias: str = MODEL_ALIAS) -> str:
    from app.modules.pipeline.era_research import resolve_model_physical

    return resolve_model_physical(alias)


def plan_calls(world_facts: str) -> List[Dict[str, Any]]:
    """살 것 목록. ★merge 는 구간 산출이 나와야 payload 가 정해진다."""
    from app.modules.pipeline import grounding_chunk as gc
    from tests.grounding.fixtures import synthetic_episode as ep

    segs = ep.segment_texts()
    out = []
    for n, bundle in enumerate(ep.bundles()):
        ids = [f"scene-{i}" for i in bundle]
        p = gc.build_chunk_payload(ids, segs, world_facts)
        out.append({"kind": "chunk", "chunk_id": f"c{n}",
                    "segment_ids": ids, "payload": p})
    return out


def run(journal_path: Path,
        send: Callable[[Dict[str, Any], str], Dict[str, Any]],
        *,
        world_facts: str,
        cap: int = APPROVED_LOGICAL,
        approved_slots: Optional[int] = APPROVED_SLOTS,
        dispatch_budget: Optional[int] = None,
        plan: Optional[Sequence[Dict[str, Any]]] = None,
        segments: Optional[Dict[str, str]] = None,
        shot_catalogs: Optional[Dict[str, Sequence[Dict[str, Any]]]] = None,
        lock_extra: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    """구간 N + merge 1. ★`send` 를 안 주면 아무것도 안 나간다.

    Args:
        send: `(payload, identity) -> 응답 dict`. 실제 전송은 **호출부**가
            정한다 — 이 함수 안에 provider 를 안 박는다.
        approved_slots: `None` 이면 슬롯 확인을 건너뛴다(**dry 전용**).
    """
    from app.modules.pipeline import grounding_chunk as gc
    from app.modules.pipeline import grounding_chunk_merge as cm
    from tests.grounding.fixtures import synthetic_episode as ep

    slots = check_slots(approved_slots) if approved_slots is not None \
        else int(approved_slots or 0)
    budget = (dispatch_budget if dispatch_budget is not None
              else dispatch_budget_for(cap))

    jr = Journal(journal_path)
    phys = physical_model()
    # ★계획·원문·샷 목록을 **밖에서 받을 수 있다** — 그래야 합성과 실제
    #  에피소드가 **같은 주행기**를 쓴다. 두 벌로 만들면 한쪽만 고쳐진다.
    segs = segments if segments is not None else ep.segment_texts()
    the_plan = list(plan) if plan is not None else plan_calls(world_facts)
    cats = shot_catalogs or {}

    # ★★실험 잠금 — 하나라도 어긋나면 **provider 0회로 선다.**
    lock = experiment_lock(world_facts, slots, cap, budget)
    if lock_extra:
        # ★실제 에피소드·샷 목록처럼 **그 판을 정하는 좌표**를 잠금에 넣는다.
        lock = {**lock, **lock_extra}
    prev = jr.lock()
    if prev is not None and prev != lock:
        diff = {k: (prev.get(k), lock.get(k)) for k in set(prev) | set(lock)
                if prev.get(k) != lock.get(k)}
        raise LockDrift(
            f"앞 판과 실험 좌표가 다르다 {diff} — 같은 장부에 이어 적으면 "
            "무엇으로 산 것인지가 섞인다. provider 를 한 번도 안 부르고 선다")
    jr.set_lock(lock)

    # ★기록 기능이 처음부터 꺼져 있으면 **구매 줄을 적기 전에** 선다
    #  (Codex NON-BLOCK). 안 그러면 sender 에서 서긴 하지만 장부에
    #  `uncertain` 이 남아, 「샀을지도 모른다」로 읽힌다 — 실제로는 안 샀다.
    if not lock.get("opik_trace_v2"):
        raise TraceUnavailable(
            "기록 기능(`opik_trace_v2`)이 꺼져 있다 — 결속 키 없이 사면 "
            "무엇을 샀는지 못 되짚는다. 장부에 아무것도 안 적고 선다")

    # ★★앞 판에 **불확실하게 실패한** 것이 있으면 자동으로 다시 안 산다.
    #  「보냈는데 답을 못 받은 것」은 **샀을 수도 있다** — 다시 보내면
    #  계획보다 많이 나가고, 물리 상한을 넘긴다 (Codex).
    pend = [e for e in jr.entries.values() if e.get("status") == "uncertain"]
    if pend:
        raise NeedsDecision(
            f"앞 판에 답을 못 받은 호출 {len(pend)}건이 있다 "
            f"({[e['identity'][:8] for e in pend]}) — 샀는지 아닌지 모른다. "
            "Opik 을 보고 사람이 정한 뒤 장부를 고쳐야 한다. 자동 재구매 안 한다")

    rid = run_id(lock, jr.nonce())
    bought = reused = 0
    sent = jr.dispatch_count()      # ★예산은 **장부 전체**로 센다

    def _one(payload: Dict[str, Any]) -> Dict[str, Any]:
        nonlocal bought, reused, sent
        ident = identity(payload["system"], payload["parts"][0]["text"],
                         payload["schema"], MODEL_ALIAS, phys, PINNED)
        if jr.has(ident) and jr.get(ident).get("status") == "ok":
            reused += 1
            return jr.get(ident)["response"]
        # ★②논리 상한은 **보내기 전에**.
        if bought + 1 > cap:
            raise CapExceeded(f"논리 상한 {cap} 을 넘는다 — 여기서 선다")
        # ★③dispatch 예산도 **보내기 전에**. 재개 누적까지 센다.
        if sent + 1 > budget:
            raise CapExceeded(
                f"dispatch 예산 {budget} 을 넘는다 (지금까지 {sent}) — "
                "여기서 선다. ★물리 시도는 이보다 클 수 있다(키 슬롯 loop)")
        # ★★보내기 **직전에** 「나갔다」를 적는다. 예외가 나도 이 줄은 남는다.
        jr.put(ident, {"status": "uncertain", "model_alias": MODEL_ALIAS,
                       "model_physical": phys,
                       "bytes": len(payload["parts"][0]["text"].encode()),
                       "response": None})
        sent += 1
        try:
            resp = (send(payload, ident, rid)
                    if _takes_run_id(send) else send(payload, ident))
        except Exception as exc:            # noqa: BLE001
            jr.put(ident, {**jr.get(ident), "status": "uncertain",
                           "error": f"{type(exc).__name__}: {exc}"})
            raise
        bought += 1
        jr.put(ident, {**jr.get(ident), "status": "ok", "response": resp})
        return resp

    rows: List[Dict[str, Any]] = []
    raw: List[Dict[str, Any]] = []
    quarantined: List[Dict[str, Any]] = []
    stamps: Dict[str, str] = {}
    for c in the_plan:
        resp = _one(c["payload"])
        raw.append({"chunk_id": c["chunk_id"], "response": resp})
        got = gc.resolve_rows(resp.get("rows") or [],
                              chunk_id=c["chunk_id"],
                              segment_ids=c["segment_ids"], segments=segs,
                              shot_catalog=cats.get(c["chunk_id"]))
        rows += got["rows"]
        quarantined += got["quarantined"]
        stamps[c["chunk_id"]] = stamp_of(c["payload"], phys, PINNED)

    mresp = _one(gc.build_merge_payload(rows))
    decisions = mresp.get("decisions") or []
    reduced = cm.reduce_episode(rows, decisions, segments=segs)
    return {"rows": rows, "raw": raw, "decisions": decisions,
            "quarantined": quarantined,
            "processing_contract": gc.PROCESSING_CONTRACT_VERSION,
            "reduced": reduced, "bought": bought, "reused": reused,
            "logical": bought + reused, "dispatched": sent, "run_id": rid,
            # ★해석 지문 — 하류가 이걸 보고 stale 을 판단한다.
            #  ★획득 신원에는 안 접힌다(후처리만 바뀌면 재구매 0).
            "processing_stamps": stamps,
            "dispatch_budget": budget,
            "model_physical": phys, "lock": lock}


def replay(journal_path: Path, *, world_facts: str,
           plan: Optional[Sequence[Dict[str, Any]]] = None,
           segments: Optional[Dict[str, str]] = None,
           shot_catalogs: Optional[Dict[str, Sequence[Dict[str, Any]]]] = None
           ) -> Dict[str, Any]:
    """얼어붙은 장부만으로 산출을 **다시 만든다**. ★보낼 길이 **없다**.

    `send` 인자가 아예 없다 — 「안 샀다」를 믿음이 아니라 **구조**로 만든다.
    장부에 없는 호출이 필요하면 그 자리에서 선다(사지 않는다).

    ★잠금은 장부의 것을 **그대로** 쓴다. 슬롯 수를 다시 재지 않는다 — 아무것도
    안 보내므로 잴 이유가 없고, 재면 기계 사정으로 재채점이 막힌다.
    """
    from app.modules.pipeline import grounding_chunk as gc
    from app.modules.pipeline import grounding_chunk_merge as cm
    from tests.grounding.fixtures import synthetic_episode as ep

    jr = Journal(journal_path)
    saved = jr.lock() or {}
    phys = str(saved.get("model_physical") or physical_model())
    contract = saved.get("request_contract") or PINNED
    segs = segments if segments is not None else ep.segment_texts()
    the_plan = list(plan) if plan is not None else plan_calls(world_facts)
    cats = shot_catalogs or {}

    def _cached(payload: Dict[str, Any]) -> Dict[str, Any]:
        ident = identity(payload["system"], payload["parts"][0]["text"],
                         payload["schema"], MODEL_ALIAS, phys, contract)
        rec = jr.entries.get(ident)
        if not rec or rec.get("status") != "ok":
            raise LookupError(
                f"장부에 {ident[:10]} 응답이 없다 — 재채점은 **사지 않는다**. "
                "이 호출이 필요하면 사람이 새로 승인해야 한다")
        return rec["response"]

    rows: List[Dict[str, Any]] = []
    raw: List[Dict[str, Any]] = []
    quarantined: List[Dict[str, Any]] = []
    stamps: Dict[str, str] = {}
    for c in the_plan:
        resp = _cached(c["payload"])
        raw.append({"chunk_id": c["chunk_id"], "response": resp})
        got = gc.resolve_rows(resp.get("rows") or [], chunk_id=c["chunk_id"],
                              segment_ids=c["segment_ids"], segments=segs,
                              shot_catalog=cats.get(c["chunk_id"]))
        rows += got["rows"]
        quarantined += got["quarantined"]
        stamps[c["chunk_id"]] = stamp_of(c["payload"], phys, contract)

    mresp = _cached(gc.build_merge_payload(rows))
    decisions = mresp.get("decisions") or []
    reduced = cm.reduce_episode(rows, decisions, segments=segs)
    return {"rows": rows, "raw": raw, "decisions": decisions,
            "quarantined": quarantined,
            "processing_contract": gc.PROCESSING_CONTRACT_VERSION,
            "reduced": reduced, "bought": 0,
            "reused": len(raw) + 1, "logical": len(raw) + 1,
            "dispatched": 0, "dispatch_budget": 0,
            "processing_stamps": stamps,
            "run_id": run_id(saved, jr.nonce()) if saved else "",
            "model_physical": phys, "lock": saved}


def _takes_run_id(fn: Callable) -> bool:
    """`send` 가 주행 표식을 받나. ★시험용 `send` 는 두 인자만 받는다."""
    import inspect

    try:
        return len(inspect.signature(fn).parameters) >= 3
    except (TypeError, ValueError):
        return False


def _dry_send(payload: Dict[str, Any], ident: str) -> Dict[str, Any]:
    """dry 응답 — ★**빈 것**을 돌려준다. 그럴듯한 답을 지어내지 않는다.

    지어낸 답으로 초록을 보면, 재는 것은 **내 상상**이지 모델이 아니다.
    dry 가 증명하는 것은 **문·장부·신원·상한**이지 산출 품질이 아니다.
    """
    return {"rows": [], "decisions": []}


#: ★Opik 에서 이 실험 것만 골라내는 표식. 유료 뒤 대조에 쓴다.
LIVE_STEP = "grounding_chunk_cc"
LIVE_TAG = "op:cc-b-only"

#: 이 실험의 부모 trace 이름. ★신원은 **여기 metadata** 로 간다.
LIVE_TRACE_NAME = "cc_b_only_chunk"
LIVE_THREAD = "cc-b-only-synthetic"

#: ★★신원을 **tag 로 실을 수 없다.** 두 번 막힌다 —
#:  ① litellm 이 `metadata["opik"]` 에서 읽는 것은 넷뿐이라(`project_name`·
#:     `current_span_data`·`tags`·`thread_id`) 자유 키는 버려진다.
#:     실측: trace 에 남은 것은 `tags: ['gemini', 'op:audit']` 뿐이었다.
#:  ② tag 로 옮겨도 **축 whitelist**(`step`·`op`·`kind`·`model`·`provider`·
#:     `status`)가 거른다. 그 문은 **고카디널리티 태그를 막으려고** 있고,
#:     호출 신원이 정확히 그것이다 — 넓히면 안 되는 가드다.
#:  그래서 `open_trace` 로 **부모 trace** 를 열고 그 metadata 에 싣는다.
#:  그쪽은 Opik client 로 직접 가서 litellm 필터를 안 탄다.
ID_META_KEY = "cc_call_identity"

#: ★**주행 범위**. 이것 없이 tag 만 보면 옛 판과 딴 도구 흔적까지 같이 잡힌다
#:  — 실제로 내 `probe` trace 하나 때문에 5↔5 가 맞는데도 「안 맞는다」가 났다.
RUN_META_KEY = "cc_run_id"


def run_id(lock: Dict[str, Any], nonce: str) -> str:
    """이번 주행의 표식. ★잠금 + 장부 표식 — 둘 다 바뀌면 다른 주행이다."""
    h = hashlib.sha256()
    h.update(json.dumps(lock, sort_keys=True, ensure_ascii=False)
             .encode("utf-8"))
    h.update(b"\x00")
    h.update(nonce.encode("utf-8"))
    return h.hexdigest()[:16]


def live_send(payload: Dict[str, Any], ident: str, rid: str = "", *,
              trace_name: str = LIVE_TRACE_NAME, tag: str = LIVE_TAG,
              thread: str = LIVE_THREAD, step: str = LIVE_STEP
              ) -> Dict[str, Any]:
    """★실제 전송. **계약을 소비한다** — `PINNED` 을 그대로 넘긴다.

    앞 판은 `PINNED` 이 preflight 문구에만 있고 아무도 안 썼다. 여기서
    `num_retries`·`enable_fallback` 을 **인자로** 넘겨야 잠근 것이 실효다.

    ★`opik_metadata` 에 신원과 표식을 남긴다 — 유료 뒤에 Opik·장부·provider
    로그를 **맞대어** 보려면 셋이 같은 열쇠를 들고 있어야 한다.
    """
    from app.modules.llm.llm_client import call_structured
    from app.modules.llm.opik_trace import open_trace

    # ★★**부모 trace 가 안 열리면 안 산다** (Codex BLOCK, 2026-08-31).
    #  `open_trace` 는 설정이 꺼졌거나 client 생성이 실패하면 **예외 없이
    #  None** 을 준다 — production 에서는 그게 맞다(기록이 본 작업을 막으면
    #  안 된다). 그런데 **이 실험**은 기록이 목적이라 반대다: 결속 키가 없는
    #  호출을 5건 사고 나서 대조기가 실패하면, 산 것을 못 되짚는다.
    #  ★production 의 non-fatal 계약은 안 건드리고 **이 sender 만** 엄격히 한다.
    # ★★trace 좌표를 **인자로** 연다 (Codex 2026-08-31). 실제 에피소드 C 를
    #  합성 B 이름으로 기록하면 감사 결과를 오독한다. sender 를 복사하지 않고
    #  좌표만 바꾼다 — 복사하면 두 벌이 되어 한쪽만 고쳐진다.
    with open_trace(name=trace_name, tags=[tag],
                    metadata={ID_META_KEY: ident, RUN_META_KEY: rid,
                              "model_alias": MODEL_ALIAS},
                    thread_id=thread,
                    input_data={"identity": ident}) as trace:
        if trace is None:
            raise TraceUnavailable(
                "부모 Opik trace 를 못 열었다 — 결속 키 없이 사면 나중에 "
                "무엇을 샀는지 못 되짚는다. provider 를 안 부르고 선다")
        return call_structured(
            step,
            payload["system"],
            payload["parts"][0]["text"],
            payload["schema"],
            project_config={step: {"model": MODEL_ALIAS}},
            schema_name=step,
            opik_metadata={"tags": [tag]},
            enable_fallback=bool(PINNED["enable_fallback"]),
            num_retries=int(PINNED["num_retries"]),
        )


def main() -> int:
    from tools.grounding_audit.call_payload_table import seal_outbound

    if len(sys.argv) < 3 or sys.argv[1] not in ("--dry", "--live"):
        print(__doc__)
        return 2
    from tests.grounding.fixtures import synthetic_episode as ep

    mode, jp = sys.argv[1], Path(sys.argv[2])
    # ★세계관은 **fixture 가 갖는다.** 여기 문자열을 적으면 두 벌이 되고,
    #  두 축의 정답 근거와 실제 입력이 갈린다 (Codex BLOCK-1).
    world = ep.WORLD_FACTS
    budget = dispatch_budget_for(APPROVED_LOGICAL)

    if mode == "--live":
        # ★유료. 슬롯을 **보내기 전에** 세고, 승인 수와 다르면 0회로 선다.
        got = run(jp, live_send, world_facts=world, cap=APPROVED_LOGICAL,
                  approved_slots=APPROVED_SLOTS, dispatch_budget=budget)
        return _report(got, jp, live=True)

    seal_outbound()                          # ★import 뒤에 잠근다
    got = run(jp, _dry_send, world_facts=world, approved_slots=None,
              dispatch_budget=budget)
    return _report(got, jp, live=False)


def score_run(got: Dict[str, Any]) -> Dict[str, Any]:
    """주행 산출을 **바로 채점**한다. ★사람 검토 전에는 최종 판정이 없다."""
    from tests.grounding.fixtures import synthetic_episode as ep
    from tools.grounding_audit import cc_scorer as sc

    red = got["reduced"]
    return sc.score(ep.EXPECTED_TARGETS, ep.target_spans(), red["rows"],
                    segments=ep.segment_texts(), relations=red["part_of"],
                    registered=red["registered"],
                    quarantined=got.get("quarantined") or [])


def _axis_review(got: Dict[str, Any]) -> List[Dict[str, Any]]:
    """두 축 **판정 대상 전부**를 실제 산출 행·flag 와 묶어 낸다.

    ★양성(hard 여야 하는 것)만이 아니라 **음성**(hard 가 아니어야 하는 것)도
    올린다. 음성을 빼면 「다 어렵다」고 답한 모델이 그냥 통과한다.
    """
    from tests.grounding.fixtures import synthetic_episode as ep
    from tools.grounding_audit import cc_scorer as sc

    rows = got["reduced"]["rows"]
    reg = got["reduced"]["registered"]
    spans = ep.target_spans()
    out = []
    for key, basis in ep.AXIS_BASIS.items():
        cover = sc.rows_covering(spans.get(key) or [], rows)
        got_rows = [{"local_id": str(r.get("local_id")),
                     "surface_form": r.get("surface_form"),
                     "owner_type": r.get("owner_type"),
                     "hard_to_generate": r.get("hard_to_generate"),
                     "viewers_would_notice": r.get("viewers_would_notice"),
                     "registered": (reg.get(str(r.get("local_id"))) or {})
                     .get("reason")}
                    for r in rows if str(r.get("local_id")) in cover]
        out.append({"target": key, "expected_hard": basis["hard"],
                    "expected_notice": basis["notice"],
                    "basis": basis["basis"], "produced": got_rows,
                    "human_verdict": None})
    return out


def _report(got: Dict[str, Any], jp: Path, *, live: bool) -> int:
    """★★채점과 **사람 검토 산출**을 같이 낸다 (Codex BLOCK-4).

    앞 판 `--live` 는 줄인 counts 만 찍고 채점기를 안 불러서, 전부 미확정이어도
    `exit 0` 이었다 — 「샀고 안 죽었다」가 「통과」로 읽힌다.
    """
    from tests.grounding.fixtures import synthetic_episode as ep

    kind = "live" if live else "dry"
    sc_out = score_run(got)
    out = jp.with_name(jp.stem + "_score.json")
    out.write_text(json.dumps({
        "kind": kind, "lock": got["lock"],
        # ★후처리 계약은 **파생 장부**에만 남긴다 — 획득 신원·잠금에 안 접는다
        #  (Codex). 실데이터를 본 뒤 고친 것이라 접으면 다시 사야 한다.
        "processing_contract": got.get("processing_contract"),
        "quarantined": got.get("quarantined") or [],
        "logical_dispatch": got["logical"], "bought": got["bought"],
        "reused": got["reused"], "dispatch_budget": got["dispatch_budget"],
        "score": sc_out,
        # ★★사람이 볼 것을 **근거와 함께** 남긴다.
        #  ★`needs_human` 만 싣던 것을 고쳤다 (Codex). 그건 `unresolved` 만
        #   모으므로, 모델이 두 축 **모양**을 우연히 맞추면 그 축이 `ok` 가
        #   되어 사람 검토에서 **빠진다**. 이 주행의 핵심이 그 두 축의
        #   **뜻**이라, 판정 대상은 양성·음성 **전부** 항상 올린다.
        "human_review": {
            "required": True,
            "machine_flagged_axes": sc_out["needs_human"],
            "two_axis_judgements": _axis_review(got),
            "unlisted_rows": [f["rows"] for f in sc_out["findings"]
                              if f["axis"] == "unlisted_rows"],
            # ★격리 행은 **전부** 사람 검토에 올린다 — 조용한 삭제 0.
            "quarantined_rows": got.get("quarantined") or [],
            "rows": got["reduced"]["rows"],
            "verdict": None,
            "how_to_record": ("사람이 본 뒤 이 파일의 human_review.verdict 를 "
                              "채우고, score.final_candidate 를 따로 적는다. "
                              "★자동은 final_candidate 를 만들지 않는다"),
        },
    }, ensure_ascii=False, indent=1), encoding="utf-8")

    print(f"■ {kind} — 논리 {got['logical']} "
          f"(산 것 {got['bought']} · 재사용 {got['reused']}) · "
          f"dispatch {got['dispatched']}/{got['dispatch_budget']}")
    ub = got["lock"]["physical_upper_bound"]
    print(f"  ★dispatch 는 `call_structured` 를 **부른 횟수**다. 실제 물리 "
          "시도는 안 쟀다 — " + (f"상한 ≤{ub} (dispatch×슬롯)"
                                if ub else "★슬롯을 안 세어 상한 **미확인**")
          + ", 실제치는 Opik+provider 로그로 본다.")
    print(f"  물리 모델 {got['model_physical']} · 장부 {jp}")
    q = got.get("quarantined") or []
    print(f"  채점 {sc_out['counts']} · 격리된 행 **{len(q)}**"
          + (f" {[x['surface_form'][:10] for x in q][:4]}" if q else ""))
    print(f"  후처리 계약 {got.get('processing_contract')} "
          "(★획득 신원·잠금에는 안 접는다)")
    print(f"  기계 후보 {sc_out['mechanical_candidate']} · "
          f"최종 **{sc_out['final_candidate']}** (사람이 보기 전엔 없다)")
    print(f"  기계가 못 정한 축 {sc_out['needs_human']}")
    print(f"  ★사람이 볼 두 축 판정 **{len(_axis_review(got))}건** — 기계가 "
          "「맞다」고 한 것도 포함한다. 모양이 맞아도 **뜻**은 사람이 본다.")
    print(f"  검토 산출 {out}")
    if live:
        print(f"  ★Opik 대조 — tag={LIVE_TAG} · trace={LIVE_TRACE_NAME} · "
              f"신원은 부모 trace metadata `{ID_META_KEY}`")
        print("  ★DB/체크포인트 대조는 **N/A** — 합성 B-only 는 production "
              "프로젝트를 안 쓴다. production canary 때 닫는다.")
    # ★기계 통과라도 사람 검토 전에는 **PASS 가 아니다.**
    return _exit_code(sc_out)


#: ★종료코드 — **CLI 성공과 acceptance 를 가른다** (Codex).
#:  자동은 `final_candidate` 를 **못 만든다.** 그래서 0 은 사람 판정 파일이
#:  있을 때만 난다 — 「돌았고 안 죽었다」가 「통과」로 읽히는 길을 없앤다.
EXIT_PASS = 0
EXIT_MACHINE_FAIL = 1
EXIT_HUMAN_REVIEW_REQUIRED = 3


def _exit_code(sc_out: Dict[str, Any]) -> int:
    if not sc_out["mechanical_candidate"]:
        return EXIT_MACHINE_FAIL
    if sc_out.get("final_candidate") is None:
        return EXIT_HUMAN_REVIEW_REQUIRED
    return EXIT_PASS


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