"""canary 주행 동안 **모든 스레드**의 글 호출을 세는 문. ★새 예산을 안 만든다.

Codex BLOCK 1 (2026-08-31) —

> 새 `TextCallBudget` 을 만들지 마십시오. **기존 `ResearchCallBudget` 한 벌**을
> canary 의 전체 text 실행 범위에 쓰십시오. 다만 thread-local 이고
> `bind_current_research_budget` 은 budget/stop 만 옮기며 **armed 상태는 안
> 옮깁니다.** 일반 fan-out worker 에서 reserve 가 no-op 이 되는 구멍을 닫아야
> 합니다.

## 왜 이 자리인가 — 재 본 것

    llm_client._completion:438      router.completion 직전   ← 문 있다
    openai_keys._invoke:394         OpenAI SDK 호출 직전     ← 문 있다
    openai_keys.llm_completion:472+ litellm.completion 직접  ← ★문 없다

앞의 둘은 **이미 물리 전송 자리**에 걸려 있다. 그런데 `is_armed()` 도
`get_current_budget()` 도 `threading.local` 이라, 팬아웃 워커에서는 둘 다
비어 **그냥 지나간다.**

v2 경로는 워커 **안에서** `research_calls_armed()` 를 다시 들어 괜찮지만,
일반 팬아웃 여덟(`scene_summary`·`beat_extract`·`shot_extract`·
`shot_validator`·`entity_t2i`·`shot_selection`·`shot_dependency`·
`scene_detail`)은 그 함수를 **아예 안 부른다**.

## 무엇을 하나

주행 동안만 그 모듈의 스레드 지역 저장소를 **모든 스레드가 같이 보는 것**으로
바꾼다. production 코드는 **한 글자도 안 바꾼다** — 여덟 파일을 고치는 대신
문 하나를 넓힌다.

★끝나면 되돌린다. ★`ResearchCallBudget` 은 제 잠금을 갖고 있어 여러 스레드가
같이 세도 된다.
"""
from __future__ import annotations

import contextlib
from typing import Any, Dict, Iterator, Optional


class _Shared:
    """모든 스레드가 **같이 보는** 저장소. ★`threading.local` 대신 쓴다."""

    armed = 0
    budget = None
    stop_check = None


@contextlib.contextmanager
def canary_text_scope(*, cap: int) -> Iterator[Any]:
    """이 안의 **모든 스레드**에서 글 호출이 세어지고 상한에 걸린다.

    Args:
        cap: 승인된 **counted** 상한. ★`cap` 번째까지만 나가고 `cap+1` 번째는
            **provider 앞에서** 선다. raw HTTP 는 그 아래에서 더 돌 수 있다.

    Yields:
        `ResearchCallBudget` — 끝나고 `snapshot()` 으로 실제 사용량을 본다.

    ★★★이것은 **모듈 전역을 잠시 바꾼다.** 그래서 —

        ①이미 누가 예산을 깔아 뒀으면 **선다** (겹쳐 쓰면 남의 수를 먹는다)
        ②이미 이 범위 안이면 **선다** (중첩하면 되돌릴 자리를 잃는다)
        ③서버 프로세스 안에서 쓰지 않는다 — **단독 canary 주행 전용**이다

    ★production 코드는 한 글자도 안 바꾼다. 팬아웃 여덟 파일을 고치는 대신
    문 하나를 넓히는 것이고, 범위를 벗어나면 **원래대로** 돌아온다.
    """
    import threading

    from app.core import research_call_budget as rb

    if not isinstance(rb._local, threading.local):
        raise RuntimeError(
            "이미 canary 범위 안이다 — 겹쳐 쓰면 되돌릴 자리를 잃는다")
    if rb.get_current_budget() is not None or rb.is_armed():
        raise RuntimeError(
            "누가 이미 조사 예산을 깔아 뒀다 — 겹쳐 쓰면 남의 수를 먹는다")

    budget = rb.ResearchCallBudget(cap=cap)
    shared = _Shared()
    shared.armed = 1                 # ★워커에서도 팔이 들려 있다
    shared.budget = budget
    prev = rb._local
    rb._local = shared               # ★주행 동안만
    try:
        yield budget
    finally:
        rb._local = prev


class StepScopedBudget:
    """run 전체와 **이 스텝**의 상한을 **동시에** 건다.

    ★★★왜 필요한가 (Codex BLOCK 2026-09-01) — 앞 판은 스텝이 **끝난 뒤**
    delta 를 보고 섰다. 그건 문이 아니라 **사후 경보기**다. 실제로 목록 밖
    스텝(`visual_continuity_anchor`)이 provider 를 부르고 **나서야** 잡혔다.
    분류를 아무리 고쳐도, 다음 실수 때 또 승인 밖 호출이 나간다.

    ★셈의 순서 — **스텝 상한을 먼저** 본다. 자리가 없으면 run 예산은 아예
    안 건드린다(장부가 안 나간 것을 「썼다」로 적으면 안 된다). run 예산이
    거절하면 그쪽 `denied` 만 오르고 이 스텝의 `used` 는 안 오른다.
    """

    __slots__ = ("_run", "_cap", "_used", "_denied", "_lock", "_step")

    def __init__(self, run: Any, *, cap: int, step: str) -> None:
        import threading

        if int(cap) < 0:
            raise ValueError(f"상한은 0 이상이어야 한다: {cap}")
        self._run = run
        self._cap = int(cap)
        self._step = str(step)
        self._used = 0
        self._denied = 0
        self._lock = threading.RLock()

    def reserve(self, *, source: str) -> None:
        from app.core.research_call_budget import ResearchCallBudgetExceeded

        with self._lock:
            if self._used >= self._cap:
                self._denied += 1
                raise ResearchCallBudgetExceeded(
                    cap=self._cap, used=self._used,
                    source=f"{source} · 스텝 {self._step} 의 상한")
            self._run.reserve(source=source)     # ★거절되면 아래로 안 간다
            self._used += 1

    def snapshot(self) -> Dict[str, int]:
        with self._lock:
            return {"cap": self._cap, "used": self._used,
                    "denied": self._denied,
                    "remaining": self._cap - self._used}


@contextlib.contextmanager
def canary_step_cap(run_budget: Any, *, cap: int, step: str) -> Iterator[Any]:
    """이 스텝 동안만 **좁은 상한**을 문에 건다. ★끝나면 되돌린다.

    ★무료·건너뜀·목록 밖 스텝은 `cap=0` 으로 부른다 — 그러면 provider 는
    **한 번도 안 불린다**.
    """
    import threading

    from app.core import research_call_budget as rb

    if isinstance(rb._local, threading.local):
        raise RuntimeError("`canary_text_scope` 안에서만 쓴다")
    narrow = StepScopedBudget(run_budget, cap=cap, step=step)
    prev = rb._local.budget
    rb._local.budget = narrow
    try:
        yield narrow
    finally:
        rb._local.budget = prev


def snapshot_of(budget: Any) -> Dict[str, Any]:
    """실제 사용량. ★「막았다」가 아니라 **몇 번 나갔나**를 적는다."""
    got = dict(budget.snapshot())
    got["★means"] = ("이 수는 **counted transmission** 이다 — SDK 안쪽 재시도는 "
                     "이 아래에서 더 돌 수 있다. raw 상한은 따로 센다")
    return got


def uncovered_paths() -> Dict[str, Any]:
    """★글 문이 **어디까지** 덮는지. 「다 막았다」고 말하지 않기 위해 낸다.

    ★2026-09-01 고침 — 이 함수는 `openai_keys.llm_completion` 을 「문 없음」
    으로 적고 있었는데, 그 자리에는 **이미 문을 달았다**(480·491·496).
    글로 적은 것과 코드가 갈라져서, 이 글만 읽으면 「그 스텝은 미측정이니
    안 돌린다」는 틀린 판단을 하게 된다.
    """
    return {
        "text_doors": ["app/modules/llm/llm_client.py:438",
                       "app/core/openai_keys.py:394",
                       "app/core/openai_keys.py:480",
                       "app/core/openai_keys.py:491",
                       "app/core/openai_keys.py:496"],
        "★covers": "글 호출만. **유료 이미지는 이 문을 하나도 안 지난다**",
        "image_gate": ("`app.core.image_call_budget` — canary 는 "
                       "`canary_image_scope` 로 따로 잠근다"),
    }