"""검색·다운로드 **문**. ★preflight 에 적는 것은 문이 아니다.

Codex (2026-09-02):

> preflight에 적고 사후 journal과 대조하는 것만으로는 문이 아닙니다.
> 검색은 `client.responses.create` 직전에 10회 문을 걸어 11번째를
> **네트워크 전에** 거절하십시오. 이 경로는 `ResearchCallBudget` 을 안
> 지나므로 93이 막지 못합니다.

## 왜 글 예산이 못 막나

글/VLM 예산(`ResearchCallBudget`)은 `llm_client._completion` 과 키 슬롯에
걸려 있다. 그런데 —

    검색   `search_grounded_ref.search_reference_images` 가 OpenAI
           `client.responses.create` 를 **직접** 부른다. litellm 을 안 지난다
    받기   `download_candidate` 가 `_fetch_safe` 로 **HTTP 를 직접** 친다

둘 다 그 예산 **밖**이다. 그래서 문을 따로 세운다.

## 세는 단위가 서로 다르다

    검색 **요청**        대상 × 라운드            — 승인 10
    받기 **operation**   요청 × 라운드당 후보     — 승인 40
    raw source fetch    operation 당 **최대 2**  — 원본 URL + fallback URL

★★`download_candidate` 하나가 원본과 fallback 을 **차례로** 친다. 그래서
40 은 **operation 수**이지 raw HTTP 40 이 아니다. 실제 fetch 는 최대 80 이고
**redirect 는 안 센다** — 그래서 「total raw HTTP 상한」이라고 쓰지 않는다
(Codex 2026-09-02).
"""
from __future__ import annotations

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


class OutboundDenied(RuntimeError):
    """승인 밖이다. ★네트워크 **전에** 선다."""


class OutboundBudget:
    """한 갈래의 문. ★쓴 것과 **막은 것**을 따로 센다."""

    def __init__(self, name: str, cap: int) -> None:
        if isinstance(cap, bool) or not isinstance(cap, int) or cap < 0:
            raise OutboundDenied(
                f"{name} 상한이 0 이상의 정확한 정수가 아니다 — {cap!r}")
        self.name, self.cap, self.used, self.denied = name, cap, 0, 0
        # ★대상 병렬 (2026-09-03 · Codex 계약 4): check→increment 가 **한 임계구역**이어야 한다.
        #  plain int 로 두면 워커 둘이 같은 자리를 잡아 상한을 넘긴다.
        self._lock = threading.RLock()

    def reserve(self, *, where: str = "") -> None:
        """자리를 **먼저** 잡는다. ★넘으면 보내기 전에 선다. ★원자적."""
        with self._lock:
            if self.used >= self.cap:
                self.denied += 1
                raise OutboundDenied(
                    f"{self.name} 승인 {self.cap} 을 넘었다 (쓴 것 {self.used}) — "
                    f"거절한 자리 {where!r}. 네트워크 전에 선다")
            self.used += 1

    def snapshot(self) -> Dict[str, Any]:
        with self._lock:
            return {"cap": self.cap, "used": self.used, "denied": self.denied,
                    "remaining": max(0, self.cap - self.used)}


class RawFetchObserver:
    """raw source fetch 를 **센다**. ★막지는 않는다 — 관측이다.

    ★redirect 는 이 자리에서 안 보인다. 그래서 이 수를 「total raw HTTP」로
    쓰지 않는다.
    """

    def __init__(self) -> None:
        self.count = 0
        self._lock = threading.RLock()

    def bump(self) -> None:
        """★원자적 증가 — 워커 여럿이 같이 센다."""
        with self._lock:
            self.count += 1

    def snapshot(self) -> Dict[str, Any]:
        with self._lock:
            n = self.count
        return {"source_fetch_attempts": n,
                "★means": ("`download_candidate` 안에서 원본·fallback URL 을 "
                           "친 횟수다. **redirect 는 안 센다** — 그래서 "
                           "total raw HTTP 상한이라고 쓰지 않는다")}


@contextlib.contextmanager
def canary_outbound_scope(*, search_cap: int, download_cap: int
                          ) -> Iterator[Dict[str, Any]]:
    """검색·받기 문을 **production 경로에** 건다. 나올 때 되돌린다.

    Yields:
        `{"search": OutboundBudget, "download": OutboundBudget,
          "raw": RawFetchObserver}`
    """
    from app.modules.pipeline import search_grounded_ref as sg

    search = OutboundBudget("검색 요청", search_cap)
    download = OutboundBudget("후보 다운로드", download_cap)
    raw = RawFetchObserver()

    real_search = sg.search_reference_images
    real_download = sg.download_candidate
    real_fetch = sg._fetch_safe

    def _gated_search(*a, **k):
        # ★`responses.create` 는 이 함수 **안**에 하나뿐이다 — 들어오기 전에
        #  자리를 잡으면 그것이 곧 네트워크 앞이다.
        search.reserve(where="search_reference_images")
        return real_search(*a, **k)

    def _gated_download(url, dest, fallback_url: str = "", *a, **k):
        download.reserve(where="download_candidate")
        return real_download(url, dest, fallback_url, *a, **k)

    def _counted_fetch(u, *a, **k):
        raw.bump()
        return real_fetch(u, *a, **k)

    sg.search_reference_images = _gated_search
    sg.download_candidate = _gated_download
    sg._fetch_safe = _counted_fetch
    try:
        yield {"search": search, "download": download, "raw": raw}
    finally:
        sg.search_reference_images = real_search
        sg.download_candidate = real_download
        sg._fetch_safe = real_fetch


def snapshot_of(scope: Dict[str, Any]) -> Dict[str, Any]:
    """durable 기록에 넣을 모양. ★셋을 **갈라** 적는다."""
    return {"search_requests": scope["search"].snapshot(),
            "download_operations": scope["download"].snapshot(),
            "raw_source_fetches": scope["raw"].snapshot()}


def assert_within(scope: Dict[str, Any]) -> Dict[str, Any]:
    """승인 안이었나. ★막은 것이 하나라도 있으면 선다."""
    got = snapshot_of(scope)
    bad = [k for k in ("search_requests", "download_operations")
           if int(got[k]["denied"])]
    if bad:
        raise OutboundDenied(f"{bad} 에서 거절이 났다 — {got}")
    return got