"""고증 canary — C(c) 판독 → **참조 획득** → 사람 비교 화면. ★`--live` 는 유료.

## 무엇을 보나

    구간 판독      합성 원고 2구간 + merge 1 = 논리 3
    참조 획득      hard AND notice 인 대상마다
                     라운드 ≤2 × (검색 1 + 거친 선택 1, **1심**)
    비교 화면      대상마다 원문 근거 · 검색 후보 · 종류 선택 · 고른 참조

## 자동으로 재는 것 / 사람이 보는 것

    자동   호출 수 · 장부 · 재개 · 상한 · 「없다」와 「못 봤다」의 구분 ·
           다섯 갈래가 다 지났나
    사람만 고증이 맞나 · 참조가 그림을 낫게 했나 · 이미지 품질

★VLM 은 **종류와 가시성**만 본다. 시대·국가·제조사·모델·정확성·좋고 나쁨은
안 묻는다 — 그건 사람 몫이다.

    python tools/grounding_audit/ref_canary.py --dry  <journal.json>
    python tools/grounding_audit/ref_canary.py --live <journal.json>   # ★유료
    …  [only=<id,id>]  [episode=<episode_id>]  [fixture=period]  [sample=N]
       [preflight=<preflight.json>]  ← ★`--live` 는 표본이 이것과 같아야 산다

★`episode=` 를 주면 **실제 원고**로 돈다.
★`fixture=period` 는 **1960년대 한국** 시대 원고다 — 다섯 갈래가 자연스럽게
 다 고증 대상이 되고, **실제로 있던 것**이라 사진이 세상에 있어 사람이 눈으로
 판정할 수 있다.
★아무것도 안 주면 지어낸 지역 fixture 다 — 그것으로는 **고증 정확성을 못
 잰다**(찾을 정답 사진이 세상에 없다, 2026-08-31 실측).
"""
from __future__ import annotations

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

ROOT = Path(__file__).resolve().parents[2]
#: ★서빙 뿌리 — 저장소 뿌리다. `backend/` 로 두면 `artifact/` 산출이 뿌리
#:  **밖**이 되어 경로가 `../` 로 새고 화면이 파일을 못 연다 (실측 2026-08-31).
SERVE_ROOT = ROOT.parent
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "tests"))

from tools.grounding_audit import cc_runner as rr  # noqa: E402
from tools.grounding_audit.call_payload_table import seal_outbound  # noqa: E402

#: ★**손으로 적은 승인 수.** 계획에서 계산해 넓히지 않는다.
#:  구간 2 + merge 1 = 3
CHUNK_LOGICAL = 3
#: 대상마다 라운드 ≤2 × (**저작 1** + 검색 1 + 선택 1).
#:
#: ★★저작 단계를 **처음엔 건너뛰었다**. 그 결과 두 가지가 났다
#:  (2026-08-31 실측) —
#:
#:  ①잔부분에 잔부분 질의가 나갔다 — 저작 팩이 「부수 부분은 버리고 주 대상
#:    하나만」을 시키는데 그 단계가 없었다
#:  ②**2라운드가 죽었다** — 좁힘 문안이 **전부 영어**인데 그것을 한국어
#:    지시문 뒤에 붙였다. 검색 팩은 「지시문에 영어 한 단어도 쓰지 마라」고
#:    못박는다. 실측: R1 은 64%가 질의를 냈는데 **R2 는 10%**.
#:
#:  좁힘 문안은 **영어 저작기에게** 주는 것이다
#:  (`outdoor_structure_form_reference_step:1012` 이 그렇게 한다).
PER_TARGET_LOGICAL = 6

#: ★★**손으로 적은 대상 수 상한.** 두 축은 모델이 runtime 에 정하므로 대상
#:  수가 계획보다 많을 수 있다 — 그때 계획에서 다시 계산해 상한을 넓히면
#:  **승인한 수와 갈린다**(앞서 실제로 겪었다). 넘으면 **안 사고 선다**.
#:
#:  ★★수를 **실측 뒤에 손으로** 고쳤다 (2026-08-31). 첫 유료 판독에서 실제
#:   대상이 **13개**로 나와 12 에 걸려 섰다 — 게이트가 제 일을 했다.
#:
#:   그래서 16 으로 올린다. 근거: 실측 13 + 여유 3. **계획에서 계산해 넓힌
#:   것이 아니라** 실제 수를 보고 사람 권한 아래 손으로 적은 것이다.
#:   (사용자 2026-08-31: 「필요하면 유료화 해도 되, 비용 걱정마」)
#:
#:   ★13개 중 8개가 `location_part` 이고 그중 여럿이 표지·값판의 **잔부분**
#:    (「위에서 아래로 내려가는 글자」·「글자 왼편에 그어진 짧은 가로줄」·
#:    「가로로 길게 켠 나무판」)이다. 과분해 여부는 **사람이 화면에서** 본다.
MAX_ACQUISITION_TARGETS = 16

#: `call_structured` 의 fallback 겹. ★코드가 「after all 3 tiers」로 적고
#:  tier1·tier2 라벨 뒤에 gpt_fallback 이 하나 더 붙는다.
CALL_STRUCTURED_TIERS = 3

#: ★★★**물리 여유 배수** — 손으로 적는다. 최악값에서 계산해 넓히지 않는다.
#:
#:  내가 「1회」로 센 논리 호출이 실제로는 **여러 번 나간다**. 겹이 셋 —
#:  `call_structured` 의 tier fallback · `_completion` 의 키 슬롯 loop ·
#:  litellm Router 의 `num_retries`. 실측 상수로 논리 1이 물리 **24**까지
#:  간다(3 × 2 × 4).
#:
#:  ★★그런데 **참조 획득 구간은 예산을 한 번도 안 켰다** (2026-08-31 발견).
#:   `reserve_current_research_call` 은 `research_calls_armed()` 밖에서
#:   **아무 일도 안 한다** — 나가는 자리에 상한이 **없었다**.
#:
#:  2 = 「논리마다 재시도 한 번까지」. 닿으면 **선다** — 그때 왜 그렇게 많이
#:  나갔는지 사람이 본다.
APPROVED_PHYSICAL_FACTOR = 2

#: ★★★**요청 계약** — 재시도 겹을 **실제 인자로** 잠근다 (Codex 2026-08-31).
#:
#:  ①`num_retries=0` — Router 가 기본 3회 더 보낸다. 돈만 문제가 아니다:
#:    `usage_sink` 는 **마지막 응답만** 보므로 앞선 과금 시도가 **기록에서
#:    사라진다**(llm_client 주석). Opik·provider·장부를 맞대야 하는 판에서
#:    보이지 않는 전송이 있으면 대조가 거짓이 된다.
#:  ②`enable_fallback=False` — tier fallback 은 **다른 모델**로 갈아탄다.
#:    장부에는 모델 X 라고 적혀 있는데 답은 Y 가 낸 것이 된다.
#:
#:  ★이 계약이 바뀌면 **다른 방식으로 사는 것**이므로 신원도 움직여야 한다 —
#:   그래서 `brief_outbound` 가 이것을 함께 낸다.
REQUEST_LOCK: Dict[str, Any] = {"num_retries": 0, "enable_fallback": False}

#: ★★**SDK 자체 재시도**도 0 으로. `reserve` **아래**라 예산이 못 보는 겹이다
#:  — 잠그지 않으면 「물리 상한 60」이 거짓이 된다(Codex 2026-08-31).
SDK_LOCK: Dict[str, Any] = {"max_retries": 0}

#: 승인된 키 슬롯 수. ★다르면 **선다** — 승인 장부와 갈리면 안 된다.
APPROVED_SLOT_COUNT = 2

#: 참조 획득 장부 이름. ★C(c) 판독 장부와 **갈라 둔다** — 신원 체계가 다르다.
REF_JOURNAL_NAME = "_ref_journal.json"

#: Opik 좌표 — C(c) 판독과 **갈라 적는다**.
REF_TRACE_NAME = "ref_canary_acquire"
REF_TAG = "op:ref-canary"
REF_THREAD = "ref-canary"
#: 단계·라운드별 사건 장부. ★참조 장부(최종 1줄)와 **다른 것**이다.
REF_EVENTS_NAME = "_ref_events.jsonl"


class ApprovedScopeMismatch(RuntimeError):
    """계획이 승인 수와 다르다. ★장부도 provider 도 안 열고 선다."""


def targets_from(reduced: Dict[str, Any]) -> List[Dict[str, Any]]:
    """참조를 사야 하는 행. ★**같은 행에서 hard AND notice**.

    ★`route` 나 `generation_difficulty` 로 고르지 않는다 — 그것은 옛 다리다.
    두 축이 그 자리를 대신한다.

    ## ★★통째가 이미 대상이면 **잔부분은 안 산다** (실측 2026-08-31)

    첫 유료 판에서 대상 13개 중 **8개가 다른 대상의 부분**이었다. 그리고 못
    고른 3개가 **전부 그 8개 안**이었다 —

        위에서 아래로 내려가는 글자    ⊂ 정류장 표지   질의가 **아예 안 나감**
        글자 왼편에 그어진 짧은 가로줄  ⊂ 정류장 표지   질의가 **아예 안 나감**
        대합실 안                  ⊂ 정류장 승강장  0장

    표지를 찍은 사진에 그 글자와 가로줄이 **이미 들어 있다.** 잔부분을 따로
    찾는 것은 살 수 없는 것을 사려는 것이고, 검색 모델도 「찾을 물건이
    아니다」로 보고 질의를 안 냈다.

    ★기존 검색 팩도 같은 말을 한다 — 「여러 부분이 나열돼 있으면 **주 대상
    하나만** 골라 쓰고 부수 부분·부착물은 버려라」(`brief_system.md`).
    그것을 **질의 자리에서** 하던 것을 **대상 자리로** 올린 것이다.

    ★관계는 **기록된 `part_of`** 로만 본다 — 이름·부분문자열로 짐작하지
    않는다. 부모가 대상이 **아니면** 그 부분은 그대로 산다(잃지 않는다).
    """
    out = []
    reg = reduced.get("registered") or {}
    both = {str(r.get("local_id"))
            for r in (reduced.get("rows") or ())
            if r.get("hard_to_generate") and r.get("viewers_would_notice")}
    # ★부모가 **같이 대상인** 부분만 뺀다
    covered = {str(x.get("part")) for x in (reduced.get("part_of") or ())
               if str(x.get("part")) in both and str(x.get("whole")) in both}
    for r in reduced.get("rows") or ():
        lid = str(r.get("local_id") or "")
        if not (r.get("hard_to_generate") and r.get("viewers_would_notice")):
            continue
        if lid in covered:
            continue
        rec = reg.get(lid) or {}
        out.append({
            "subject_id": rec.get("final_id") or lid,
            "local_id": lid,
            "owner_type": r.get("owner_type"),
            "surface_form": r.get("surface_form"),
            # ★★★**칸을 더하면 그 칸을 복사하는 줄을 찾아라** (2026-08-31 실측).
            #  팩 `4.202608311430` 에 `coarse_type_label` 을 required 로 넣고
            #  `kind_name_of` 가 그것만 쓰게 고쳤는데, **여기서 안 옮겼다**.
            #  그래서 유료 판 12개 대상 **전부** label 이 None 이 되고,
            #  fallback 으로 심판이 `surface_form`(「됫박」)을 그대로 받았다 —
            #  사용자가 막으라던 「VLM 이 판단 못 하는 협소한 것」 바로 그것이다.
            #  모델은 46/46 냈고 `resolve_rows`·`reduce_episode` 도 38/38 을
            #  들고 왔는데 **이 한 줄이 없어서** 통째로 사라졌다.
            "coarse_type_label": str(r.get("coarse_type_label") or ""),
            "visual_brief": r.get("visual_brief") or "",
            "terms_native": list(r.get("search_terms_native") or ()),
            "language_lock_native": str(r.get("language_lock_native") or ""),
            "occurrences": r.get("occurrences") or [],
            "registered": rec.get("registered"),
            "final_id": rec.get("final_id"),
        })
    return sorted(out, key=lambda x: str(x["local_id"]))


def parts_skipped(reduced: Dict[str, Any]) -> List[Dict[str, str]]:
    """통째가 대상이라 **안 산** 부분들. ★조용히 사라지지 않게 남긴다."""
    by = {str(r.get("local_id")): r for r in (reduced.get("rows") or ())}
    both = {lid for lid, r in by.items()
            if r.get("hard_to_generate") and r.get("viewers_would_notice")}
    out = []
    for x in reduced.get("part_of") or ():
        p, w = str(x.get("part")), str(x.get("whole"))
        if p in both and w in both:
            out.append({"part": p, "part_form": by[p].get("surface_form"),
                        "whole": w, "whole_form": by[w].get("surface_form")})
    return sorted(out, key=lambda d: str(d["part"]))


def build_directive(t: Dict[str, Any]) -> str:
    """무엇을 찾을지. ★**이름만** 쓴다 — production 경로와 같게.

    ★★첫 판은 `surface_form` **+ `visual_brief`** 를 통째로 넣었다. 그랬더니
    검색 모델이 긴 설명을 **따옴표로 묶은 구절 질의**로 바꿨고 결과가 **0장**
    이었다 (실측 2026-08-31):

        '"세로 글자" "왼쪽 가로줄" 정류장 표지'   → 0장
        '한쪽 열린 정류장 대기실 벽걸이 긴 벤치'    → 4장

    production 경로(`era_research.research_reference`)는 `directive_native=name`
    으로 **이름만** 넘긴다. 겉모습은 질의가 아니라 **판정 자료**다.

    ★코드가 문장을 짓지 않는다 — 모델이 낸 이름과 낱말을 그대로 쓴다.
    """
    return str(t.get("surface_form") or "").strip()


def real_plan(episode_id: str):
    """실제 원고의 계획. ★selector·조립은 **preflight 것을 그대로** 쓴다."""
    from app.core.world_context import build_grounding_world_facts
    from app.modules.pipeline import grounding_chunk as gc
    from app.modules.pipeline import grounding_shot_catalog as sc
    from tools.grounding_audit import cc_c_preflight as pf

    hits = [c for c in pf.candidates()
            if str(c["episode_id"]).startswith(episode_id)]
    if not hits:
        raise LookupError(f"{episode_id} 가 후보에 없다 — 선행 CP 를 확인하라")
    pick = hits[0]
    base = (pf.PROJECTS / pick["project_id"] / "checkpoints" / "episodes"
            / pick["episode_id"])
    S = pf._load(base / "scene_save" / "manifest.json") or {}
    V = pf._load(base / "shot_validator" / "manifest.json") or {}
    W = pf._load(base / "visual_world_rules" / "manifest.json") or {}
    pick = {**pick, "era": str(W.get("era") or ""),
            "region": str(W.get("region") or "")}
    Wcp = pf._load_cp(base / "visual_world_rules" / "manifest.json")
    world = build_grounding_world_facts(Wcp)
    segs = {f"scene-{x['scene_index']}": (x.get("text") or "")
            for x in S.get("segments") or []}
    plan, cats = [], {}
    for n, b in enumerate(pick["bundles"]):
        ids = [f"scene-{i}" for i in b]
        cat = sc.build_catalog(V.get("scenes") or [], ids)
        cats[f"c{n}"] = cat
        plan.append({"chunk_id": f"c{n}", "segment_ids": ids,
                     "payload": gc.build_chunk_payload(ids, segs, world,
                                                       shot_catalog=cat)})
    return pick, plan, segs, {"world": world, "catalogs": cats}


def cover_sample(targets: List[Dict[str, Any]], per_owner: int
                 ) -> List[Dict[str, Any]]:
    """갈래를 **덮는** 표본. ★결정적 — 같은 입력이면 같은 표본.

    ★★실제 원고에서 대상이 **40개**로 나왔다 (2026-08-31). 한 판에 다 사면
    논리 165회다. canary 가 보려는 것은 「기구가 실제 내용에서 도나」와
    「참조가 맞나」이므로, **갈래마다 몇 개**면 그것을 본다.

    ★이것은 **표본이라고 적는다** — 전수인 척하면 「40개를 다 봤다」가 된다.
    ★갈래로 **거르는 것이 아니다.** 있는 갈래를 다 덮고, 갈래마다 앞에서부터
     `per_owner` 개를 집는다(신원 차례라 결정적).
    """
    by: Dict[str, List[Dict[str, Any]]] = {}
    for t in targets:
        by.setdefault(str(t["owner_type"]), []).append(t)
    out: List[Dict[str, Any]] = []
    for owner in sorted(by):
        out += by[owner][:max(1, per_owner)]
    return sorted(out, key=lambda x: str(x["local_id"]))


def approved_logical(n_targets: int) -> int:
    return CHUNK_LOGICAL + n_targets * PER_TARGET_LOGICAL


def _sdk_max_retries() -> Any:
    """OpenAI SDK 자체 재시도. ★**세는 자리 아래**라 예산이 못 본다.

    `reserve_current_research_call` 은 `FailoverOpenAIClient._invoke` 에
    있는데, SDK 는 그 **안쪽**에서 HTTP 를 다시 던진다. 대개 429·5xx 라
    과금은 안 되지만 「HTTP 요청 수」와 「센 수」가 다르다 — 감사에서 두 수가
    갈리면 이것부터 본다.
    """
    try:
        from app.core.openai_keys import openai_client

        # ★**잠근 인자로** 만든 것을 본다 — 기본값을 보면 실제와 다른 수가 난다
        c = openai_client(**SDK_LOCK)
        inner = c._raw()[1] if hasattr(c, "_raw") else c
        return getattr(inner, "max_retries", None)
    except Exception:                       # noqa: BLE001
        return None


class ApprovedPlanMismatch(RuntimeError):
    """계획이 **승인된 것과 다르다**. ★장부·trace·provider **앞에서** 선다."""


class _RoundClock:
    """이 대상이 **지금 몇 번째 라운드**인가. ★`acquire_one` 의 차례를 읽는다.

    `acquire_one` 은 라운드마다 **저작 → 검색 → 선택** 차례로 부른다. 저작이
    라운드의 시작이므로 거기서 하나 올린다. 사건 장부가 어느 호출이 어느
    라운드인지 알아야 Opik·provider 와 양방향으로 맞댈 수 있다.
    """

    def __init__(self) -> None:
        self.n = 0

    def begin(self) -> int:
        self.n += 1
        return self.n

    @property
    def now(self) -> int:
        return max(1, self.n)


def instrumented(t: Dict[str, Any], *, ident: str, events, ctx: Dict[str, Any],
                 trace_id: Optional[str] = None):
    """이 대상의 **네 callable** 을 만든다 — 나가기 직전에 사건을 적는다.

    ★★★적는 자리가 **판정 앞**이다 (Codex BLOCK 2026-08-31). 참조 장부는
    대상마다 최종 한 줄뿐이라, 한 대상이 최대 6번 나가는 것을 못 되짚는다.
    여기서 단계·라운드마다 **실제 나간 것**을 남긴다.

    Returns:
        `(write_brief, search, judge, clock)`.
    """
    from app.modules.pipeline.search_grounded_ref import download_candidate
    from tools.grounding_audit import ref_events as ev

    clock = _RoundClock()

    def _write(_t, narrow):
        n = clock.begin()
        o = brief_outbound(_t, narrow=narrow, **ctx)
        events.put(target_identity=ident, stage=ev.STAGE_WRITE, round_no=n,
                   trace_id=trace_id,
                   outbound={"model": o["model"], "narrow": bool(narrow),
                             "request_lock": o["request_lock"],
                             "system_sha": _sha(o["system"]),
                             "user_sha": _sha(o["user"]),
                             "user_chars": len(o["user"])})
        return _write_brief(_t, world_facts=ctx["world_facts"],
                            source_text=ctx["source_text"], narrow=narrow,
                            era_declaration=ctx["era_declaration"],
                            region_declaration=ctx["region_declaration"])

    def _search(**kw):
        events.put(target_identity=ident, stage=ev.STAGE_SEARCH,
                   round_no=clock.now, trace_id=trace_id,
                   # ★**실제 나가는 질의 재료**를 그대로 적는다
                   outbound={"directive_native": kw.get("directive_native"),
                             "terms_native": kw.get("terms_native"),
                             "language_lock_native":
                                 kw.get("language_lock_native"),
                             "max_results": kw.get("max_results")})
        return _live_search(**kw)

    def _judge(paths):
        kind = kind_name_of(t)
        events.put(target_identity=ident, stage=ev.STAGE_PICK,
                   round_no=clock.now, trace_id=trace_id,
                   outbound={"kind_name": kind,
                             "photographs": [p.get("path") for p in paths],
                             "request_lock": dict(REQUEST_LOCK)})
        return _live_judge(paths, kind_name=kind)

    return _write, _search, _judge, download_candidate, clock


def _sha(text: str) -> str:
    import hashlib

    return hashlib.sha256(str(text or "").encode("utf-8")).hexdigest()[:16]

def assert_plan_matches_approval(sample: List[Dict[str, Any]],
                                 preflight: Optional[Dict[str, Any]] = None
                                 ) -> None:
    """유료로 사기 **직전**의 문. ★아무것도 안 하고 선다.

    ★★★앞 판은 「참조 획득을 안 지나는 갈래」를 **찍기만 하고 그냥 갔다**
    (Codex 2026-08-31). 이 canary 의 목적이 **다섯 갈래 각 1개**인데, 갈래가
    빠진 채 사면 그 판은 목적을 못 이룬 채 돈만 쓴다.

    ★`preflight` 를 주면 **사기 전에 적어 둔 것과 같은지**도 본다. 결과를
    보고 표본을 넓히지 않기 위해서다.
    """
    from app.core.openai_keys import slot_count
    from app.modules.pipeline.grounding_entity_contract import owners

    want = set(owners())
    got = {str(t.get("owner_type") or "") for t in sample}
    if got != want:
        raise ApprovedPlanMismatch(
            f"표본 갈래가 {sorted(got)} 인데 정본은 {sorted(want)} 이다 — "
            "다섯 갈래 각 1개가 이 판의 목적이다. 안 사고 선다")
    if len(sample) != len(want):
        raise ApprovedPlanMismatch(
            f"표본이 {len(sample)}개인데 갈래는 {len(want)}개다 — "
            "갈래마다 하나여야 한다. 안 사고 선다")
    n = int(slot_count())
    if n != APPROVED_SLOT_COUNT:
        raise ApprovedPlanMismatch(
            f"키 슬롯이 {n} 인데 승인은 {APPROVED_SLOT_COUNT} 이다 — "
            "물리 상한 계산이 승인 장부와 갈린다. 안 사고 선다")
    if not preflight:
        return
    want_ids = [str(r.get("subject_id")) for r in (preflight.get("sample")
                                                   or [])]
    got_ids = [str(t.get("subject_id")) for t in sample]
    if want_ids and want_ids != got_ids:
        raise ApprovedPlanMismatch(
            f"표본이 preflight 와 다르다: 적어 둔 것 {want_ids} · 지금 "
            f"{got_ids} — 결과를 보고 넓히지 않는다. 안 사고 선다")

#: 대상 하나가 내는 논리 호출의 **갈래별 수**. ★`PER_TARGET_LOGICAL` 의 속.
#:  라운드마다 저작(structured) + 검색(OpenAI 직접) + 선택(structured).
LOGICAL_PER_TARGET_STRUCTURED = 4      # 2라운드 × (저작 1 + 선택 1)
LOGICAL_PER_TARGET_SEARCH = 2          # 2라운드 × 검색 1

#: ★litellm 이 **제 OpenAI client** 에 주는 재시도. `OpenAIChatCompletion.
#:  completion()` 이 `inference_params.pop("max_retries", 2)` 로 집는다.
#:  `call_structured` 공개 인자에는 `num_retries` 만 있어 이 값을 못 바꾼다 —
#:  그리고 `_completion` 의 reserve 는 `router.completion` **바깥**이라
#:  이 겹을 **못 센다** (Codex 2026-08-31).
#: ★값은 **공용 계약 한 곳**에서 온다 — 같은 수를 두 곳에 적으면
#:  한쪽만 고쳐진다 (Codex 2026-08-31).
from tools.grounding_audit.call_bound_contract import (  # noqa: E402
    CALL_STRUCTURED_TIERS as _SHARED_TIERS,
    LITELLM_SDK_MAX_RETRIES)


def physical_upper_bound(logical: int, *, targets: int = 0) -> Dict[str, Any]:
    """상한 둘을 **갈라서** 낸다. ★식은 **공용 계약**에 있다.

    ★★★**세는 수와 나가는 HTTP 수는 다르다** (Codex 2026-08-31) —

        counted_attempt_cap   `reserve` 가 세는 provider 시도. 상한을 **거는
                              수**다(넘으면 네트워크 전에 선다).
        raw_http_upper_bound  실제 HTTP 요청. structured 경로는 litellm 이
                              제 client 에 `max_retries` 를 주는데 그것이
                              `reserve` **아래**라 안 세어진다.

    ★검색 경로는 `openai_client(max_retries=0)` 으로 잠가서 둘이 같다.

    ★★식을 여기 다시 적지 않는다 — `call_bound_contract.layers`/`bounds` 를
    부른다. 같은 산식을 두 벌로 두면 한쪽만 고쳐진다 (Codex NON-BLOCK).

    Args:
        logical: 이 범위에서 나갈 수 있는 논리 호출 수.
        targets: 대상 수. 주면 raw HTTP 상한을 갈래별로 정확히 낸다.
    """
    from app.core.config import settings
    from app.core.openai_keys import slot_count

    from tools.grounding_audit.call_bound_contract import bounds, layers

    slots = max(1, int(slot_count()))
    locked = REQUEST_LOCK.get("num_retries")
    contract = {
        "num_retries": (settings.llm_max_retries if locked is None
                        else locked),
        "enable_fallback": REQUEST_LOCK.get("enable_fallback", True),
    }
    lay = layers(contract=contract, slots=slots)
    b = bounds(logical=logical, contract=contract, slots=slots)
    counted = b["counted"]
    n = int(targets or 0)
    if n:
        # ★갈래가 둘이면 **갈래별로** 센다 — 검색은 SDK 를 0으로 잠갔다
        search = bounds(logical=n * LOGICAL_PER_TARGET_SEARCH,
                        contract={"num_retries": 0, "enable_fallback": False},
                        slots=slots,
                        sdk_max_retries=int(SDK_LOCK.get("max_retries", 0)))
        structured = bounds(logical=n * LOGICAL_PER_TARGET_STRUCTURED,
                            contract=contract, slots=slots)
        raw = structured["raw_http"] + search["raw_http"]
    else:
        raw = b["raw_http"]
    return {
        "logical": int(logical),
        "armed_cap": int(logical) * APPROVED_PHYSICAL_FACTOR,
        "counted_attempt_cap": counted,
        "raw_http_upper_bound": raw,
        "layers": {"tiers": lay["tiers"], "key_slots": lay["key_slots"],
                   "router_tries": lay["router_tries"],
                   "per_structured_call": lay["per_logical_counted"],
                   "search_sdk_max_retries":
                       int(SDK_LOCK.get("max_retries", 0)),
                   "litellm_sdk_max_retries": LITELLM_SDK_MAX_RETRIES,
                   "counted_below_this": lay["★counted_below_this"]},
        "request_lock": dict(REQUEST_LOCK),
        "sdk_lock": dict(SDK_LOCK),
    }

def search_pick_contract(*, rounds: int) -> Dict[str, Any]:
    """검색·선택·라운드가 **무엇으로 도는가**. ★**이미 있는 것을 부른다**.

    ★★★내가 새 산식을 쓰려다 Codex 에 잡혔다 (2026-08-31). production 에
    `reference_acquisition.acquisition_contract_sha` 가 **이미** 넷을 한 벌로
    접고 있다 — 거친 선택 팩(지문·스키마·좁힘) · 검색 계약 sha · 라운드 수 ·
    라운드당 상한. 두 벌을 만들면 한쪽만 고쳐진다(그 함수 docstring 이
    바로 그 사고를 적고 있다).

    ★특히 `_live_judge` 는 `ctp.load_pack()` 을 실제로 쓴다. picker 팩이 한
    바이트만 바뀌어도 다른 선택이 나오는데, 앞 판 신원은 그것을 안 접어서
    **옛 chosen 을 hit** 했다.
    """
    from app.modules.pipeline import reference_acquisition as ra
    from app.modules.pipeline import reference_acquisition_rounds as rar

    return {"acquisition_contract_sha": ra.acquisition_contract_sha(
                rounds=int(rounds)),
            "rounds_module_contract": rar.ROUNDS_CONTRACT_VERSION}

def acquisition_identity(t: Dict[str, Any], *, pack: str,
                         models: Dict[str, str],
                         context: Dict[str, Any],
                         rounds: int = 2) -> str:
    """이 대상의 참조를 **무엇으로 사는가**. ★같으면 다시 안 산다.

    ★★참조 획득에 장부가 **없어서** 실제로 사고를 냈다 (2026-08-31): 일부만
    다시 돌리며 기록 파일을 통째로 다시 써서 **앞서 산 10개 기록이 사라졌다.**

    ★★★**필드를 손으로 나열하지 않는다** (Codex BLOCK 2026-08-31). 앞 판은
    `subject_id`·옛 directive·옛 terms 만 접어서, **같은 이름이 다른 시대·
    지역에 나와도** 옛 결과를 되썼다. 이제 `brief_outbound` 가 만드는
    **나가는 payload 그대로**를 라운드마다 접는다 — 저작 입력이 늘어도
    신원이 저절로 따라 움직인다.

    ★**후처리 계약은 안 섞는다** — 파서를 고친 것만으로 다시 사면 안 된다.

    Args:
        context: `brief_outbound` 에 그대로 넘길 것 (`world_facts`·
            `source_text`·`era_declaration`·`region_declaration`).
            ★여기 있는 것이 곧 **나가는 것**이다.
        rounds: 이 계약이 살 수 있는 라운드 수. 라운드마다 나가는 것이
            다르므로(좁힘 문안) **전부** 접는다.
    """
    import hashlib

    h = hashlib.sha256()
    parts = [str(t.get("subject_id") or "")]
    for n in range(max(1, int(rounds))):
        o = brief_outbound(t, narrow=(n > 0), **context)
        parts += [str(o["system"]), str(o["user"]), str(o["model"]),
                  json.dumps(o["schema"], sort_keys=True,
                             ensure_ascii=False),
                  json.dumps(o.get("request_lock") or {}, sort_keys=True)]
    parts += [str(pack), json.dumps(models, sort_keys=True),
              json.dumps({"rounds": int(rounds)}, sort_keys=True),
              # ★검색·선택·라운드 계약도 「무엇으로 사는가」다 (Codex BLOCK)
              json.dumps(search_pick_contract(rounds=int(rounds)),
                         sort_keys=True, ensure_ascii=False)]
    for part in parts:
        h.update(part.encode("utf-8"))
        h.update(b"\x00")
    return h.hexdigest()[:20]


#: 저작기 모델. ★신원과 전송이 **같은 값**을 본다.
BRIEF_MODEL = "gpt"


def brief_outbound(t, *, world_facts: str, source_text: str, narrow: bool,
                   era_declaration: str = "",
                   region_declaration: str = "") -> Dict[str, Any]:
    """저작기로 **실제 나가는 것**. ★전송과 신원이 **한 자리**에서 나온다.

    ★★★왜 한 자리인가 (Codex BLOCK 2026-08-31). 앞 판 신원은 필드를 **손으로
    나열**했다 — `subject_id` · 옛 directive · 옛 terms · pack · models ·
    저작 계약. 그런데 저작기로 실제 나가는 것은 그보다 넓다: owner ·
    `coarse_type_label` · **`visual_brief`** · world facts · 시대 · 지역 ·
    원문 언어 표본. 그래서 **같은 이름이 다른 시대·지역에 나와도** 옛 검색·
    선택 결과를 되썼다. 다른 검색인데 같은 것으로 본 것이다.

    이제 **나가는 payload 자체가 신원**이다 —
    `grounding_chunk.acquisition_identity` 와 같은 원칙. 필드를 다시 나열하지
    않으므로 저작 입력이 늘어도 신원이 **저절로** 따라 움직인다.

    Returns:
        `{"system", "user", "schema", "model"}` — provider 로 나가는 그대로.
    """
    from app.modules.pipeline import grounding_ref_brief as grb

    desc = "\n".join(x for x in (str(t.get("surface_form") or ""),
                                  str(t.get("visual_brief") or "")) if x)
    user = grb.build_brief_user(
        owner_type=str(t.get("owner_type") or ""),
        coarse_type_label=kind_name_of(t),
        subject_description=desc, world_facts_block=world_facts,
        era_declaration=era_declaration,
        region_declaration=region_declaration,
        source_text=source_text, researched_facts_block="")
    if narrow:
        # ★좁힘 문안은 **여기**다 — 한국어 지시문 뒤가 아니다
        user += "\n\n" + grb.load_narrow_retry_hint()
    return {"system": grb.load_brief_system(), "user": user,
            "schema": grb.build_brief_schema(), "model": BRIEF_MODEL,
            # ★계약이 바뀌면 **다른 방식으로 사는 것**이다
            "request_lock": dict(REQUEST_LOCK)}


def _write_brief(t: Dict[str, Any], *, world_facts: str, source_text: str,
                 narrow: bool, era_declaration: str = "",
                 region_declaration: str = "") -> Dict[str, Any]:
    """검색 지시문을 **원어로 저작한다**. ★**다섯 갈래 공용** 팩을 쓴다.

    ★보내는 것은 `brief_outbound` 가 만든다 — 신원이 보는 것과 **같은 것**을
    보낸다. 두 벌로 만들면 한쪽만 고쳐진다.
    """
    from app.modules.llm.llm_client import call_structured

    o = brief_outbound(t, world_facts=world_facts, source_text=source_text,
                       narrow=narrow, era_declaration=era_declaration,
                       region_declaration=region_declaration)
    tag = "ref_canary_brief"
    return call_structured(
        tag, o["system"], o["user"], o["schema"],
        project_config={tag: {"model": o["model"]}}, schema_name=tag,
        opik_metadata={"tags": [REF_TAG]}, **o["request_lock"])


def _live_search(**kw) -> Dict[str, Any]:
    """실제 검색. ★`search_grounded_ref` 를 **그대로** 부른다."""
    # ★조립부에서 **그대로** 가져온다 — 이름을 짐작하지 않는다
    from app.core.openai_keys import openai_client
    from app.modules.pipeline.search_grounded_ref import (
        SEARCH_ORCHESTRATOR, search_reference_images)

    # ★★★SDK 자체 재시도를 **0** 으로 잠근다 (Codex 2026-08-31).
    #  `max_retries` 는 `reserve_current_research_call` **아래**라 예산이
    #  못 본다 — 잠그지 않으면 「물리 60」이 **거짓**이 된다(HTTP 요청은
    #  그보다 많을 수 있다). Opik·provider·장부를 맞대려면 세는 수와 나가는
    #  수가 같아야 한다.
    return search_reference_images(
        openai_client(**SDK_LOCK), model=SEARCH_ORCHESTRATOR, **kw)


def kind_name_of(t: Dict[str, Any]) -> str:
    """심판에게 줄 **찾던 부류의 이름**. ★추출이 낸 `coarse_type_label` 뿐.

    ## ★★왜 `visual_brief` 를 빼는가 (사용자 확정 2026-08-31)

    「니 말이 맞아. 너무 협소하게 판단해 VLM이 판단할 수 없어. 절대로 사람도
    잘 못하는데.」

    앞 판은 `surface_form` + `visual_brief` 를 줬다. 그러면 심판이
    「발목까지 내려오고 앞을 끈으로 여미는 긴 겉옷인가」를 보게 되는데,
    그것은 **시대·형태 세부 판정**이다 — 사람도 사진 한 장으로 잘 못한다.
    실제로 모자 후보가 **세부 명칭 불일치**로 거절됐다.

    그래서 심판에게는 **뜻이 갈리지 않을 만큼의 부류 이름**만 준다 —
    「신발의 앞부분」·「사람이 쓰는 모자」 정도. 연도·나라·제조·모델·재질·
    생김새·옛 명칭·좋고 나쁨은 **주지 않는다**.

    ★시대·지역·겉모습은 **검색 쪽에 그대로 쓴다** — 무엇을 찾을지는 그것들이
    정하고, 심판은 「찾던 부류가 맞나」만 본다.

    ★새 호출을 안 만든다. 같은 추출 응답이 낸 칸이다.
    """
    label = str(t.get("coarse_type_label") or "").strip()
    if label:
        return label
    # ★없으면 **이름만** — 겉모습을 섞지 않는다. 옛 팩으로 산 행이 여기 온다.
    return str(t.get("surface_form") or "").strip()


def _live_judge(paths: List[Dict[str, Any]], *, kind_name: str
                ) -> Dict[str, Any]:
    """거친 종류 선택 — **1심**. ★기존 `era_research` 와 같은 모델·같은 팩.

    ★심판을 늘리지 않는다 (Codex). coarse 계약은 1심에서도 그대로 돈다 —
    「살아남은 **모든** 심판이 yes+visible」이 심판 하나면 그 하나가 정한다.
    """
    from app.modules.llm.llm_client import call_structured
    from app.modules.pipeline import coarse_type_pick as ctp
    from app.modules.pipeline.era_research import PICK_MODEL
    from app.modules.pipeline.multiroll_gemini import png_part

    pack = ctp.load_pack()
    sys_text = pack["stems"][ctp.SYSTEM_STEM]["content"].strip()
    parts: List[Dict[str, Any]] = [{
        "type": "text",
        "text": "THE KIND OF THING the photograph must show:\n" + kind_name,
    }]
    for g in paths:
        parts.append({"type": "text", "text": f"PHOTOGRAPH {g['index']}:"})
        parts.append(png_part(Path(SERVE_ROOT) / g["path"]))
    tag = "ref_canary_pick"
    raw = call_structured(
        tag, sys_text, parts, pack["stems"][ctp.SCHEMA_STEM]["content"],
        project_config={tag: {"model": PICK_MODEL}}, schema_name=tag,
        opik_metadata={"tags": [REF_TAG]}, **REQUEST_LOCK)
    return {PICK_MODEL: raw}


def main() -> int:
    if len(sys.argv) < 3 or sys.argv[1] not in ("--dry", "--live"):
        print(__doc__)
        return 2
    mode, jp = sys.argv[1], Path(sys.argv[2])
    #: ★일부만 다시 돌린다. 참조 획득은 아직 장부가 없어서(다음 과제),
    #:  전부 다시 돌리면 이미 산 것을 다시 산다.
    args = {a.split("=", 1)[0]: a.split("=", 1)[1]
            for a in sys.argv[3:] if "=" in a}
    only = {x for x in (args.get("only") or "").split(",") if x}
    episode = (args.get("episode") or "").strip()
    #: 갈래마다 몇 개를 살까. 0 이면 **전수**.
    per_owner = int(args.get("sample") or 0)
    which = (args.get("fixture") or "").strip()
    #: ★사기 **전에** 적어 둔 계획. 주면 표본이 그것과 같은지 본다 —
    #:  결과를 보고 넓히지 않기 위해서다.
    #: ★★★**판독을 얼어붙은 장부에서 되살린다** (Codex BLOCK 2026-08-31).
    #:
    #:  앞 판은 `--live` 에서도 `rr.run` 을 써서, 장부에 **merge 응답이 없으면
    #:  그 자리에서 새로 샀다.** 실측으로 merge 신원이 장부에 없었다 —
    #:  즉 `new_dispatch=0` 이라고 적어 둔 것이 거짓이었다. 게다가 계획 문
    #:  (`assert_plan_matches_approval`)은 `rr.run` **뒤**라 그 구매를 못 막는다.
    #:
    #:  `rr.replay` 는 **보낼 길이 아예 없다**(`send` 인자가 없다). 그래서
    #:  판독 단계의 신규 전송이 **믿음이 아니라 구조**로 0 이 된다.
    _replay_from = (args.get("chunk_replay") or "").strip()
    _pf_path = (args.get("preflight") or "").strip()
    _preflight = (json.loads(Path(_pf_path).read_text(encoding="utf-8"))
                  if _pf_path else None)
    if mode == "--dry":
        seal_outbound()

    # ★★★**판독 단계에서 모르는 사이에 사는 길을 막는다** (Codex BLOCK).
    #  이 canary 의 승인 장부는 「판독 재사용 · 참조만 신규」다. 그런데
    #  `rr.run` 은 장부에 없는 응답을 **그 자리에서 산다** — 실측으로
    #  merge 신원이 얼어붙은 장부에 없었다. 계획 문은 `rr.run` **뒤**라
    #  그 구매를 못 막는다. 그래서 **앞에서** 선다.
    if mode == "--live" and not _replay_from \
            and not args.get("allow_chunk_dispatch"):
        raise ApprovedPlanMismatch(
            "`--live` 인데 `chunk_replay=` 가 없다 — 판독 단계가 장부에 없는 "
            "응답을 그 자리에서 살 수 있고, 그러면 승인 장부의 「판독 신규 0」이 "
            "거짓이 된다. 되살릴 장부를 주거나, 판독 구매를 정말 승인한다면 "
            "`allow_chunk_dispatch=1` 을 명시하라. 아무것도 안 하고 선다")
    send = (rr.live_send if mode == "--live" else rr._dry_send)
    if episode:
        # ★실제 원고 — 계획·원문·샷 목록은 **preflight 것을 그대로** 쓴다
        pick, plan, segs, extra = real_plan(episode)
        cap_chunks = len(plan) + 1
        print(f"■ 실제 원고 {pick['project_id'][:8]}/{pick['episode_id'][:8]} · "
              f"시대 {pick.get('era','')[:24]} / 지역 {pick.get('region','')[:20]}")
        got = rr.run(jp, send, world_facts=extra["world"], cap=cap_chunks,
                     dispatch_budget=cap_chunks,
                     approved_slots=(rr.APPROVED_SLOTS if mode == "--live"
                                     else None),
                     plan=plan, segments=segs,
                     shot_catalogs=extra["catalogs"],
                     lock_extra={"project_id": pick["project_id"],
                                 "episode_id": pick["episode_id"],
                                 "chunks": len(plan)})
    else:
        if which == "period":
            from tests.grounding.fixtures import period_episode as ep
        else:
            from tests.grounding.fixtures import synthetic_episode as ep
        ep.assert_planted()
        segs = ep.segment_texts()
        cats = {}
        plan = []
        from app.modules.pipeline import grounding_chunk as gc
        from app.modules.pipeline import grounding_shot_catalog as sc
        for n, b in enumerate(ep.bundles()):
            ids = [f"scene-{i}" for i in b]
            cat = sc.build_catalog(ep.shot_scenes(), ids)
            cats[f"c{n}"] = cat
            plan.append({"chunk_id": f"c{n}", "segment_ids": ids,
                         "payload": gc.build_chunk_payload(
                             ids, segs, ep.WORLD_FACTS, shot_catalog=cat)})
        cap_chunks = len(plan) + 1
        print(f"■ fixture={which or 'synthetic'} · 구간 {len(plan)} · "
              f"샷 {sum(len(c) for c in cats.values())}")
        if _replay_from:
            # ★보낼 길이 **없는** 경로다 — 판독 신규 전송이 구조적으로 0
            got = rr.replay(Path(_replay_from), world_facts=ep.WORLD_FACTS,
                            plan=plan, segments=segs, shot_catalogs=cats)
            print(f"  ★판독은 **되살렸다** ({Path(_replay_from).name}) — "
                  f"보낼 길이 없어 신규 전송 **구조적으로 0**")
        else:
            got = rr.run(jp, send, world_facts=ep.WORLD_FACTS, cap=cap_chunks,
                         dispatch_budget=cap_chunks,
                         approved_slots=(rr.APPROVED_SLOTS if mode == "--live"
                                         else None),
                         plan=plan, segments=segs, shot_catalogs=cats,
                         lock_extra={"fixture": which or "synthetic",
                                     "chunks": len(plan)})
    reduced = got["reduced"]
    targets = targets_from(reduced)

    owners = sorted({str(t["owner_type"]) for t in targets})
    print(f"■ {mode[2:]} — 구간 판독 논리 {got['logical']} "
          f"(산 것 {got['bought']} · 재사용 {got['reused']})")
    print(f"  행 {len(reduced['rows'])} · 격리 {len(got['quarantined'])}")
    skipped = parts_skipped(reduced)
    print(f"  참조 획득 대상 {len(targets)}개 · 갈래 {owners}")
    if skipped:
        print(f"  ★통째가 이미 대상이라 **안 사는 부분** {len(skipped)}개:")
        for x in skipped:
            print(f"     {str(x['part_form'])[:22]:24} ⊂ {x['whole_form']}")

    from app.modules.pipeline.grounding_entity_contract import owners as _own
    want = sorted(_own())
    missing = sorted(set(want) - set(owners))
    if missing:
        print(f"  ★참조 획득을 안 지나는 갈래: {missing} — "
              "다섯 갈래 끝점이 성립하지 않는다")

    sampled = False
    if per_owner:
        full = len(targets)
        targets = cover_sample(targets, per_owner)
        sampled = True
        print(f"  ★**표본**이다 — 갈래마다 {per_owner}개씩 {len(targets)}개 "
              f"(전체 {full}개 중). 「다 봤다」가 아니다")
    # ★★상한은 **실제로 살 것**에 건다 — 표본을 뽑은 뒤다.
    #  앞 판은 표본 뽑기 **앞**에 있어서, 8개만 사려는데도 전체 40으로 막혔다.
    if len(targets) > MAX_ACQUISITION_TARGETS:
        raise ApprovedScopeMismatch(
            f"실제로 살 대상이 {len(targets)}개인데 승인 상한은 "
            f"{MAX_ACQUISITION_TARGETS} 이다 — 사람이 다시 정해야 한다. "
            "참조는 한 건도 안 사고 선다"
            + ("" if per_owner else
               " (`sample=N` 으로 갈래마다 N개만 살 수 있다)"))
    # ★★★**사기 직전의 문** — 장부도 안 열고 provider 도 안 부른다.
    #  앞 판은 「참조 획득을 안 지나는 갈래」를 **찍기만 하고 그냥 갔다**.
    #  이 판의 목적이 다섯 갈래 각 1개인데 갈래가 빠지면 돈만 쓴다.
    if mode == "--live":
        assert_plan_matches_approval(targets, _preflight)
    cap = approved_logical(len(targets))
    print(f"  ★승인 논리 상한 {cap} = 판독 {CHUNK_LOGICAL} + "
          f"대상 {len(targets)} × {PER_TARGET_LOGICAL}")

    # ── ②참조 획득 ────────────────────────────────────────────────────
    from app.core.research_call_budget import (research_calls_armed,
                                               research_run_scope)
    from app.modules.llm.opik_trace import open_trace
    from app.modules.pipeline import reference_acquisition_rounds as rar
    from app.modules.pipeline.grounding_chunk_journal import ChunkJournal
    from app.modules.pipeline.era_research import PICK_MODEL
    from app.modules.pipeline.search_grounded_ref import (
        REF_PACK_VERSION, SEARCH_ORCHESTRATOR, download_candidate)

    refdir = jp.parent / "refs"
    # ★저작기에 줄 것 — 세계 사실과 **언어 판정용** 원문
    world_block = extra["world"] if episode else ep.WORLD_FACTS
    # ★★시대는 **제 칸**에서 온다 — 세계 사실 전문에서 짐작하지 않는다
    #  (Codex BLOCK 2026-08-31: 값·번지·수량의 숫자를 시대로 읽고 있었다).
    era_declaration = str(pick.get("era") or "") if episode else ep.ERA
    # ★★지역도 **제 칸**에서. 2차 검색에서 지역을 버리면 아무도 안 본다 —
    #  검색은 질의·이미지만 돌려주고 심판은 부류·보임만 본다 (Codex BLOCK).
    region_declaration = (str(pick.get("region") or "") if episode
                          else ep.REGION)
    source_text = ("\n\n".join(segs.values()) if episode
                   else "\n\n".join(ep.segment_texts().values()))
    # ★★대상마다 **durable 장부**. 같은 신원이면 다시 안 산다.
    models = {"search": SEARCH_ORCHESTRATOR, "pick": PICK_MODEL}
    # ★저작 계약 — 장부 **계약 지문**(epoch)용. 신원은 아래 `brief_ctx` 가
    #  나르는 **나가는 payload 그대로**가 정한다.
    from app.modules.pipeline import grounding_ref_brief as _grb
    brief_contract = {**_grb.brief_contract(), "model": BRIEF_MODEL,
                      # ★심판에게 주는 것 — 부류 이름 **하나**
                      "kind": "coarse_type_label"}
    # ★★신원이 보는 것 = **저작기로 나가는 것**. 필드를 손으로 나열하지
    #  않는다 — 이 안의 무엇이 바뀌든 신원이 저절로 움직인다.
    brief_ctx = {"world_facts": world_block, "source_text": source_text,
                 "era_declaration": era_declaration,
                 "region_declaration": region_declaration}
    ref_jr = ChunkJournal(jp.parent / REF_JOURNAL_NAME,
                          contract={"pack": REF_PACK_VERSION,
                                    "models": models,
                                    "brief": brief_contract})
    ref_jr.assert_no_uncertain()
    # ★★단계·라운드마다 **나가기 직전에** 적는다. 참조 장부는 대상마다 최종
    #  한 줄뿐이라 한 대상의 최대 6번 전송을 못 되짚는다 (Codex BLOCK).
    from tools.grounding_audit import ref_events as _ev
    _run_nonce = uuid.uuid4().hex[:16]
    events = _ev.RefEvents(jp.parent / REF_EVENTS_NAME, run_nonce=_run_nonce)
    print(f"  ★사건 장부 {events.path.name} · run_nonce={_run_nonce}")
    records: List[Dict[str, Any]] = []
    # ★★★**나가는 자리에서 센다** (2026-08-31 발견 — 여기가 비어 있었다).
    #
    #  `reserve_current_research_call` 은 provider 경계 둘(`llm_client.
    #  _completion` 의 키 슬롯 loop · `openai_keys._invoke` 의 failover loop)
    #  에 이미 박혀 있는데 `research_calls_armed()` **밖에서는 아무 일도 안
    #  한다**. 이 canary 는 한 번도 팔을 안 들어서 참조 획득 구간에 **물리
    #  상한이 없었다** — 내가 센 논리 수만 있었다.
    #
    #  ★상한에 닿으면 `ResearchCallBudgetExceeded` 가 **네트워크에 닿기 전에**
    #   오른다. 아래 `except BaseException` 이 「샀는지 모른다」로 적으므로
    #   기록도 안 잃는다.
    # ★★★상한은 **이 범위 안에서 나갈 수 있는 것**에만 건다 (Codex
    #  2026-08-31). 구간 판독은 이 `with` **밖**에서 이미 끝났는데 그 몫
    #  (`CHUNK_LOGICAL`)까지 세어 66 을 열고 있었다 — 보고한 승인 수 60 과
    #  어긋난다. 상한이 실제보다 넓으면 그것은 상한이 아니다.
    _new_logical = len(targets) * PER_TARGET_LOGICAL
    _pb = physical_upper_bound(_new_logical, targets=len(targets))
    print(f"  ★물리 상한 **{_pb['armed_cap']}** 을 나가는 자리에 건다 "
          f"(이 범위의 신규 논리 {_pb['logical']} × "
          f"{APPROVED_PHYSICAL_FACTOR}; 판독 {CHUNK_LOGICAL} 은 밖이다). "
          f"세는 시도 {_pb['counted_attempt_cap']} · ★실제 HTTP 는 "
          f"최대 {_pb['raw_http_upper_bound']} — litellm 이 제 client 에 주는 "
          f"재시도 {_pb['layers']['litellm_sdk_max_retries']} 는 세는 자리 "
          f"**아래**다 (검색 경로는 0 으로 잠갔다)")
    with research_run_scope(cap=_pb["armed_cap"]) as _budget, \
            research_calls_armed():
        for t in targets:
            if only and str(t["subject_id"]) not in only:
                continue
            if mode == "--dry":
                records.append({"subject_id": t["subject_id"], "status": "dry",
                                "chosen": None, "rounds": [], "candidates": [],
                                "target": t})
                continue
            ident = acquisition_identity(t, pack=REF_PACK_VERSION,
                                         models=models, context=brief_ctx)
            hit = ref_jr.get(ident)
            if hit is not None:
                # ★★**이미 산 것은 다시 안 산다.** 신원이 같으면 같은 검색이다.
                hit["target"] = t
                hit["reused"] = True
                records.append(hit)
                print(f"   {t['owner_type']:14} {str(t['surface_form'])[:16]:18} "
                      f"→ {hit['status']} (되씀)")
                continue
            try:
                # ★★★**부모 trace 가 안 열리면 안 산다** (Codex BLOCK).
                #  `open_trace` 는 꺼져 있으면 예외 없이 None 을 준다 —
                #  production 은 그게 맞지만 **이 판은 기록이 목적**이다.
                #  결속 키 없이 사면 나중에 무엇을 샀는지 못 되짚는다.
                #  ★신원은 **부모 trace metadata** 로 간다 — 고카디널리티
                #   값을 tag 에 넣으면 Opik whitelist 가 터진다.
                with open_trace(
                        name=REF_TRACE_NAME, tags=[REF_TAG],
                        metadata={rr.ID_META_KEY: ident,
                                  rr.RUN_META_KEY: _run_nonce,
                                  "owner_type": t["owner_type"],
                                  "subject_id": t["subject_id"]},
                        thread_id=REF_THREAD,
                        input_data={"identity": ident}) as _tr:
                    if _tr is None:
                        raise rr.TraceUnavailable(
                            "부모 Opik trace 를 못 열었다 — 결속 키 없이 "
                            "사면 나중에 무엇을 샀는지 못 되짚는다. "
                            "provider 를 안 부르고 선다")
                    # ★`TraceHandle` 은 **`uid`** 를 갖는다. 앞 판은
                    #  `id`/`trace_id` 를 찾아 **전부 None** 이었다 —
                    #  사건 장부의 `trace_id` 가 통째로 비었다(실측).
                    _tid = getattr(_tr, "uid", None)
                    if not _tid:
                        raise rr.TraceUnavailable(
                            "trace 손잡이에 `uid` 가 없다 — 사건 장부와 "
                            "Opik 을 못 잇는다. provider 를 안 부르고 선다")
                    _w, _s_, _j, _dl, _clk = instrumented(
                        t, ident=ident, events=events, ctx=brief_ctx,
                        trace_id=_tid)
                    rec = rar.acquire_one(
                        {**t, "directive_native": build_directive(t)},
                        workdir=refdir, rel_root=SERVE_ROOT,
                        search=_s_, download=_dl, judge=_j,
                        era_tokens=rar.era_tokens_of(era_declaration),
                        write_brief=_w)
                    rec["trace_id"] = _tid
                    rec["run_nonce"] = _run_nonce
            except BaseException:
                # ★샀는지 아닌지 모른다 — **「안 샀다」로 적지 않는다**
                ref_jr.put(ident, None, status="uncertain",
                           meta={"subject_id": t["subject_id"]})
                raise
            ref_jr.put(ident, rec, meta={"subject_id": t["subject_id"]})
            rec["target"] = t
            records.append(rec)
            cov = [rd.get("era_coverage") or {} for rd in (rec.get("rounds") or ())]
            tot = sum(c.get("total", 0) for c in cov)
            wit = sum(c.get("with_era", 0) for c in cov)
            print(f"   {t['owner_type']:14} {str(t['surface_form'])[:16]:18} "
                  f"→ {rec['status']}  (질의 {tot} 중 시대 붙음 {wit})")
    _tally = events.tally()
    print(f"  ★사건 장부 {_tally['total']}줄 · 단계별 {_tally['by_stage']} "
          f"· 서로 다른 호출 {_tally['distinct_calls']} "
          f"· trace 붙은 것 {_tally['traced']}")
    _spent = _budget.snapshot()
    print(f"  ★실제 물리 전송 {_spent['used']}/{_spent['cap']} "
          f"(거절 {_spent['denied']})")

    out = jp.with_name(jp.stem + "_ref.json")

    # ★★**산 것을 덮어쓰지 않는다** (2026-08-31 실측 사고).
    #
    #  `--only` 로 3개만 다시 돌렸는데 기록 파일을 통째로 다시 써서 **앞서 산
    #  10개 기록이 사라졌다.** 사진 파일은 남았지만 어느 후보를 왜 골랐는지가
    #  통째로 날아갔다. `artifact/` 는 gitignore 라 되돌릴 곳도 없었다.
    #
    #  이제 **합친다** — 이번 판에 안 돈 대상은 앞 기록을 그대로 둔다.
    prev: Dict[str, Any] = {}
    if out.exists():
        try:
            prev = json.loads(out.read_text(encoding="utf-8"))
        except Exception as exc:                   # noqa: BLE001
            # ★못 읽는 것을 「없다」로 읽지 않는다 — 덮어쓰면 산 것을 잃는다
            raise RuntimeError(
                f"앞 기록을 못 읽는다 ({out}): {exc} — 덮어쓰면 산 것을 "
                "잃는다. 사람이 보고 정해야 한다") from exc
    merged = {str((r.get("target") or {}).get("final_id")
                  or (r.get("target") or {}).get("local_id")
                  or r.get("subject_id")): r
              for r in (prev.get("records") or ())}
    for r in records:
        merged[str(r["subject_id"])] = r
    records = [merged[k] for k in sorted(merged)]
    kept = len(merged) - len([r for r in records
                              if str(r["subject_id"]) in
                              {str(x["subject_id"]) for x in records}])

    out.write_text(json.dumps({
        "mode": mode[2:], "chunk": {k: v for k, v in got.items()
                                    if k not in ("raw",)},
        "targets": targets, "records": records,
        "parts_skipped": skipped,
        # ★표본이면 **표본이라고 적는다** — 전수인 척하지 않는다
        "sampled": sampled, "sample_per_owner": per_owner,
        "journal": {"bought": ref_jr.bought() if mode == "--live" else 0,
                    "reused": sum(1 for r in records if r.get("reused")),
                    "epoch": ref_jr.epoch if mode == "--live" else ""},
        "ran_this_pass": sorted(only) if only else "all",
        "approved_logical": cap,
        "owners_covered": owners, "owners_missing": missing,
        "note": ("★자동은 **구조만** 봤다. 고증이 맞나 · 참조가 그림을 낫게 "
                 "했나 · 이미지 품질은 **사람만** 본다."),
    }, ensure_ascii=False, indent=1, default=str), encoding="utf-8")
    print(f"  적었다: {out} (기록 {len(records)}개 — 이번 판에 안 돈 것은 "
          "앞 기록 그대로)")
    return 0


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