"""구간 판독 장부 — **끊긴 자리부터 이어 산다**. ★유료 0 (여기서 안 부른다).

## 왜

crash 뒤 재개하면 **이미 끝난 구간을 다시 산다** (Codex 2026-08-31). 구간
하나가 원문 전문 + 샷 전문이라 재구매는 그대로 돈이다.

## 신원으로 잇는다 — 순번이 아니라

`grounding_chunk.acquisition_identity` 를 **그대로** 쓴다. 팩·schema·모델·요청
계약이 바뀌면 신원이 달라져 **저절로 안 맞고 다시 산다** — 팩을 고쳤는데 옛
산출을 재사용하는 일이 구조적으로 없다.

★순번(`c0`·`c1`)으로 이으면 구간 나누기가 바뀔 때 **다른 원문의 답**을 이어
붙인다.

## 원자적으로 쓴다

`tmp` 에 쓰고 `os.replace` 로 바꾼다. 그냥 덮어쓰면 쓰는 도중에 끊겼을 때
다음 판이 **반쪽 파일**을 읽는다. (`grounding_screen_step._save_partial` 이
같은 관례를 쓴다 — 그 자리는 별도 리팩터링에서 이 함수로 모은다.)

## 예산은 **보내기 전에** 본다

    ①멈추라는 말이 왔나   ← 제일 먼저. 돈이 안 나가야 한다
    ②논리 상한을 넘나
    ③이미 산 것인가       ← 넘으면 안 사고 되쓴다

★상한을 넘으면 **선다**. 「넘었으니 그만 사자」로 조용히 줄이면 산출이 반쪽인데
완료로 보인다.
"""
from __future__ import annotations

import json
import logging
import os
import threading
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional

logger = logging.getLogger(__name__)

JOURNAL_CONTRACT_VERSION = "2.202608312000"
JOURNAL_NAME = "_chunk_journal.json"

STATUS_OK = "ok"
#: ★보냈는데 답을 못 받았다 — **샀을 수도 있다**. 자동으로 다시 안 산다.
STATUS_UNCERTAIN = "uncertain"
#: 사람이 「보내기 전에 거절됐다 — 안 샀다」고 정한 줄. ★append-only 정정 — 줄은 남고 상태만 바뀐다.
STATUS_NOT_SENT = "not_sent"
#: 자리를 잡았다가 보내기 전에 놓은 줄 — 예산·회계에 안 센다.
STATUS_RELEASED = "released"
#: provider 가 **받기를 거절**한 줄(503·429·연결 실패) — 처리된 적이 없으니 빈 자리, 다음 판이 다시 산다.
STATUS_REFUSED = "refused"
#: 답은 왔지만 **일이 끝나지 않은** 줄(마지막 라운드가 provider·예산으로 빈손) — 되쓰지 않고 자리도 비운다.
STATUS_INCOMPLETE = "incomplete"


def is_local_stop(exc: BaseException) -> bool:
    """**우리 쪽 문**이 provider 앞에서 세운 것인가 — 전송 예산(`ResearchCallBudgetExceeded`) · 주행 마감
    (`RunDeadlineExceeded`) · 사용자 중단(`run_control.is_abort`). 이 셋은 우리 코드가 **보내기 전에** 올리므로
    bytes 가 나가지 않았다 — `uncertain`(샀을 수도 있다) 이 아니라 `not_sent`(빈 자리) 다.
    ★PR #82 리뷰 2026-09-03: 사용자 정지 한 번이 대상을 `uncertain` 으로 봉인해 영영 못 사게 했다."""
    names = {c.__name__ for c in type(exc).__mro__}
    if names & {"ResearchCallBudgetExceeded", "RunDeadlineExceeded", "ImageCallBudgetExceeded"}:
        return True
    try:
        from app.core.run_control import is_abort
        return bool(is_abort(exc))
    except Exception:            # noqa: BLE001 — 판정 도구가 없으면 보수적으로 아니다
        return False


def is_provider_refusal(exc: BaseException) -> bool:
    """provider 가 **명시 응답으로** 거절했나(503 ServiceUnavailable · 429 RateLimit) — 예외 **종류**로만.
    ★연결 예외(APIConnectionError)·Timeout 은 **아니다** — 요청 bytes 가 나갔는지 클래스 이름으로 증명이
    안 된다(Codex BLOCK 2026-09-03 · `wrap_if_unconfirmed(request_left_the_process=)` 계약). 그것들은
    uncertain 으로 남아 사람의 settle 이나 운반층 증거 전에는 다시 안 산다."""
    names = {c.__name__ for c in type(exc).__mro__}
    return bool(names & {"ServiceUnavailableError", "RateLimitError"})
#: ★보내기 **직전에 잡아 둔 자리**. 아직 안 보냈지만 예산은 이미 먹었다.
#:  이것이 없으면 여러 worker 가 동시에 「아직 여유 있다」를 보고 다 나간다.
STATUS_RESERVED = "reserved"


def _epoch_of(contract: Dict[str, Any]) -> str:
    """획득 계약 → **판 표식**. ★같은 계약이면 같고, 바뀌면 다르다."""
    import hashlib

    return hashlib.sha256(
        json.dumps(contract or {}, sort_keys=True,
                   ensure_ascii=False).encode("utf-8")).hexdigest()[:12]


class BudgetExceeded(RuntimeError):
    """논리 상한을 넘는다. ★안 보내고 선다."""


class JournalUnreadable(RuntimeError):
    """장부 파일이 깨졌다. ★「안 샀다」가 **아니다** — 모르는 것이다."""


class NeedsHumanDecision(RuntimeError):
    """앞 판에 **답을 못 받은 호출**이 있다. 샀는지 아닌지 모른다."""


class TransferBroken(RuntimeError):
    """이관 기록이 깨졌다. ★「없던 일」로 안 읽는다 — 회계가 틀어진다."""


class ChunkJournal:
    """구간 판독 장부. ★적을 때마다 파일까지 내려쓴다."""

    def __init__(self, path: Path, *, contract: Optional[Dict[str, Any]] = None
                 ) -> None:
        self.path = Path(path)
        self.contract = dict(contract or {})
        # ★★**epoch** — 지금 계약으로 산 것만 이 판의 것이다.
        #
        #  앞 판은 `bought()` 가 **옛 계약 줄까지** 셌다. 그러면 팩·모델을 바꾼
        #  뒤 첫 구매가 「이미 상한만큼 샀다」로 막혀 **영구히 못 산다**
        #  (Codex 재현 2026-08-31). 옛 raw 는 감사용으로 **남기되** 예산은
        #  **이 판의 것만** 센다.
        self.epoch = _epoch_of(self.contract)
        # ★★구간을 **병렬로** 읽는다. 같은 장부에 여러 worker 가 적으므로
        #  ①`entries` 를 함께 고치는 것 ②파일에 내려쓰는 것 **둘 다** 잠근다.
        #  안 잠그면 나중 쓰기가 앞 쓰기를 통째로 덮어 **산 것이 사라진다**
        #  — 그러면 다음 판이 다시 사서 상한을 넘긴다 (Codex 2026-08-31).
        self._lock = threading.RLock()
        self._reused = 0
        #: ★★append-only **사건 로그** (Codex BLOCK 2026-09-02 밤): 같은 신원의 구매 시도마다 한
        #:  줄(`attempt`), `settlements` 는 사람의 정정 사건. `entries` 는 그것을 접은 **유효 뷰**다 —
        #:  거기 든 dict 는 사본이라 고쳐도 로그가 안 변한다. 앞 판은 `settle` 이 제자리로 status 를
        #:  바꾸고 `reserve` 가 줄을 통째로 덮어 결정 기록이 사라졌다(실측: replay 장부).
        self.events: List[Dict[str, Any]] = []
        self.settlements: List[Dict[str, Any]] = []
        self.entries: Dict[str, Dict[str, Any]] = {}
        #: **다른 장부로 옮긴** 줄들. ★원 줄은 그대로 두고 여기에 덧붙인다.
        #:  ★★★왜 (실측 2026-09-02): 재판정 4건이 구매 장부에 적혔고, 복사만
        #:   해서는 회계가 안 돌아온다 — `bought()` 가 12 대신 16 을 셌다.
        #:   지우면 역사가 사라지므로 **덧붙여 정정**한다 (Codex BLOCK).
        self.transfers: Dict[str, Dict[str, Any]] = {}
        self._saved_contract: Optional[Dict[str, Any]] = None
        if self.path.exists():
            try:
                d = json.loads(self.path.read_text(encoding="utf-8")) or {}
            except Exception as exc:                # noqa: BLE001
                # ★★**빈 것으로 읽으면 안 된다** (Codex 2026-08-31).
                #  장부가 깨진 것은 「아무것도 안 샀다」가 아니라 **「무엇을
                #  샀는지 모른다」**다. 비었다고 치면 이미 산 구간을 통째로
                #  다시 산다. 사람이 Opik 을 보고 정해야 한다.
                raise JournalUnreadable(
                    f"장부를 못 읽는다 ({self.path}): {exc} — 「안 샀다」가 "
                    "아니라 **무엇을 샀는지 모른다**는 뜻이다. 비었다고 치면 "
                    "이미 산 것을 다시 산다. Opik 을 보고 사람이 정해야 한다")
            self._saved_contract = d.get("contract")
            for e in d.get("calls") or []:
                self.events.append(dict(e))
            for s_ in d.get("settlements") or []:
                self.settlements.append(dict(s_))
            self._refold()
            for t in d.get("transfers") or []:
                self._adopt_transfer(t)

    def _refold(self) -> None:
        """사건 로그 + 정정 → 유효 뷰. ★신원마다 **마지막 시도**가 유효하고, 그 시도에 `not_sent`
        정정이 있으면 유효 상태만 그렇게 접는다(로그의 줄은 그대로)."""
        view: Dict[str, Dict[str, Any]] = {}
        for e in self.events:
            if e.get("status") == STATUS_RELEASED:
                view.pop(str(e.get("identity")), None)   # ★놓은 자리 — 유효 뷰에 없다
                continue
            cur = dict(e)
            if not cur.get("slot"):
                # ★slot 이 생기기 전에 산 줄 — 응답이 대상(subject_id)과 라운드 수를 알면 그것으로
                #  자리를 **유효 뷰에서만** 접는다(로그의 줄은 그대로). 실측 2026-09-02 밤: 옛 줄이
                #  제 신원으로 자리를 따로 먹어 같은 대상이 자리 둘을 차지했고 2차 pass 가 굶었다.
                resp = cur.get("response") if isinstance(cur.get("response"), dict) else {}
                sid = str(resp.get("subject_id") or "")
                if sid:
                    n_rounds = len(resp.get("rounds") or ())
                    cur["slot"] = f"{sid}:p{1 if n_rounds <= 1 else 2}"
                    cur["slot_derived"] = True
            view[str(e.get("identity"))] = cur
        voided = {(str(s_.get("identity") or ""), str(s_.get("of") or ""))
                  for s_ in self.settlements if s_.get("kind") == "alias_void"}
        for s_ in self.settlements:
            ident = str(s_.get("identity") or "")
            if s_.get("kind") == "alias":
                if (ident, str(s_.get("of") or "")) in voided:
                    continue                                # ★뒤에 무효로 적힌 alias
                # ★새 신원 → 옛 줄의 유효 뷰를 그대로 본다(같은 구매 · 같은 자리)
                src = view.get(str(s_.get("of") or ""))
                if src is not None:
                    view[ident] = {**src, "aliased_from": str(s_.get("of") or "")}
                continue
            cur = view.get(ident)
            if (cur is not None and s_.get("to") == STATUS_NOT_SENT
                    and int(cur.get("attempt") or 1) == int(s_.get("attempt") or 1)
                    and cur.get("status") == STATUS_UNCERTAIN):
                cur["status"] = STATUS_NOT_SENT
                cur["decision"] = {k: s_.get(k) for k in ("by", "why", "at")}
            elif (cur is not None and s_.get("to") == STATUS_INCOMPLETE
                    and int(cur.get("attempt") or 1) == int(s_.get("attempt") or 1)
                    and cur.get("status") == STATUS_OK):
                cur["status"] = STATUS_INCOMPLETE
                cur["decision"] = {k: s_.get(k) for k in ("by", "why", "at")}
            elif (cur is not None and s_.get("to") == STATUS_RELEASED
                    and int(cur.get("attempt") or 1) == int(s_.get("attempt") or 1)
                    and cur.get("status") == STATUS_RESERVED):
                # ★사람이 「죽은 worker 의 자리」를 놓았다 — 원 줄은 그대로, 유효 뷰만 접는다 (Codex PR #82 B)
                cur["status"] = STATUS_RELEASED
                cur["decision"] = {k: s_.get(k) for k in ("by", "why", "at")}
        self.entries = view

    def _latest_index(self, identity: str) -> Optional[int]:
        for i in range(len(self.events) - 1, -1, -1):
            if str(self.events[i].get("identity")) == str(identity):
                return i
        return None

    # ── 계약 ────────────────────────────────────────────────────────────
    def contract_drifted(self) -> bool:
        """저장된 계약과 지금 계약이 다른가. ★다르면 **이어 쓰지 않는다**."""
        return (self._saved_contract is not None
                and self._saved_contract != self.contract)

    def settle_uncertain(self, identity: str, *, bought: bool, why: str,
                         decided_by: str) -> Dict[str, Any]:
        """사람이 `uncertain` 줄을 매듭짓는다. ★안 샀다고 정하면 `not_sent` — 자리가 풀려
        다음 판이 다시 산다. 샀다고 정하는 것은 답이 없으니 **못 한다**(다시 사야 한다).

        ★실측 2026-09-02 밤 (attempt 21c47e1b→4d649ab6): 재판정 구매 하나가 canary 상한 문에
        **보내기 전에** 거절됐는데 장부는 예외 종류를 못 가려 `uncertain` 으로 적었고, 다음
        재개가 「샀는지 모른다 — 사람이 정해야 한다」로 섰다. 정하는 자리가 없어서 만들었다.
        """
        with self._lock:
            e = self.entries.get(str(identity))
            if not e or e.get("status") != STATUS_UNCERTAIN:
                raise ValueError(f"{str(identity)[:12]} 는 uncertain 줄이 아니다 ({(e or {}).get('status')})")
            if bought:
                raise ValueError("「샀다」로는 못 매듭짓는다 — 답이 없으니 다시 사야 한다. "
                                 "Opik 에서 답을 찾았으면 그 응답으로 put 하라")
            # ★원 줄은 손대지 않는다 — 정정 **사건**을 덧붙이고 유효 뷰만 접는다
            self.settlements.append({
                "identity": str(identity), "attempt": int(e.get("attempt") or 1),
                "from": STATUS_UNCERTAIN, "to": STATUS_NOT_SENT,
                "by": str(decided_by), "why": str(why),
                "at": datetime.now(timezone.utc).isoformat(timespec="seconds")})
            self._refold()
            self._flush()
            return dict(self.entries[str(identity)])

    def settle_reserved(self, identity: str, *, why: str, decided_by: str) -> Dict[str, Any]:
        """사람이 **reserved** 줄(자리만 잡고 죽은 worker)을 놓는다 — append-only 정정 사건. ★정확히 reserved 만.

        ★Codex PR #82 재리뷰(B): `release()` 는 calls 의 원행을 제자리 교체하고 사건을 안 남겨 운영자 CLI 에는 못 쓴다.
        여기서는 원행을 한 바이트도 안 건드리고 settlements 에 by/why 를 남긴다."""
        with self._lock:
            e = self.entries.get(str(identity))
            if not e or e.get("status") != STATUS_RESERVED:
                raise ValueError(f"{str(identity)[:12]} 는 reserved 줄이 아니다 ({(e or {}).get('status')}) — 놓을 자리가 아니다")
            self.settlements.append({
                "identity": str(identity), "attempt": int(e.get("attempt") or 1),
                "from": STATUS_RESERVED, "to": STATUS_RELEASED,
                "by": str(decided_by), "why": str(why),
                "at": datetime.now(timezone.utc).isoformat(timespec="seconds")})
            self._refold()
            self._flush()
            return dict(self.entries[str(identity)])

    def alias(self, identity: str, of: str, *, why: str) -> Dict[str, Any]:
        """새 신원이 옛 신원의 구매를 **잇는다** — append-only 사건. 둘은 같은 자리다."""
        with self._lock:
            if str(of) not in self.entries:
                raise ValueError(f"이관할 옛 줄이 없다 {str(of)[:12]}")
            rec = {"identity": str(identity), "kind": "alias", "of": str(of), "why": str(why),
                   "at": datetime.now(timezone.utc).isoformat(timespec="seconds")}
            self.settlements.append(rec)
            self._refold()
            self._flush()
            return dict(rec)

    def void_alias(self, identity: str, of: str, *, why: str) -> Dict[str, Any]:
        """앞서 적은 alias 를 **무효로** 한다 — append-only 사건. 새 신원은 다시 빈 자리다."""
        with self._lock:
            rec = {"identity": str(identity), "kind": "alias_void", "of": str(of), "why": str(why),
                   "at": datetime.now(timezone.utc).isoformat(timespec="seconds")}
            self.settlements.append(rec)
            self._refold()
            self._flush()
            return dict(rec)

    def settle_incomplete(self, identity: str, *, why: str, by: str) -> Dict[str, Any]:
        """사람이 `ok` 로 저장된 줄을 「일이 끝나지 않았다」로 매듭짓는다 — append-only 사건. 자리가 풀린다."""
        with self._lock:
            e = self.entries.get(str(identity))
            if not e or e.get("status") != STATUS_OK:
                raise ValueError(f"{str(identity)[:12]} 는 ok 줄이 아니다 ({(e or {}).get('status')})")
            self.settlements.append({"identity": str(identity), "attempt": int(e.get("attempt") or 1),
                                     "from": STATUS_OK, "to": STATUS_INCOMPLETE, "by": str(by), "why": str(why),
                                     "at": datetime.now(timezone.utc).isoformat(timespec="seconds")})
            self._refold()
            self._flush()
            return dict(self.entries[str(identity)])

    def append_settlement(self, identity: str, *, kind: str, by: str, why: str,
                          **meta: Any) -> Dict[str, Any]:
        """사람의 다른 정정·복구 사건을 덧붙인다 (예: 앞 판 코드가 결정 기록을 덮은 사실).
        ★유효 상태는 안 바꾼다 — 감사 기록만 남긴다."""
        with self._lock:
            rec = {"identity": str(identity), "kind": str(kind), "by": str(by), "why": str(why),
                   "at": datetime.now(timezone.utc).isoformat(timespec="seconds"), **meta}
            self.settlements.append(rec)
            self._flush()
            return dict(rec)

    def assert_no_uncertain(self) -> None:
        """앞 판에 답을 못 받은 것이 있으면 **선다**.

        ★「보냈는데 답을 못 받은 것」은 **샀을 수도 있다**. 자동으로 다시
        보내면 계획보다 많이 나가고 물리 상한을 넘긴다. 사람이 Opik 을 보고
        정해야 한다.
        """
        pend = [k for k, e in self.entries.items()
                if e.get("status") == STATUS_UNCERTAIN]
        if pend:
            raise NeedsHumanDecision(
                f"앞 판에 답을 못 받은 호출 {len(pend)}건이 있다 "
                f"— `python tools/grounding_audit/journal_settle.py list --journal <이 장부>` 로 보고 Opik 과 대조해 settle 한다 "
                f"({[k[:8] for k in sorted(pend)]}) — 샀는지 아닌지 모른다. "
                "Opik 을 보고 사람이 정한 뒤 장부를 고쳐야 한다")

    # ── 읽기 ────────────────────────────────────────────────────────────
    def get(self, identity: str) -> Optional[Dict[str, Any]]:
        """★**같은 epoch 의** 성공 응답만 되쓴다.

        옛 계약 줄은 남아 있어도 **안 쓴다** — 다른 팩·모델·요청 계약으로 산
        것이라 같은 대상의 답이 아니다.
        """
        if str(identity) in self.transfers:
            # ★이 lane 의 것이 아니다 — 옮긴 장부가 답을 갖고 있다
            return None
        e = self.entries.get(str(identity))
        if not e or e.get("status") != STATUS_OK:
            return None
        if e.get("epoch") != self.epoch:
            return None
        return e.get("response")

    def bought(self) -> int:
        """**이 판(epoch)에서** 보낸 횟수. ★성공만 세면 답 못 받은 것이 샌다.

        ★옛 계약 줄은 안 센다 — 세면 계약을 바꾼 뒤 첫 구매가 막힌다.
        """
        return sum(1 for k, e in self.entries.items()
                   if k not in self.transfers
                   and e.get("epoch") == self.epoch
                   and e.get("status") in (STATUS_OK, STATUS_UNCERTAIN,
                                           STATUS_RESERVED))

    def bought_slots(self, among: Optional[Any] = None) -> int:
        """**이 판(epoch)에서** 찬 논리 자리 수 — 상한은 이것으로 센다.

        ★★같은 `slot` 의 줄은 **하나**로 센다 (2026-09-02 밤 실측): 처리 계약이 바뀌어
        행 집합이 달라지면 merge 는 **같은 자리**를 새 신원으로 다시 산다. 그것을 두 번째
        구매로 세면 상한 2(구간 1 + merge 1)에 막혀 force 재실행이 통째로 죽는다. 옛 줄은
        감사용으로 남고(`bought()` 는 여전히 보낸 횟수), 자리 수만 상한에 댄다.
        slot 이 없는 줄은 제 신원이 자리다 — 옛 뜻 그대로.
        """
        slots = set()
        for k, e in self.entries.items():
            if k in self.transfers or e.get("epoch") != self.epoch:
                continue
            if e.get("status") not in (STATUS_OK, STATUS_UNCERTAIN, STATUS_RESERVED):
                continue
            sl = str(e.get("slot") or k)
            # ★`among` 이 있으면 **지금 대상의 자리만** 센다 — 옛 의무 ID 의 고아 자리가 지금 상한을 먹지 않게.
            #  (PR #82 리뷰 2026-09-03: 앞 판은 그 고아 수만큼 상한을 **넓혀서** 승인한 수가 최대가 아니었다)
            if among is not None and e.get("slot") and sl.rsplit(":", 1)[0] not in among:
                continue
            slots.add(sl)
        return len(slots)

    def reused(self) -> int:
        """이 판에서 **되쓴** 횟수. ★산 것과 갈라 센다."""
        with self._lock:
            return self._reused

    def kept_from_other_epochs(self) -> int:
        """옛 계약 줄 — **버리지 않았다**는 감사 수."""
        return sum(1 for e in self.entries.values()
                   if e.get("epoch") != self.epoch)

    # ── 쓰기 ────────────────────────────────────────────────────────────
    def put(self, identity: str, response: Any, *, status: str = STATUS_OK,
            meta: Optional[Dict[str, Any]] = None) -> None:
        with self._lock:
            i = self._latest_index(identity)
            prev = self.events[i] if i is not None else {}
            rec = {
                "identity": str(identity), "status": status,
                "epoch": self.epoch, "response": response,
                "attempt": int(prev.get("attempt") or 1) if prev else 1,
                # ★자리 이름은 reserve 가 적은 것을 잃지 않는다
                **({"slot": prev["slot"]} if prev.get("slot") else {}),
                **(meta or {})}
            if prev and prev.get("status") == STATUS_RESERVED and prev.get("epoch") == self.epoch:
                self.events[i] = rec               # ★같은 구매 시도의 마무리 — 자리 잡기 줄을 채운다
            else:
                rec["attempt"] = int(prev.get("attempt") or 0) + 1 if prev else 1
                self.events.append(rec)            # ★새 시도 — 옛 줄은 그대로
            self._refold()
            self._flush()

    def _adopt_transfer(self, rec: Dict[str, Any]) -> None:
        """저장된 이관 줄 하나를 읽는다. ★깨졌으면 **선다**."""
        ident = str((rec or {}).get("identity") or "")
        to = str((rec or {}).get("moved_to") or "")
        if not ident or not to:
            raise TransferBroken(
                f"이관 기록에 대상이나 옮긴 곳이 없다 ({rec!r}) — 회계를 "
                f"모르는 채 안 이어 간다")
        if ident not in self.entries:
            raise TransferBroken(
                f"이관 기록이 가리키는 줄 {ident[:12]} 이 이 장부에 없다 — "
                f"무엇을 뺐는지 모른다")
        self.transfers[ident] = dict(rec)

    def transfer_out(self, identity: str, *, moved_to: str,
                     why: str) -> Dict[str, Any]:
        """이 줄이 **다른 장부의 몫**임을 덧붙여 적는다. ★원 줄은 안 지운다.

        ★★★`ChunkJournal` 은 **무엇이 어느 lane 인지 모른다** — 부르는 쪽이
        안다 (Codex 2026-09-02: 「일반 ChunkJournal 전역에 replay 접두 의미를
        하드코딩하지 말라」). 여기는 「이 줄은 내 몫이 아니다」만 적는다.

        Raises:
            TransferBroken: 이 장부에 없는 줄이다.
        """
        from datetime import datetime, timedelta, timezone

        ident = str(identity)
        with self._lock:
            rec = {"identity": ident, "moved_to": str(moved_to),
                   "why": str(why),
                   "recorded_kst": datetime.now(
                       timezone(timedelta(hours=9))).isoformat(
                           timespec="seconds"),
                   "★means": ("원 줄은 그대로 두고 **회계에서만** 뺀다 — "
                              "무엇이 언제 어디로 갔는지가 남아야 한다")}
            self._adopt_transfer(rec)
            self._flush()
            return rec

    def transferred_out(self) -> int:
        """다른 장부로 넘긴 줄 수. ★감사용."""
        return len(self.transfers)

    def reserve(self, identity: str, *, cap: int, slot: Optional[str] = None,
                slot_universe: Optional[Any] = None) -> bool:
        """보내도 되는 **자리를 잡는다**. ★잠금 **안에서** 세고 적는다.

        Returns:
            True 면 이 호출이 자리를 잡았으니 보낸다. False 면 **다른 worker 가
            이미 같은 것을 잡았다** — 기다렸다 그 답을 되쓴다.

        Raises:
            BudgetExceeded: 이 판의 상한을 넘는다.

        ★앞 판은 `get → bought → send` 가 잠금 **밖**이었다. 상한 1 에 worker
        12개를 서로 다른 신원으로 태우니 **12개가 다 나갔다** (Codex 재현).
        세는 것과 자리를 잡는 것이 **한 임계 구역**이어야 한다.
        """
        with self._lock:
            e = self.entries.get(str(identity))
            if (e is not None and e.get("epoch") == self.epoch
                    and e.get("status") not in (STATUS_NOT_SENT, STATUS_RELEASED, STATUS_REFUSED, STATUS_INCOMPLETE)):
                return False            # ★이미 누가 잡았거나 샀다 (`not_sent`·`released` 는 빈 자리다)
            # ★같은 자리를 다시 사는 것(slot 이 이미 찬 것)은 상한을 안 늘린다
            taken = self.bought_slots(among=slot_universe)
            mine_is_new = slot is None or not any(
                e.get("slot") == slot and e.get("epoch") == self.epoch
                and e.get("status") in (STATUS_OK, STATUS_UNCERTAIN, STATUS_RESERVED)
                for k, e in self.entries.items() if k not in self.transfers)
            if mine_is_new and taken + 1 > cap:
                raise BudgetExceeded(
                    f"논리 상한 {cap} 을 넘는다 (이 판에서 이미 "
                    f"{taken}자리 · 보낸 횟수 {self.bought()}) — 여기서 선다. "
                    "조용히 줄이면 반쪽 산출이 완료로 보인다")
            i = self._latest_index(identity)
            prev = self.events[i] if i is not None else {}
            self.events.append({
                "identity": str(identity), "status": STATUS_RESERVED,
                "epoch": self.epoch, "response": None,
                # ★앞 시도(uncertain→not_sent · 옛 계약)는 그대로 두고 **새 시도**로 적는다
                "attempt": int(prev.get("attempt") or 0) + 1 if prev else 1,
                **({"slot": str(slot)} if slot else {})})
            self._refold()
            self._flush()
            return True

    def release(self, identity: str) -> bool:
        """**안 나간 것이 확실한** 자리를 놓는다. ★샀으면 절대 안 놓는다.

        Returns:
            놓았으면 True. 없거나 이미 답이 있으면 False.

        ★★자리를 잡았는데 provider 를 **부르기도 전에** 막혀 끝난 판이 있다
        (부모 trace 가 안 열림·승인 밖 capability 등). 그 줄을 그대로 두면
        다음 판이 `reserve` 에서 막혀 **사람에게 장부를 고치라**고 요구하게
        된다 — 자동화 정책과 반대다 (Codex 2026-08-31).

        ★**「샀는지 모른다」에는 쓰지 않는다.** 그건 놓으면 두 번 산다.
        부르는 쪽이 「한 번도 안 나갔다」를 **관측으로** 보인 자리에만 쓴다.
        """
        with self._lock:
            e = self.entries.get(str(identity))
            if e is None:
                return False
            if e.get("status") in (STATUS_OK, STATUS_UNCERTAIN):
                return False
            i = self._latest_index(identity)
            if i is not None:
                # ★지우지 않는다 — 자리 잡기 줄을 「놓았다」로 적고 유효 뷰에서 뺀다
                self.events[i] = {**self.events[i], "status": STATUS_RELEASED}
            self._refold()
            self.entries.pop(str(identity), None)
            self._flush()
            return True

    def note_reuse(self) -> None:
        with self._lock:
            self._reused += 1

    def _flush(self) -> None:
        """★`self._lock` 을 **쥔 채** 부른다."""
        self.path.parent.mkdir(parents=True, exist_ok=True)
        # ★임시 이름을 **호출마다 다르게** 둔다. 같은 이름이면 worker 둘이
        #  같은 파일에 동시에 쓰고, `os.replace` 가 **반쪽을 확정**한다.
        tmp = self.path.with_suffix(f".{uuid.uuid4().hex[:8]}.tmp")
        tmp.write_text(json.dumps({
            "contract_version": JOURNAL_CONTRACT_VERSION,
            "contract": self.contract,
            # ★append-only 사건 로그 — 같은 신원의 시도마다 한 줄, 정정은 따로
            "calls": list(self.events),
            "settlements": list(self.settlements),
            # ★덧붙이기만 한다 — 원 줄을 지우지 않고 회계만 바로잡는다
            "transfers": [self.transfers[k] for k in sorted(self.transfers)],
        }, ensure_ascii=False), encoding="utf-8")
        # ★원자적으로. 덮어쓰다 끊기면 다음 판이 반쪽을 읽는다.
        os.replace(tmp, self.path)


def buy_or_reuse(journal: ChunkJournal, identity: str, *,
                 cap: int, send: Callable[[], Any],
                 stop_check: Optional[Callable[[], None]] = None,
                 slot: Optional[str] = None,
                 complete: Optional[Callable[[Any], bool]] = None,
                 slot_universe: Optional[Any] = None) -> Any:
    """이미 샀으면 되쓰고, 아니면 **순서를 지켜** 산다.

    순서가 계약이다 —

        ①멈추라는 말      ← 제일 먼저. 돈이 안 나가야 한다
        ②자리 잡기        ← **잠금 안에서** 세고 적는다. 넘으면 선다
        ③구매

    ★②를 잠금 밖에 두면 worker 여럿이 동시에 「아직 여유 있다」를 보고 **다
    나간다**. 실제로 상한 1 에 12개가 나갔다 (Codex 2026-08-31).

    ★같은 신원으로 동시에 들어와도 **한 번만** 산다 — 자리를 못 잡은 쪽은
    잡은 쪽이 끝나기를 기다렸다 그 답을 되쓴다.

    Raises:
        BudgetExceeded: 이 판의 논리 상한을 넘는다.
    """
    import time

    hit = journal.get(identity)
    if hit is not None:
        journal.note_reuse()
        return hit
    if stop_check is not None:
        stop_check()

    if not journal.reserve(identity, cap=cap, slot=slot, slot_universe=slot_universe):
        # ★다른 worker 가 잡았다. 그 답이 나올 때까지 **짧게** 기다린다.
        for _ in range(600):                      # 최대 60초
            hit = journal.get(identity)
            if hit is not None:
                journal.note_reuse()
                return hit
            e = journal.entries.get(str(identity)) or {}
            if e.get("status") == STATUS_UNCERTAIN:
                raise NeedsHumanDecision(
                    f"같은 신원 {identity[:8]} 을 다른 worker 가 사다가 답을 "
                    "못 받았다 — 샀는지 모른다. 사람이 정해야 한다")
            time.sleep(0.1)
        raise NeedsHumanDecision(
            f"같은 신원 {identity[:8]} 을 잡은 worker 가 안 끝난다 — "
            "두 번 사지 않으려고 선다")

    from app.core.research_call_budget import OutboundAttempt, outbound_attempt
    att = OutboundAttempt()
    try:
        with outbound_attempt(att):
            resp = send()
    except BaseException as exc:
        if is_provider_refusal(exc):
            # ★provider 가 받기를 거절했다 — 산 것이 없다. 자리를 비워 다음 판이 다시 산다.
            journal.put(identity, None, status=STATUS_REFUSED,
                        meta={"why": f"{type(exc).__name__}: {str(exc)[:160]}"})
            raise
        if is_local_stop(exc) and att.armed and att.count == 0:
            # ★우리 문(예산·마감·중단)이 세웠고 **이 시도에서 나간 전송이 0** 이라는 계수 증거가 있다 — 빈 자리.
            #  ★Codex BLOCK 2026-09-03: send 하나가 전송 여럿(저작→검색→받기→판정)이라 「우리 문이 세웠다」만으로는
            #   앞 전송이 안 나갔다는 증명이 못 된다. 하나라도 나갔거나(count>0) 셀 수 없었으면(armed=False) uncertain.
            journal.put(identity, None, status=STATUS_NOT_SENT,
                        meta={"why": f"local_stop {type(exc).__name__}: {str(exc)[:160]}", "outbound": 0})
            raise
        # ★보냈는지 아닌지 모르는 자리다. **샀을 수도 있다**고 적는다 —
        #  「안 샀다」로 적으면 다음 판이 다시 사서 상한을 넘긴다.
        journal.put(identity, None, status=STATUS_UNCERTAIN,
                    meta={"why": f"{type(exc).__name__}: {str(exc)[:160]}",
                          "outbound": int(att.count), "counted": bool(att.armed)})
        raise
    if complete is not None and not complete(resp):
        # ★★답은 왔지만 일이 끝나지 않았다(마지막 라운드가 상한·provider 로 빈손) — 감사용으로 남기되
        #  **되쓰지 않고 자리도 비운다** (실측 2026-09-03 새벽: 상한에 막힌 2차 결과가 ok 로 저장돼
        #  다음 판이 그 빈손을 되썼고 O02 는 영원히 retryable 이었다).
        journal.put(identity, resp, status=STATUS_INCOMPLETE)
        return resp
    journal.put(identity, resp)
    return resp