"""fixture 를 **production 스텝으로** scene_detail 까지 돌린다. ★상한 안에서.

Codex (2026-08-31) —

> 부트스트랩을 먼저 **독립 cap/journal/trace** 로 끝내고, pipeline 시작 때
> **새 budget** 을 여십시오. process-wide budget 은 전체 run 동안 계속 armed
> 로 두어 뜻밖의 dispatch 가 생기면 즉시 counted 되고 **per-step delta** 에
> 남게 하십시오.

## 왜 한 스텝씩 부르나

`run_steps_batch` 는 목록을 받아 **안에서 죽 돈다**. 그러면 어느 스텝이 몇 번
보냈는지 못 가른다. 그래서 **한 스텝씩** 부르고 앞뒤로 예산을 찍는다 —

    ①무료로 분류한 스텝이 **보냈으면** 분류가 틀린 것이다 → 선다
    ②스텝이 제 논리 상한을 넘으면 **다음 스텝 앞에서** 선다
    ③전체가 emergency ceiling 에 닿으면 그 판은 **inconclusive** 로 보존한다

★자동으로 늘리거나 다시 사지 않는다.
"""
from __future__ import annotations

import json
import os
import sys
import uuid
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Sequence

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

from tools.grounding_audit import canary_cost_table as ct  # noqa: E402
from tools.grounding_audit import canary_isolation as ci  # noqa: E402

KST = timezone(timedelta(hours=9))

TRACE_TAG = "op:bundle-canary-fixture"
TRACE_THREAD = "bundle-canary-fixture"


class CanaryStopped(RuntimeError):
    """상한이나 분류가 어긋났다. ★자동으로 안 늘린다."""


class StepDidNotFinish(RuntimeError):
    """스텝이 **끝나지 않았다**. ★「예외가 안 나왔다」로 completed 를 안 닫는다.

    ★★★실측 (2026-09-02 유료 canary ①): `grounding_chunk` 가
    `AttributeError` 로 죽었는데 장부에 **`status: completed`,
    `stopped_at: null`** 로 적혔다. `analysis_dispatch_service.run_steps_batch`
    가 안에서 실패를 잡아 기록하고 `None` 을 돌려주기 때문에, 밖에서는
    「예외가 안 나왔다」로 보인다.
    그래서 **부른 뒤 durable 상태를 되읽는다** (Codex BLOCK 2026-09-02).
    """


#: 스텝이 **끝났다**고 볼 상태들. ★`step_runner` 가 의존을 통과시키는 것과
#:  같은 집합이다(`step_runner.py:407·474`) — 여기서 따로 정하지 않는다.
FINISHED_STATES = ("completed", "partial", "not_applicable")
#: ★★cap 0 으로 **되쓰는** 상태는 이 둘뿐이다 (Codex 재리뷰 2026-09-02). `partial`
#:  은 의존 통과 상태일 수는 있어도 되쓰기 상태가 아니다 — production resume 이
#:  그 스텝의 **남은 일**을 다시 사므로 그 몫의 상한을 열어 둬야 한다. 한 벌
#:  (`FINISHED_STATES`)로 둘을 합쳤더니 partial 스텝의 재시도 27건이 cap 0 에서
#:  전부 거절되고 그 거절이 문을 세웠다(attempt f8d157fadf6d · 유료 0).
REUSE_ZERO_STATES = ("completed", "not_applicable")


def contract_drift_of(step_id: str, project_id: str, episode_id: str,
                      project_config: Optional[Dict[str, Any]]) -> Optional[str]:
    """runner 가 이 스텝의 완료 CP 를 **어긋났다**고 볼지 — runner 의 그 함수로 미리 묻는다.

    ★★★실측 2026-09-02 밤: runner 는 지문 어긋남을 `AppError(409)` 로 올리는데
    `run_steps_batch` 가 안에서 삼키고 로그만 남긴다. canary 는 옛 CP 를 `completed` 로
    되읽어 **지나갔고**, grounding_chunk 가 옛 처리 계약 산출 그대로 하류로 갔다.
    같은 규칙을 여기 다시 적지 않는다 — runner 인스턴스의 `_check_cp_mismatch` 를 부른다."""
    from app.core.database import SessionLocal
    from app.services.analysis_dispatch_service import get_step_runner

    db = SessionLocal()
    try:
        r = get_step_runner(step_id, project_id, episode_id, db,
                            dict(project_config or {}), {})
        cp = r.load_checkpoint()
        if not cp:
            return None
        return r._check_cp_mismatch(cp) or None
    finally:
        db.close()


def current_config_hash_of(step_id: str, project_id: str, episode_id: str,
                           project_config: Optional[Dict[str, Any]]) -> str:
    """runner 인스턴스가 **지금 코드로** 계산하는 이 스텝의 config_hash — hash adoption 의 새 값은 여기서만 온다."""
    from app.core.database import SessionLocal
    from app.services.analysis_dispatch_service import get_step_runner

    db = SessionLocal()
    try:
        r = get_step_runner(step_id, project_id, episode_id, db,
                            dict(project_config or {}), {})
        return str(r._config_hash())
    finally:
        db.close()


def completion_debt_of(step_id: str, project_id: str, episode_id: str,
                       project_config: Optional[Dict[str, Any]]) -> Optional[str]:
    """runner 가 이 스텝의 완료 CP 를 **아직 안 끝났다**고 볼지 — runner 의 `_safe_verify_completion` 로 묻는다.
    ★같은 규칙을 여기 다시 적지 않는다. 미완료면 그 까닭(missing 을 이어 붙인 글), 끝났으면 None.
    ★실측 4398a55dc0bb (2026-09-03 05:30): 중앙 스텝이 completed 인데 취소에 맞은 9줄이 빚으로 남았다 — canary 가 durable
    completed 만 보고 cap 0 을 주면 runner 의 RERUN_SELF 가 닫힌 문에 부딪힌다."""
    from app.core.database import SessionLocal
    from app.services.analysis_dispatch_service import get_step_runner

    db = SessionLocal()
    try:
        r = get_step_runner(step_id, project_id, episode_id, db,
                            dict(project_config or {}), {})
        if not r.load_checkpoint():
            return None
        rep = r._safe_verify_completion()
        if getattr(rep, "is_complete", True):
            return None
        return "; ".join(str(m) for m in (getattr(rep, "missing", None) or ())) or "미완료"
    finally:
        db.close()


def code_transition_covers_head(root: Path) -> bool:
    """이 run 의 장부에 **지금 코드**로 잇는 전이가 적혀 있나 (마지막 전이의 `to_tip`)."""
    from tools.grounding_audit import canary_run as cr    # ★늦게 — 순환 import 회피

    p = root / "pipeline_attempts.json"
    if not p.is_file():
        return False
    rows = [e for e in (json.loads(p.read_text(encoding="utf-8")) or [])
            if isinstance(e, dict) and e.get("kind") == "code_transition"]
    if not rows:
        return False
    tip = cr.git_tip()                      # ★dict — {"tip", "clean", "dirty_files"} (실측: str 로 읽어 항상 거짓이었다)
    head = str((tip.get("tip") if isinstance(tip, dict) else tip) or "")
    to = str(rows[-1].get("to_tip") or "")
    return bool(head) and bool(to) and (head.startswith(to) or to.startswith(head))


def plan_reentry(durable: Optional[str], drift: Optional[str], transition_ok: bool,
                 reopened: bool, debt: Optional[str] = None) -> Dict[str, Any]:
    """끝난 스텝을 다시 들어갈 때의 결정 — 순수 함수라 시험이 표로 잠근다.

        끝났고 · 어긋남 없음 · 빚 없음 · 되열기 아님 → resume · cap 0 (되쓴다)
        끝났고 · 어긋남 없음 · **빚 있음**(runner 의 verify_completion 이 미완료) → resume · **정상 cap**
            (runner 가 RERUN_SELF 로 다시 돈다 — 장부 되쓰기 · 빚진 대상만 산다. 실측 4398a55dc0bb: 취소에 맞은 9줄)
        끝났고 · 어긋남 있음 · 전이 승인 → **force** (하류 무효화 · 정상 cap)
            ★「자기만 다시 돌고 하류 보존」은 **없다** (Codex BLOCK 2026-09-03): 다시 도는 스텝은 실제 LLM 을 부르므로 산출이
             달라질 수 있고 그러면 보존한 하류는 다른 계보다. hash 조리법만 바뀐 경우(실측 f7cc45c576c0 entity_detail · 팩 지문
             결속)는 재실행이 아니라 **정확한 hash adoption**(canary_hash_adoption.adopt · data 바이트 불변 · tuple 결속)으로
             어긋남 자체를 없애 이 표의 첫 줄(cap 0 · 되쓴다)로 들어온다.
        끝났고 · 어긋남 있음 · 전이 없음 → 선다
        안 끝났거나 되열기 → resume · 정상 cap
    """
    if durable in REUSE_ZERO_STATES and not reopened:
        if drift:
            if transition_ok:
                return {"mode": "force", "cap_zero": False, "stop_why": None}
            return {"mode": None, "cap_zero": False,
                    "stop_why": f"지문 어긋남({drift}) — 승인된 코드 전이가 없다"}
        if debt:
            return {"mode": "resume", "cap_zero": False, "stop_why": None}
        return {"mode": "resume", "cap_zero": True, "stop_why": None}
    if durable == "completed" and reopened:
        # ★사람이 지목해 다시 연 끝난 스텝은 **force** 다 — resume 이면 runner 가 깨끗한 CP 를
        #  SKIP 해 아무것도 안 산다(실측 2026-09-02 밤: 조사 스텝을 다시 열어 굶은 여섯을 사려면).
        #  저널이 그대로라 이미 산 신원은 재구매 0 이다.
        return {"mode": "force", "cap_zero": False, "stop_why": None}
    return {"mode": "resume", "cap_zero": False, "stop_why": None}


def durable_status(project_id: str, episode_id: str, step_id: str,
                   *, projects_dir: Optional[str] = None) -> Optional[str]:
    """그 스텝의 **파일에 남은** 상태. 체크포인트가 없으면 `None`.

    ★`None` 은 「안 돌았다」다 — 실측에서 죽은 스텝과 그 뒤는 manifest 가
    **아예 없었다**.
    """
    import json as _json

    from app.core.config import settings

    base = Path(projects_dir or settings.projects_dir)
    cp = (base / project_id / "checkpoints" / "episodes" / episode_id
          / step_id / "manifest.json")
    if not cp.exists():
        return None
    try:
        return str((_json.loads(cp.read_text(encoding="utf-8")) or {}
                    ).get("status") or "") or None
    except Exception:                               # noqa: BLE001
        return "★못 읽음"


def assert_step_finished(project_id: str, episode_id: str, step_id: str
                         ) -> str:
    """★부른 **뒤** 되읽어 확인한다. 안 끝났으면 선다."""
    got = durable_status(project_id, episode_id, step_id)
    if got is None:
        raise StepDidNotFinish(
            f"{step_id} 의 체크포인트가 **없다** — 스텝이 죽었거나 안 돌았다. "
            f"「예외가 안 나왔다」로 completed 를 닫지 않는다")
    if got not in FINISHED_STATES:
        raise StepDidNotFinish(
            f"{step_id} 의 상태가 {got!r} 다 — {FINISHED_STATES} 중 하나여야 "
            f"한다")
    return got


def run_pipeline(*, run_id: str, project_id: str, episode_id: str,
                 plan: Dict[str, Any], caps: Dict[str, int],
                 emergency_counted: int, approved_image_calls: int,
                 approved_search: int = 0, approved_downloads: int = 0,
                 attempt_id: Optional[str] = None,
                 project_config: Optional[Dict[str, Any]] = None,
                 reopen: Sequence[str] = (),
                 live: bool = False,
                 before_step: Optional[Callable[[str], Any]] = None) -> Dict[str, Any]:
    """**closure 전부**를 차례로 하나씩 돌린다. ★`live` 아니면 안 산다.

    `before_step(step_id)` — **살 스텝을 부르기 직전**(문 열기 전·provider 앞)에 불린다. producer 산출로
    상한을 세는 문(`canary_run.producer_cap_gate_before`)이 여기 걸린다. 던지면 `stopped_at` 을 적고 그대로
    올린다 — `run_steps_batch` 는 안 불린다. 반환값은 `per_step[s]["producer_gate"]` 에 남는다.

    ★★★**적용 안 되는 스텝도 뺀 채로 두면 안 된다** (실측 2026-09-01).
    `StepRunner.check_gate` 는 의존 스텝의 **`step_run` 기록**을 본다 —
    기록이 **아예 없으면** 막는다(`step_runner.py:224`).

    ★★★**정지선은 이 attempt 가 아니라 run 전체에 걸린다** (Codex BLOCK
    2026-09-01). 앞 판은 재개마다 `cap=92` 를 새로 열어 **몇 번이든 92 를 더
    쓸 수 있었다**. 이제 `정지선 − 누적 사용` 으로 연다. 장부를 못 믿으면
    **provider 0 으로 선다**.

    ★★★**글 예산은 이미지를 못 센다** (실측 2026-09-01). 이 closure 안의
    `floor_plan_render` 는 `gpt-image-2` 를 사는데, 그 자리는 글 문 셋을
    하나도 안 지난다 — 글 장부에는 0 으로 적히고 실제로는 돈이 나간다.
    그래서 `approved_image_calls` 를 **따로 받아** 이미지 문도 같이 잠근다.
    승인이 없으면 0 이고, 0 이면 **문 앞에서 선다**.
    """
    from tools.grounding_audit.canary_image_budget import (canary_image_scope,
                                                           image_delta)
    from tools.grounding_audit.canary_text_budget import (canary_step_cap,
                                                          canary_text_scope)

    steps: List[str] = ct.execution_steps_to(plan.get("target")
                                             or "scene_detail")
    # ★★★**다시 여는 스텝만 도는 판** (2026-09-02 실측). closure 전부를 돌면
    #  `partial` 로 끝난 스텝이 **제 실패를 다시 시도**한다 — 그런데 그것은
    #  `already` 라 상한이 0 이라 전부 거절되고, 그 거절이 문을 세운다.
    #  실제로 `entity_t2i` 의 거절 16건이 남은 상한 23을 다 먹어 정작 하려던
    #  재판정이 **시작도 못 했다**(Opik 성공 0 — 한 푼도 안 나갔다).
    #  ★재판정은 앞 스텝을 하나도 안 쓴다 — 체크포인트가 이미 다 있다.
    if reopen:
        want = [x for x in steps if x in set(reopen)]
        missing = sorted(set(reopen) - set(steps))
        if missing:
            raise CanaryStopped(
                f"다시 열라는 스텝 {missing} 이 이 closure 에 없다 — "
                f"무엇을 도는지 모르는 채 안 연다")
        steps = want
    free = set(plan["applied_free"])
    metered = set(plan["applied_metered"])
    root = ci.root_dir(run_id)
    # ★★신원을 **바깥이 먼저** 만들 수 있다 — 운반 기록이 첫 전송 전에 그
    #  id 로 결속돼야 한다. 앞 판은 여기서 만들어서, 바깥의 운반 줄이
    #  **빈 문자열**로 적혔다 (Codex BLOCK 2026-09-02).
    attempt_id = str(attempt_id or uuid.uuid4().hex[:12])
    out = {"run_id": run_id, "attempt_id": attempt_id, "live": live,
           "steps": steps, "free_steps": sorted(free), "per_step": {},
           # ★dry 에서도 남는다 — 무엇을 다시 열려 했는지가 증거다
           "reopened": sorted(set(reopen or ())),
           "stopped_at": None,
           "approved_image_calls": int(approved_image_calls),
           "★means": ("closure 전부를 돌린다 — 적용 안 되는 스텝도 **돌아서** "
                      "`not_applicable` 로 찍혀야 뒤가 지나간다")}
    # ★★이 run 이 **이미 쓴 것**을 빼고 연다 — dry 에서도 **같은 수**를 낸다
    #  (Codex 2026-09-02: 승인안에 「누계/이번에 여는 것/정지선」이 문마다 보여야 한다).
    #  장부 읽기라 유료 0 이고, 열린 attempt 가 있으면 dry 도 여기서 선다.
    #  ★새 run(장부 없음)의 dry 는 옛 그대로 — 누계 0 을 지어내지 않는다.
    if live or (root / "pipeline_attempts.json").is_file():
        before_total = cumulative_used(root)
        scope_cap = remaining_cap(root, ceiling=emergency_counted)
        out.update({"ceiling": int(emergency_counted),
                    "cumulative_before": before_total, "scope_cap": scope_cap})
        # ★★이미지도 **run 전체** 정지선 (Codex BLOCK 2026-09-02) — 이미 산 것을 빼고 연다
        image_before = cumulative_image_used(root)
        image_scope_cap = remaining_image_cap(root, ceiling=int(approved_image_calls))
        out.update({"image_ceiling": int(approved_image_calls),
                    "image_cumulative_before": image_before,
                    "image_scope_cap": image_scope_cap})
    if not live:
        out["note"] = "★안 돌렸다 — `live=True` 여야 산다"
        return out


    from app.services.analysis_dispatch_service import run_steps_batch

    started = datetime.now(KST).isoformat(timespec="seconds")
    # ★★★**사기 전에** 자리를 durable 하게 잡는다. 프로세스가 통째로
    #  사라져도 「열린 판이 있었다」가 남아, 다음 재개가 **덜 세지 않는다**.
    reserve_attempt(root, {
        "attempt_id": attempt_id, "started_kst": started, "source": "measured",
        "ceiling": int(emergency_counted), "cumulative_before": before_total,
        "scope_cap": scope_cap, "per_step": {}, "stopped_at": None,
        "approved_image_calls": int(approved_image_calls),
        "image_ceiling": int(approved_image_calls),
        "image_cumulative_before": image_before,
        "image_scope_cap": image_scope_cap,
        "★why_open": ("열린 채로 남았다면 그 판이 얼마 썼는지 모른다는 뜻이다 — "
                      "Opik·provider 로그를 보고 사람이 닫아야 한다"),
    })

    def _mark(status: str) -> None:
        """같은 attempt 를 **제자리에서** 닫는다. ★앞의 것은 안 건드린다."""
        b = out.get("budget") or {}
        im = out.get("image_budget") or {}
        update_attempt(root, attempt_id, {
            "finished_kst": datetime.now(KST).isoformat(timespec="seconds"),
            "status": status,
            "used": int(b.get("used", 0)), "denied": int(b.get("denied", 0)),
            # ★글과 **따로** 적는다 — 합치면 무엇에 돈이 나갔는지 못 읽는다
            "image_used": int(im.get("used", 0)),
            "image_denied": int(im.get("denied", 0)),
            "per_step": dict(out["per_step"]),
            "stopped_at": out.get("stopped_at"),
        })

    def _progress() -> None:
        """스텝마다 **장부에도** 남긴다 — 죽어도 어디까지 갔는지 안다."""
        b = out.get("budget") or {}
        im = out.get("image_budget") or {}
        update_attempt(root, attempt_id, {
            "used_so_far": int(b.get("used", 0)),
            "denied_so_far": int(b.get("denied", 0)),
            "image_used_so_far": int(im.get("used", 0)),
            "image_denied_so_far": int(im.get("denied", 0)),
            "per_step": dict(out["per_step"]),
        })

    _close = _mark
    #: 앞 판에 이미 끝난 스텝들 — 이번 판에서 **0 으로 연다**
    already: set = set()
    out["reused_from_earlier_attempt"] = already

    try:
        from tools.grounding_audit.canary_outbound_gates import (
            canary_outbound_scope, snapshot_of as _outbound_snapshot)

        with canary_text_scope(cap=scope_cap) as budget, \
                canary_image_scope(cap=image_scope_cap) as ibudget, \
                canary_outbound_scope(
                    search_cap=int(approved_search),
                    download_cap=int(approved_downloads)) as obound:
            out["outbound"] = _outbound_snapshot(obound)
            out["image_budget"] = dict(ibudget.snapshot())
            for s in steps:
                b0 = dict(budget.snapshot())
                i0 = dict(ibudget.snapshot())
                # ★★★**부르기 전에** 이 스텝 몫만 문에 건다 (Codex 2026-09-01).
                #  무료·건너뜀·목록 밖은 0 이라 provider 가 아예 안 불린다.
                step_cap = (int(caps.get(s, 0)) * _per_logical(plan)
                            if s in metered else 0)
                # ★★★**앞 판에 이미 끝난 스텝은 0 이다** (Codex BLOCK 09-02).
                #  재개가 되쓰는지를 **사후에 관찰**하면 「15 를 또 쓴 뒤」
                #  알게 된다. 무효화가 잘못 일어났으면 **첫 외부 호출 전에**
                #  서야 한다.
                durable = durable_status(project_id, episode_id, s)
                # ★적용 안 되는 스텝은 지문을 안 묻는다 — 돌지 않는 것의 어긋남은 뜻이 없다
                #  (실측: grounding_a0 가 base 해시 CP 로 drift 라 서 버렸다)
                drift = (contract_drift_of(s, project_id, episode_id, project_config)
                         if durable == "completed" else None)
                # ★끝났어도 runner 가 「빚이 남았다」면(취소·검색 실패로 못 산 대상) 정상 cap 으로 들어간다 —
                #  runner 의 RERUN_SELF 가 장부를 되쓰며 빚진 대상만 산다
                debt = (completion_debt_of(s, project_id, episode_id, project_config)
                        if durable == "completed" and not drift else None)
                decision = plan_reentry(durable, drift, code_transition_covers_head(root),
                                        s in set(reopen or ()), debt=debt)
                if debt:
                    out["per_step"].setdefault(s, {})["reopened_for_debt"] = debt
                if decision["stop_why"]:
                    out["stopped_at"] = {"step": s, "why": f"{s}: {decision['stop_why']}",
                                         "budget": dict(budget.snapshot()),
                                         "image_budget": dict(ibudget.snapshot())}
                    _write(root, out)
                    raise CanaryStopped(out["stopped_at"]["why"])
                run_mode = str(decision["mode"])
                if decision["cap_zero"]:
                    step_cap = 0
                    out["per_step"].setdefault(s, {})
                    already.add(s)
                elif run_mode == "force":
                    # ★★승인된 전이 뒤 지문이 어긋난 스텝은 force 로 다시 돈다 — 하류가 무효화되고
                    #  그 뒤 스텝은 CP 가 없어 제 차례에 새로 돈다. 저널은 그대로라 재구매는 신원이 같은
                    #  것만큼 0 이다.
                    out["per_step"].setdefault(s, {})["forced_for_drift"] = drift
                if before_step is not None and step_cap > 0:
                    # ★★★provider 앞 — CP 를 읽는 producer 상한 문. 넘으면 사지 않고 선다.
                    try:
                        pg = before_step(s)
                    except BaseException as exc:        # noqa: BLE001
                        out["stopped_at"] = {"step": s, "why": f"{s}: producer 상한 문 — {exc}",
                                             "budget": dict(budget.snapshot()),
                                             "image_budget": dict(ibudget.snapshot())}
                        _write(root, out)
                        raise
                    if pg is not None:
                        out["per_step"].setdefault(s, {})["producer_gate"] = pg
                        # ★문이 실제 producer 산출로 **더 큰** 상한을 냈으면 그 수로 문을 연다 — 서지 않고 이어간다.
                        #  작게 내는 일은 없다(문은 계획을 줄이지 않는다). 승인 정지선(run-wide counted)은 그대로.
                        _re = pg.get("recomputed_logical_cap")
                        if _re is not None and int(_re) * _per_logical(plan) > step_cap:
                            _new = int(_re) * _per_logical(plan)
                            _left = int((budget.snapshot() or {}).get("remaining") or 0)
                            # ★재산정한 상한이 run 전체의 **남은** ceiling 을 넘으면 그때만 선다 (Codex 2026-09-03 06:10) —
                            #  정지선 480 과 검색·받기 문은 우회 불가 그대로.
                            if _new > _left:
                                out["stopped_at"] = {"step": s, "why": (f"{s}: 재산정 step cap {_new} 이 남은 ceiling {_left} 을 넘는다 — "
                                                                       "사지 않고 선다(정지선은 그대로)"),
                                                     "budget": dict(budget.snapshot()),
                                                     "image_budget": dict(ibudget.snapshot())}
                                _write(root, out)
                                raise CanaryStopped(out["stopped_at"]["why"])
                            out["per_step"][s]["step_cap_recomputed"] = {"from": step_cap, "to": _new, "remaining_ceiling": _left,
                                                                          "why": pg.get("why")}
                            step_cap = _new
                gate = canary_step_cap(budget, cap=step_cap, step=s)
                sb = gate.__enter__()
                try:
                    run_steps_batch(
                        project_id=project_id, episode_id=episode_id,
                        step_ids=[s], run_mode=run_mode,
                        project_config=dict(project_config or {}),
                        opik_context={"tag": TRACE_TAG,
                                      "thread": TRACE_THREAD,
                                      "canary_run_id": run_id})
                except BaseException as exc:        # noqa: BLE001
                    gate.__exit__(type(exc), exc, None)
                    out["budget"] = dict(budget.snapshot())
                    out["image_budget"] = dict(ibudget.snapshot())
                    # ★죽은 스텝의 몫도 적는다 — 안 적으면 「안 썼다」로 읽힌다
                    out["per_step"][s] = {
                        **ct.per_step_delta(b0, dict(budget.snapshot())),
                        **image_delta(i0, dict(ibudget.snapshot())),
                        **_gate_row(sb, step_cap),
                        "★partial": "이 스텝은 끝나지 못했다"}
                    out["stopped_at"] = {
                        "step": s, "why": f"{type(exc).__name__}: {exc}",
                        "budget": out["budget"],
                        "image_budget": out["image_budget"]}
                    raise
                gate.__exit__(None, None, None)
                # ★★★**되읽어** 확인한다 — `run_steps_batch` 는 안에서
                #  실패를 잡고 `None` 을 돌려준다. 이것이 없으면 크래시가
                #  `completed` 로 닫힌다 (Codex BLOCK · 실측 09-02).
                try:
                    step_state = assert_step_finished(project_id, episode_id, s)
                except StepDidNotFinish as exc:
                    out["budget"] = dict(budget.snapshot())
                    out["image_budget"] = dict(ibudget.snapshot())
                    out["per_step"][s] = {
                        **ct.per_step_delta(b0, dict(budget.snapshot())),
                        **image_delta(i0, dict(ibudget.snapshot())),
                        **_gate_row(sb, step_cap),
                        "durable_status": durable_status(project_id,
                                                         episode_id, s),
                        "★partial": "이 스텝은 끝나지 못했다"}
                    out["stopped_at"] = {
                        "step": s, "why": f"{type(exc).__name__}: {exc}",
                        "budget": out["budget"],
                        "image_budget": out["image_budget"]}
                    raise
                b1 = dict(budget.snapshot())
                i1 = dict(ibudget.snapshot())
                out["budget"] = b1
                out["image_budget"] = i1
                delta = {**ct.per_step_delta(b0, b1), **image_delta(i0, i1),
                         **_gate_row(sb, step_cap),
                         "durable_status": step_state}
                # ★★★**되쓴 스텝이 실제로 provider 를 부르려 했으면 선다**
                #  (Codex BLOCK 2026-09-02). `cap=0` 에서 거절당한 예외를
                #  `run_steps_batch` 가 안에서 삼키면, **옛 completed
                #  manifest** 가 남아 있어 되읽기가 통과한다 — 그러면
                #  「되썼다」가 아니라 「다시 사려다 막혔다」인데 초록이 된다.
                if s in already and int(delta.get("denied_by_step_cap") or 0):
                    out["stopped_at"] = {
                        "step": s, "budget": b1, "image_budget": i1,
                        "why": (f"{s} 는 앞 판에 끝났는데 이번 판이 provider 를 "
                                f"{delta['denied_by_step_cap']}번 부르려 했다 "
                                f"— 체크포인트 재사용이 안 됐다. 되쓴 것으로 "
                                f"세지 않는다")}
                    already.discard(s)
                    raise CanaryStopped(out["stopped_at"]["why"])
                # ★문 앞에서 적은 것(producer_gate · step_cap_recomputed · forced_for_drift)을 지우지 않는다 — 덧붙인다
                out["per_step"][s] = {**(out["per_step"].get(s) or {}), **delta}
                # ★검색·받기 문은 **따로** 적는다 — 글 예산과 다른 갈래다
                out["outbound"] = _outbound_snapshot(obound)
                _write(root, out)               # ★한 스텝마다 내려쓴다
                _progress()                     # ★장부에도 남긴다
                # ★①무료로 적은 스텝이 보냈으면 분류가 틀렸다
                if s in free:
                    try:
                        if delta["image_counted"] or delta["image_denied"]:
                            raise CanaryStopped(
                                f"{s} 는 무료로 적었는데 이미지 문을 "
                                f"{delta['image_counted'] + delta['image_denied']}"
                                "번 두드렸다 — 분류가 틀렸다")
                        ct.assert_free_step_sent_nothing(s, delta)
                    except BaseException as exc:    # noqa: BLE001
                        out["stopped_at"] = {"step": s, "why": str(exc),
                                             "budget": b1}
                        raise
                # ★②상한. ★목록 밖(적용 안 되는) 스텝은 **0만**
                if s in metered:
                    try:
                        ct.assert_within(
                            observed={s: delta["counted"]},
                            approved={s: int(caps.get(s, 0))
                                      * _per_logical(plan)})
                    except BaseException as exc:    # noqa: BLE001
                        out["stopped_at"] = {"step": s, "why": str(exc),
                                             "budget": b1}
                        raise
                elif delta["counted"]:
                    out["stopped_at"] = {
                        "step": s, "budget": b1,
                        "why": (f"{s} 는 이번 판의 유료 목록에 없는데 "
                                f"{delta['counted']}번 보냈다")}
                    raise CanaryStopped(out["stopped_at"]["why"])
                # ★②-b 이미지 승인이 **0 인 판**이면 한 장이라도 나갔을 때 선다.
                #  ★실측 2026-09-02 stage2a: 승인 80 인데 이 줄이 승인값을 안 보고
                #   ref_image_gen 4장 뒤에 세웠다 — 글만 사던 판의 규칙이 남아 있었다.
                #   승인이 있으면 run 전체 이미지 문(`canary_image_scope` · 남은 것만
                #   연다)이 정지선이다 — 여기서 또 세우지 않는다.
                if delta["image_counted"] and int(image_scope_cap) <= 0:
                    out["stopped_at"] = {
                        "step": s, "budget": b1, "image_budget": i1,
                        "why": (f"{s} 가 이미지를 {delta['image_counted']}장 "
                                f"샀다 — 이번 판의 이미지 승인은 "
                                f"{approved_image_calls} 이다")}
                    raise CanaryStopped(out["stopped_at"]["why"])
                # ★③비상 정지선 — **run 전체** 기준
                if before_total + int(b1["used"]) >= int(emergency_counted):
                    out["stopped_at"] = {"step": s, "budget": b1,
                                         "why": "emergency ceiling"}
                    raise CanaryStopped(
                        f"이 run 의 누적이 정지선 {emergency_counted} 에 닿았다 "
                        f"(앞 {before_total} + 이번 {b1['used']}) — 이 판은 "
                        "**inconclusive** 로 보존한다. 자동으로 늘리거나 다시 "
                        "사지 않는다")
    except BaseException as exc:
        # ★★죽은 까닭을 **갈라 적는다** — 상한에 닿아 멈춘 것과 스텝이
        #  끝나지 못한 것은 다르다. 앞 판은 둘 다 `stopped` 였고, 크래시는
        #  아예 `completed` 로 닫혔다 (Codex BLOCK 09-02).
        out["reused_from_earlier_attempt"] = sorted(already)
        _close("crashed" if isinstance(exc, StepDidNotFinish) else "stopped")
        _write(root, out)
        raise
    out["reused_from_earlier_attempt"] = sorted(already)
    _close("completed")
    out["cumulative_after"] = cumulative_used(root)
    from tools.grounding_audit.canary_image_budget import image_doors

    out["image_doors"] = image_doors()
    _write(root, out)
    return out


def _gate_row(sb: Any, step_cap: int) -> Dict[str, int]:
    """이 스텝의 **문**이 무엇을 했나. ★막은 것을 적어야 「안 썼다」와 갈린다."""
    got = dict(sb.snapshot())
    return {"step_cap": int(step_cap),
            "denied_by_step_cap": int(got.get("denied", 0))}


def _per_logical(plan: Dict[str, Any]) -> int:
    """논리 하나가 여는 counted 수. ★공용 계약에서 온다."""
    from tools.grounding_audit.call_bound_contract import layers

    lay = layers(contract=plan.get("contract") or
                 {"num_retries": 0, "enable_fallback": False},
                 slots=ct.slots_for("gpt")["slots"])
    return int(lay["per_logical_counted"] or 1)


def _write(root: Path, rec: Dict[str, Any]) -> None:
    root.mkdir(parents=True, exist_ok=True)
    (root / "pipeline_run.json").write_text(
        json.dumps(rec, ensure_ascii=False, indent=1, default=str),
        encoding="utf-8")


# ─────────────────────────────────────────────────────────────────────
# ★★★재개마다 상한이 **처음부터 다시** 열리던 것 (Codex BLOCK 2026-09-01)
#
# > `run_pipeline` 은 재개할 때마다 `canary_text_scope(cap=92)` 를 새로 엽니다.
# > 그래서 같은 run 이 재개될 때마다 다시 92 를 쓸 수 있습니다. 그리고
# > `pipeline_run.json` 이 덮어써져 **첫 attempt 의 per-step 기록이 이미
# > 사라졌습니다.**
#
# 그래서 —
#     ①attempt 를 **덧붙이기만** 한다 (덮지 않는다)
#     ②새 scope 상한 = **정지선 − 누적 사용**
#     ③장부가 없거나 깨졌거나 「모른다」면 **provider 0 으로 선다**
# ─────────────────────────────────────────────────────────────────────

ATTEMPTS = "pipeline_attempts.json"


class LedgerRefused(RuntimeError):
    """누적 장부를 못 믿는다. ★아무것도 안 사고 선다."""


def _ledger_path(root: Path) -> Path:
    return Path(root) / ATTEMPTS


def read_attempts(root: Path) -> List[Dict[str, Any]]:
    """지난 attempt 전부. ★못 읽으면 **0으로 안 읽는다** — 선다."""
    p = _ledger_path(root)
    if not p.is_file():
        return []
    try:
        got = json.loads(p.read_text(encoding="utf-8"))
    except Exception as exc:                        # noqa: BLE001
        raise LedgerRefused(
            f"누적 장부를 못 읽었다 ({type(exc).__name__}) — 얼마나 썼는지 "
            "모르는 채로 살 수 없다") from exc
    if not isinstance(got, list):
        raise LedgerRefused("누적 장부가 목록이 아니다 — 못 믿는다")
    return got


def cumulative_used(root: Path) -> int:
    """이 run 이 지금까지 **정말** 쓴 수. ★모르는 것이 있으면 **선다**.

    ★열린(`running`/`uncertain`) attempt 를 **0 으로 세지 않는다** — 그것이
    바로 「샀는데 안 적힌」 자리다 (Codex 2026-09-01).
    """
    total = 0
    for a in read_attempts(root):
        # ★정정·전이 줄은 **attempt 가 아니다** — 여기서 세면 두 번 센다
        if a.get("kind") is not None:
            continue
        st = str(a.get("status"))
        if st in OPEN_STATES:
            raise LedgerRefused(
                f"attempt {a.get('attempt_id')} 이 **아직 열려 있다**({st}) — "
                "그 판이 얼마 썼는지 모른다. 0 으로 세지 않는다. "
                "Opik·provider 로그를 보고 사람이 닫아야 한다")
        u = effective_used(root, str(a.get("attempt_id") or ""))
        if u is None:
            raise LedgerRefused(
                f"attempt {a.get('attempt_id')} 의 사용량이 **없다** — "
                "0 으로 발명하지 않는다. 사람이 정해야 한다")
        total += int(u)
    return total


def remaining_cap(root: Path, *, ceiling: int) -> int:
    """이번 attempt 가 열 수 있는 상한. ★정지선은 **run 전체**에 걸린다."""
    used = cumulative_used(root)
    left = int(ceiling) - used
    if left <= 0:
        raise LedgerRefused(
            f"이 run 이 이미 {used} 를 썼다 (정지선 {ceiling}) — 남은 것이 "
            "없다. 자동으로 늘리지 않는다")
    return left


def append_attempt(root: Path, rec: Dict[str, Any]) -> None:
    """attempt 를 **덧붙인다**. ★앞의 것을 절대 안 덮는다."""
    got = read_attempts(root)
    got.append(rec)
    p = _ledger_path(root)
    p.parent.mkdir(parents=True, exist_ok=True)
    tmp = p.with_suffix(f".{uuid.uuid4().hex[:8]}.tmp")
    tmp.write_text(json.dumps(got, ensure_ascii=False, indent=1, default=str),
                   encoding="utf-8")
    os.replace(tmp, p)


#: 아직 안 끝난 attempt. ★이것이 남아 있으면 **얼마 썼는지 모른다**.
OPEN_STATES = ("running", "uncertain")


def _write_attempts(root: Path, rows: List[Dict[str, Any]]) -> None:
    p = _ledger_path(root)
    p.parent.mkdir(parents=True, exist_ok=True)
    tmp = p.with_suffix(f".{uuid.uuid4().hex[:8]}.tmp")
    tmp.write_text(json.dumps(rows, ensure_ascii=False, indent=1, default=str),
                   encoding="utf-8")
    os.replace(tmp, p)                      # ★원자적으로


#: 정정·전이는 **덧붙이는 줄**이다. ★앞 줄을 고쳐 쓰지 않는다.
EVENT_CORRECTION = "correction"
EVENT_CODE_TRANSITION = "code_transition"
EVENT_HASH_ADOPTION = "config_hash_adoption"     # ★canary_hash_adoption.adopt — data 불변 · config_hash 메타데이터만


def append_event(root: Path, rec: Dict[str, Any]) -> Dict[str, Any]:
    """장부에 **사건 한 줄**을 덧붙인다. ★앞의 것을 절대 안 덮는다.

    ★★★앞 판의 판정이 틀렸을 때 그 줄을 **고쳐 쓰면 증거가 사라진다** —
    「completed 였다가 crashed 로 바뀌었다」는 사실 자체가 증거다
    (Codex BLOCK 2026-09-02).
    """
    # ★★`recorded_kst` 를 열쇠로 쓰면 **같은 초에 적힌 둘이 안 갈린다**
    #  (실측 2026-09-02: 대신하는 줄이 자기까지 죽였다). 고유 id 를 준다.
    row = {"kind": str(rec.get("kind") or ""),
           "event_id": uuid.uuid4().hex[:12],
           "recorded_kst": datetime.now(KST).isoformat(timespec="seconds"),
           **rec}
    if row["kind"] not in (EVENT_CORRECTION, EVENT_CODE_TRANSITION, EVENT_HASH_ADOPTION):
        raise CanaryStopped(f"모르는 사건 {row['kind']!r}")
    got = read_attempts(root)
    got.append(row)
    _write_attempts(root, got)
    return row


def correct_attempt(root: Path, attempt_id: str, *, was: str, now: str,
                    why: str, used: Optional[int] = None) -> Dict[str, Any]:
    """앞 판의 판정을 **덧붙여** 바로잡는다. ★그 줄은 그대로 둔다.

    Raises:
        CanaryStopped: 그 attempt 가 없거나 지금 상태가 `was` 와 다르다.
    """
    rows = read_attempts(root)
    hit = [a for a in rows if a.get("attempt_id") == attempt_id
           and a.get("kind") is None]
    if not hit:
        raise CanaryStopped(f"attempt {attempt_id} 가 장부에 없다")
    got = str(hit[-1].get("status"))
    if got != was:
        raise CanaryStopped(
            f"attempt {attempt_id} 의 상태가 {got!r} 다 — {was!r} 로 알고 "
            f"정정하려 했다. 무엇을 고치는지 모르는 채 안 고친다")
    return append_event(root, {
        "kind": EVENT_CORRECTION, "attempt_id": attempt_id,
        "status_was": was, "status_now": now, "why": why,
        "used": int(hit[-1].get("used", 0) if used is None else used),
        "★means": ("앞 줄은 **그대로 둔다** — 판정이 바뀌었다는 사실이 증거다"),
    })


#: 운반 기록 파일. ★줄마다 `attempt_id` 를 결속한다.
TRANSPORT_LOG = "transport_used.jsonl"


def read_transport(root: Path) -> List[Dict[str, Any]]:
    """운반 기록 줄들. ★못 읽는 줄이 있으면 **선다**."""
    p = root / TRANSPORT_LOG
    if not p.is_file():
        return []
    out = []
    for ln in p.read_text(encoding="utf-8").splitlines():
        if not ln.strip():
            continue
        try:
            out.append(json.loads(ln))
        except Exception as exc:                    # noqa: BLE001
            raise LedgerRefused(
                f"운반 기록에 못 읽는 줄이 있다 ({exc}) — 얼마 보냈는지 "
                "모르는 채로 살 수 없다") from exc
    return out


def unsettled_transport(root: Path, scope: str) -> Dict[str, Any]:
    """**아직 장부에 안 접힌** 운반. ★두 번 차감하지 않으려고 가른다.

    ★★★실측 지적 (Codex 2026-09-02): 앞 판은 그 run 의 **과거 최대 used** 를
    그대로 이어받았다. 그런데 그 사용량은 이미 `effective_used` 로 누계에
    접혀 `remaining = 상한 − 누계` 에 반영돼 있다 — 다시 이어받으면 **같은
    것을 두 번 뺀다**.

    그래서 **terminal 로 닫힌 attempt 의 줄은 안 이어받는다**. 아직 안 닫힌
    (열려 있거나 장부에 아예 없는) attempt 의 줄만 화해한다 — 그것이
    「샀는데 아직 안 적힌」 몫이다.

    Returns:
        `{"used": n, "by_source": {...}, "from_attempts": [...]}`
    """
    settled = {str(a.get("attempt_id")) for a in read_attempts(root)
               if a.get("kind") is None
               and str(a.get("status")) not in OPEN_STATES}
    used, by, seen = 0, {}, []
    for r in read_transport(root):
        if str(r.get("scope")) != scope:
            continue
        aid = str(r.get("attempt_id") or "")
        if aid and aid in settled:
            continue                    # ★이미 누계에 접혔다 — 또 안 뺀다
        if int(r.get("used", 0)) >= used:
            used = int(r.get("used", 0))
            by = dict(r.get("by_source") or {})
            seen = [aid or "미정"]
    return {"used": used, "by_source": by, "from_attempts": seen}


def effective_used(root: Path, attempt_id: str) -> Optional[int]:
    """정정을 **반영한** 사용량. ★누계·남은 상한이 이것을 쓴다.

    ★★★앞 판은 정정을 「메모」로만 뒀다 — `cumulative_used` 가 `kind` 있는
    줄을 전부 건너뛰므로 **누계가 안 움직였다** (Codex 2026-09-02).
    정정 줄에 `used_now` 가 있으면 그것이 정본이다.
    ★가짜 attempt 를 더해서 수를 맞추지 않는다 — 그러면 판 수가 거짓이 된다.
    """
    got = None
    for a in read_attempts(root):
        if a.get("attempt_id") != attempt_id:
            continue
        if a.get("kind") is None:
            if a.get("used") is not None:
                got = int(a["used"])
        elif a.get("kind") == EVENT_CORRECTION:
            # ★승인선에서 빼는 수는 **`budget_debit_now`** 다
            if a.get("budget_debit_now") is not None:
                got = int(a["budget_debit_now"])
    return got


def correct_usage(root: Path, attempt_id: str, *, debit_was: int,
                  debit_now: int, why: str,
                  logical_calls: Optional[int] = None,
                  physical_exact: Optional[int] = None,
                  physical_lower: Optional[int] = None,
                  physical_upper: Optional[int] = None,
                  evidence: Optional[Dict[str, Any]] = None
                  ) -> Dict[str, Any]:
    """사용량을 **덧붙여** 바로잡는다. ★앞 줄은 그대로 둔다.

    ★★★세 수를 **갈라 적는다** (Codex 2026-09-02) —

        logical_calls      구매 수. 장부가 세는 것
        physical_attempts  실제 운반 시도. **모르면 `exact=None` 에 범위**
        budget_debit       승인선에서 **빼는 수**. 모르면 상한을 뺀다

    ★모르는 것을 「확정」으로 적지 않는다. 안전을 위해 상한을 빼되, 기록은
    「2~4 · 정확값 미확정」으로 남는다.
    """
    now = effective_used(root, attempt_id)
    if now is None:
        raise CanaryStopped(f"attempt {attempt_id} 의 사용량을 못 읽었다")
    if int(now) != int(debit_was):
        raise CanaryStopped(
            f"attempt {attempt_id} 의 지금 차감이 {now} 다 — {debit_was} 로 "
            f"알고 고치려 했다. 무엇을 고치는지 모르는 채 안 고친다")
    if physical_exact is None and (physical_lower is None
                                   or physical_upper is None):
        raise CanaryStopped(
            "실제 운반 수를 모르면 **범위**라도 적어야 한다 "
            "(`physical_lower`/`physical_upper`)")
    upper = (int(physical_exact) if physical_exact is not None
             else int(physical_upper))
    if int(debit_now) < upper:
        raise CanaryStopped(
            f"차감 {debit_now} 이 알려진 상한 {upper} 보다 작다 — "
            f"모르는 만큼은 **크게** 빼야 넘치지 않는다")
    return append_event(root, {
        "kind": EVENT_CORRECTION, "attempt_id": attempt_id,
        "logical_calls": logical_calls,
        "physical_attempts": {"exact": physical_exact,
                              "lower": physical_lower,
                              "upper": physical_upper},
        "budget_debit_was": int(debit_was), "budget_debit_now": int(debit_now),
        "why": why, "evidence": evidence or {},
        "★means": ("`budget_debit_now` 는 **승인선에서 빼는 수**다 — 실제 "
                   "운반 수가 아니다. 실제는 `physical_attempts` 이고 "
                   "`exact` 가 `None` 이면 **모른다**는 뜻이다"),
    })


def effective_status(root: Path, attempt_id: str) -> Optional[str]:
    """정정을 **반영한** 상태. ★보고서는 이것을 읽는다."""
    st = None
    for a in read_attempts(root):
        if a.get("attempt_id") != attempt_id:
            continue
        if a.get("kind") == EVENT_CORRECTION:
            # ★★사용량 정정은 **상태를 안 건드린다** — 앞 판은 `status_now`
            #  가 없어도 덮어써서 `crashed` 가 문자열 `"None"` 이 됐다
            #  (실측 2026-09-02).
            if a.get("status_now") is not None:
                st = str(a["status_now"])
        elif a.get("kind") is None:
            st = str(a.get("status")) if a.get("status") else st
    return st


def effective_image_used(root: Path, attempt_id: str) -> Optional[int]:
    """정정을 반영한 **이미지** 사용량. ★글과 따로 — 합치면 무엇에 돈이 나갔는지 못 읽는다.
    attempt 줄의 `image_used`, 정정 줄의 `image_debit_now` 가 정본이다."""
    got = None
    for a in read_attempts(root):
        if a.get("attempt_id") != attempt_id:
            continue
        if a.get("kind") is None:
            if a.get("image_used") is not None:
                got = int(a["image_used"])
        elif a.get("kind") == EVENT_CORRECTION:
            if a.get("image_debit_now") is not None:
                got = int(a["image_debit_now"])
    return got


def cumulative_image_used(root: Path) -> int:
    """이 run 이 지금까지 **정말 산 이미지** 수. ★열린 attempt 가 있으면 선다(글과 같다).

    ★★★Codex BLOCK (2026-09-02): 글은 `remaining_cap` 으로 누계를 빼고 열었는데
    이미지는 attempt 마다 `canary_image_scope(cap=승인값)` 을 **통째로 새로** 열었다 —
    20장 뒤 crash→재개면 40장을 더 열어 승인 40 을 넘는다. 이제 이미지도 run 전체다.
    ★옛 attempt 줄에 `image_used` 가 없으면(이미지 문 전의 판) 0 으로 본다 — 그 판들은
    이미지 승인 0 으로 돌았고 문이 닫혀 있었다.
    """
    total = 0
    for a in read_attempts(root):
        if a.get("kind") is not None:
            continue
        st = str(a.get("status"))
        if st in OPEN_STATES:
            raise LedgerRefused(
                f"attempt {a.get('attempt_id')} 이 **아직 열려 있다**({st}) — "
                "그 판이 이미지를 얼마 샀는지 모른다. 0 으로 세지 않는다")
        u = effective_image_used(root, str(a.get("attempt_id") or ""))
        total += int(u or 0)
    return total


def remaining_image_cap(root: Path, *, ceiling: int) -> int:
    """이번 attempt 가 열 수 있는 **이미지** 상한 — run 전체 정지선에서 누계를 뺀다.
    ★승인 0 이면 0 이다(글만 사는 판 — 문은 닫힌 채 선다). 누계가 승인을 넘어 있으면
    선다(그 run 은 이미 거짓이다). 승인이 있는데 남은 것이 0 이면 선다 — 자동 확대 0."""
    used = cumulative_image_used(root)
    ceiling = int(ceiling)
    if used > ceiling:
        raise LedgerRefused(
            f"이 run 이 이미지를 {used} 장 샀는데 승인은 {ceiling} 이다 — 장부가 승인과 "
            "어긋난다. 사람이 봐야 한다")
    if ceiling == 0:
        return 0
    left = ceiling - used
    if left <= 0:
        raise LedgerRefused(
            f"이 run 이 이미지를 이미 {used} 장 샀다 (승인 {ceiling}) — 남은 것이 없다. "
            "자동으로 늘리지 않는다")
    return left


def open_attempts(root: Path) -> List[Dict[str, Any]]:
    """아직 안 닫힌 attempt. ★있으면 다음 판은 **provider 0 으로** 선다."""
    return [a for a in read_attempts(root)
            if a.get("kind") is None
            and str(a.get("status")) in OPEN_STATES]


def reserve_attempt(root: Path, rec: Dict[str, Any]) -> None:
    """**사기 전에** 자리를 durable 하게 잡는다.

    ★★★앞 판은 attempt 를 **끝나고 나서** 적었다. 그래서 provider 를 부른 뒤
    프로세스가 `SIGKILL`·`os._exit` 로 사라지면 **장부에 한 줄도 안 남고**,
    다음 재개가 그만큼을 **덜 세어** 상한을 넘길 수 있었다
    (Codex BLOCK 2026-09-01).

    ★열린 attempt 가 이미 있으면 **아무것도 안 하고 선다** — 동시에 둘이
    열리면 누가 얼마 썼는지 못 가른다.
    """
    got = read_attempts(root)
    live = [a for a in got if str(a.get("status")) in OPEN_STATES]
    if live:
        raise LedgerRefused(
            f"아직 안 닫힌 attempt 가 있다: {[a.get('attempt_id') for a in live]} "
            "— 얼마 썼는지 모르는 채로 또 살 수 없다. 사람이 정해야 한다")
    got.append({**rec, "status": "running", "used": None, "denied": None})
    _write_attempts(root, got)


def update_attempt(root: Path, attempt_id: str, patch: Dict[str, Any]) -> None:
    """열린 attempt 를 **제자리에서** 갱신한다. ★앞의 것들은 그대로 둔다."""
    got = read_attempts(root)
    for i, a in enumerate(got):
        if a.get("attempt_id") == attempt_id:
            got[i] = {**a, **patch}
            _write_attempts(root, got)
            return
    raise LedgerRefused(f"attempt {attempt_id} 이 장부에 없다 — 못 닫는다")