"""canary 주행 동안 **모든 스레드**의 이미지 호출을 세는 문.

★왜 따로 있나 — 글 예산(`canary_text_scope`)은 **글 문**만 본다.
`ResearchCallBudget` 은 `llm_client._completion` 과 `openai_keys` 세 자리에
걸려 있고, 유료 **이미지** 호출은 그 셋을 하나도 안 지난다. 그래서 글 예산만
깔고 돌면 이미지는 **한 푼도 안 세어진 채** 나간다.

    실측 2026-09-01 — `scene_detail` 까지의 closure 안에 `floor_plan_render`
    가 있고, 그것은 `app/modules/pipeline/floor_plan_render.py` 의 세 자리
    (`edit_multi`·`edit_single`·`generate`)에서 `gpt-image-2` 를 산다.
    글 장부에는 **0 으로 적히고** 실제로는 돈이 나간다.

`app.core.image_call_budget` 의 문(`reserve_current_call`)은 **예산이 안
깔려 있으면 그냥 지나간다**(no-op). 즉 「문이 있다」와 「문이 잠겼다」는 다르다.

## 무엇을 하나

주행 동안만 그 모듈의 스레드 지역 저장소를 **모든 스레드가 같이 보는 것**으로
바꾼다 — `canary_text_scope` 와 같은 수법이다. production 코드는 한 글자도
안 바꾼다.

★`run_with_budget` 은 「이전 것」을 되돌리는데, 같이 보는 저장소에서는 그
이전 것이 **바로 이 예산**이라 되돌려도 그대로다. 그리고 `bind_current_budget`
을 안 부르는 팬아웃도 이 예산을 본다 — 문이 팬아웃마다 뚫리지 않는다.
"""
from __future__ import annotations

import contextlib
from pathlib import Path
from typing import Any, Dict, Iterator

#: ★`cwd` 가 아니라 **모듈 자리**에서 뽑는다 — 어디서 부르든 같은 답이다.
BACKEND = Path(__file__).resolve().parents[2]


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

    budget = None
    stop_check = None


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

    Args:
        cap: 승인된 이미지 호출 수. ★**0 이면 한 장도 안 산다** — 문 앞에서
            `ImageCallBudgetExceeded` 로 선다. 그러면 그 스텝이 실패하고
            장부에 「여기서 이미지를 사려 했다」가 남는다.

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

    ★겹쳐 깔면 남의 수를 먹으므로 **선다**. 서버 프로세스 안에서 쓰지 않는다.
    """
    import threading

    from app.core import image_call_budget as ib

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

    budget = ib.ImageCallBudget(cap=int(cap))
    shared = _Shared()
    shared.budget = budget
    prev = ib._local
    ib._local = shared               # ★주행 동안만
    try:
        yield budget
    finally:
        ib._local = prev


def image_delta(before: Dict[str, Any], after: Dict[str, Any]) -> Dict[str, int]:
    """스텝 하나가 **실제로 산 이미지** 수. ★막힌 것도 같이 적는다."""
    return {
        "image_counted": int(after.get("used", 0)) - int(before.get("used", 0)),
        "image_denied": int(after.get("denied", 0)) - int(before.get("denied", 0)),
    }


def image_doors() -> Dict[str, Any]:
    """★유료 이미지가 나가는 자리. 「글만 셌다」고 말하지 않기 위해 같이 낸다."""
    return {
        "gate": "app/core/image_call_budget.py:reserve_current_call",
        "★no_budget_means_no_gate": (
            "예산이 안 깔려 있으면 이 문은 **그냥 지나간다**(no-op) — "
            "「문이 있다」와 「문이 잠겼다」는 다르다"),
        # ★길과 설명을 **한 글자에 담지 않는다** — 담으면 길로 못 쓴다
        "in_this_closure": {
            "floor_plan_render": {
                "file": "app/modules/pipeline/floor_plan_render.py",
                "sites": ["edit_multi", "edit_single", "generate"],
            },
        },
        "not_in_this_closure": [
            "outdoor_place_canon", "marker_map_engine", "location_aerial",
            "background_render", "background_chain_render", "reve_image_client",
            "gemini_image_client", "grok_image_client", "fal_angle_helpers",
        ],
        "★scene_detail": ("`bind_current_budget` 을 부르지만 **정지 표를 워커로 "
                          "나르려는 것**이다 — 이미지를 사지 않는다"),
    }


def resolve(rel: str) -> Path:
    """적어 둔 상대 경로를 **모듈 자리 기준**으로 푼다."""
    return BACKEND / rel