"""GROUNDING-V2 §8.5 — **조사 호출**의 누계 상한. ★★`ImageCallBudget` 과 **같은 꼴의 다른 owner** 다 (Codex). counter 를 나눠 쓰면 서로의 상한을 갉아먹는다 — 참조 이미지를 많이 구운 주행이 그날의 조사 예산을 먹어 치우는 식이다. 그래서 **세는 통을 따로** 둔다. ★그렇지만 **정지 기구는 새로 안 만든다.** `install_stop_check` · `get_current_stop_check` · `bind_current_stop_check` 는 이미지 전용이 아니라 일반 기구라 그대로 가져다 쓴다. 두 벌로 만들면 한쪽만 고쳐진다. ★★세는 것은 **물리 전송**이다 — 논리 호출이 아니다. 키 슬롯 failover 는 `FailoverOpenAIClient._invoke` 안에서 도므로 논리 하나가 물리 둘이 될 수 있다. 「몇 번 나갔나」를 막으려면 **나가는 자리**에서 세야 한다. """ from __future__ import annotations import contextlib import threading from typing import Any, Dict, Optional # ★정지 기구는 **그쪽 것을 쓴다.** 여기서 다시 만들지 않는다. from app.core.image_call_budget import get_current_stop_check class ResearchCallBudgetExceeded(RuntimeError): """상한에 닿았다. ★**네트워크에 닿기 전에** 오른다. 호출부는 이것을 잡아 그 대상을 `time_capped` · `limit_kind=transmission_budget` 으로 남긴다 — 「조사할 필요가 없었다」가 아니라 「이 주행의 상한 때문에 못 샀다」다. """ def __init__(self, *, cap: int, used: int, source: str) -> None: self.cap = int(cap) self.used = int(used) self.source = str(source) super().__init__( f"research_call_budget 초과 — 상한={self.cap} · 쓴 것={self.used} · " f"거절한 자리={self.source!r}") class ResearchCallBudget: """조사용 **물리 전송** counter. 상한을 넘으면 거절한다.""" __slots__ = ("_cap", "_used", "_denied", "_lock") def __init__(self, cap: int) -> None: cap_int = int(cap) if cap_int < 0: raise ValueError(f"상한은 0 이상이어야 한다: {cap_int}") self._cap = cap_int self._used = 0 self._denied = 0 self._lock = threading.RLock() def reserve(self, *, source: str) -> None: """전송 **직전에** 한 자리를 잡는다. 없으면 거절한다.""" with self._lock: if self._used >= self._cap: self._denied += 1 raise ResearchCallBudgetExceeded( cap=self._cap, used=self._used, 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} # ★스레드마다 따로 건다 — 이미지 예산과 같은 방식이다. 자식 스레드는 깨끗하게 # 시작하므로, 팬아웃에는 아래 `bind_current_research_budget` 으로 실어 보낸다. _local = threading.local() def install_budget(budget: ResearchCallBudget) -> None: _local.budget = budget def uninstall_budget() -> None: _local.budget = None def get_current_budget() -> Optional[ResearchCallBudget]: return getattr(_local, "budget", None) @contextlib.contextmanager def research_calls_armed(): """이 블록 **안의** provider 전송만 조사 호출로 센다. ★★★왜 필요한가 (Codex ②). `reserve` 를 `FailoverOpenAIClient._invoke` 라는 **공용 경계**에 두면, 같은 스레드에서 나가는 **다른** OpenAI 호출(이미지· 요약 등)까지 조사 예산을 먹는다. 그러면 상한이 재는 것이 「조사 전송」이 아니라 「그 스레드의 모든 전송」이 된다. 그래서 조사 호출을 내는 자리에서만 **팔을 든다**. 밖에서는 no-op 이라 다른 호출부의 동작이 **한 글자도 안 바뀐다**. ★중첩을 세는 것은 재진입(retry 안의 retry) 때문이다. """ _local.armed = getattr(_local, "armed", 0) + 1 try: yield finally: _local.armed = max(0, getattr(_local, "armed", 1) - 1) def is_armed() -> bool: return getattr(_local, "armed", 0) > 0 class OutboundAttempt: """한 **논리 시도**(journal 의 send 하나) 동안 이 스레드에서 팔을 든 채 나간 전송 수. ★Codex BLOCK (PR #82 재리뷰 2026-09-03): 중앙 경로의 send 하나는 저작→검색→받기→판정 **여러 전송**이라, 우리 문이 두 번째 전송 앞에서 세웠어도 첫 전송은 이미 나갔다. 「안 나갔다」는 이 계수가 **0** 일 때만 증명된다. 팔을 안 든 채 돈 시도는 셀 수 없으므로(`armed=False`) 증명이 없다 → uncertain. """ def __init__(self) -> None: self.count = 0 self.armed = is_armed() @contextlib.contextmanager def outbound_attempt(att: Optional["OutboundAttempt"] = None): """이 블록의 전송을 `att.count` 에 센다 — 스레드 지역 stack 이라 worker 는 제 시도만 센다.""" att = att or OutboundAttempt() stack = getattr(_local, "attempts", None) if stack is None: stack = [] _local.attempts = stack stack.append(att) try: yield att finally: stack.pop() def _count_outbound() -> None: for att in getattr(_local, "attempts", None) or (): att.count += 1 def reserve_current_research_call(*, source: str) -> None: """조사 전송 **직전**. ★`research_calls_armed()` 밖에서는 **아무 일도 안 한다**. ★★그래서 이미지·다른 스텝의 호출부가 영향을 안 받는다 — 같은 공용 경계를 지나가지만 팔을 안 들었으면 그냥 지나간다. ★정지 확인이 **예산보다 먼저**다 (Codex ①). 이미 마감된 주행은 예산도 슬롯도 안 먹고 **provider 를 0회** 부른다 — 예산이 안 깔린 조사 경로에서도 정지는 들려야 하므로, `budget is None` 보다 **앞에** 둔다. """ if not is_armed(): return stop_check = get_current_stop_check() if stop_check is not None: stop_check() budget = get_current_budget() if budget is not None: budget.reserve(source=source) # ★넘으면 여기서 던진다 — 전송은 안 나갔다(계수 안 함) _count_outbound() # ★예산 문을 지났다 = 이제 나간다 class RunDeadlineExceeded(RuntimeError): """주행 전체 벽시계가 지났다. ★`limit_kind=run_deadline` 로 남는다. ★표식(`is_run_deadline`)을 달고 다닌다 — 정지 표는 호출부가 아무 예외나 올릴 수 있어서, 타입 하나로 못 가른다. """ is_run_deadline = True @contextlib.contextmanager def research_run_scope(*, cap: int, deadline_seconds: Optional[float] = None, now: Any = None): """한 **주행**의 조사 상한을 연다 — 전송 수 ②와 벽시계 ③. ★★③ 을 「새 대상을 안 시작한다」는 loop 조건으로만 두면 **팬아웃 안에서 이미 제출된 것들**이 계속 나간다. 정지 표로 깔면 `reserve` 자리에서 막힌다 — 그게 **네트워크 직전**이다 (GROUNDING-V2 §8.5). ★들어올 때 걸려 있던 예산·정지 표는 **되돌린다**. 안 그러면 이 주행이 끝난 뒤에도 남의 호출이 이 상한을 본다. """ import time as _time from app.core.image_call_budget import (get_current_stop_check, install_stop_check, uninstall_stop_check) clock = now or _time.monotonic started = clock() prev_budget = get_current_budget() prev_check = get_current_stop_check() budget = ResearchCallBudget(cap=cap) install_budget(budget) if deadline_seconds is not None: def _check(_prev=prev_check): # ★앞 표를 **먼저** 본다 — 사용자 취소가 마감보다 우선이다. if _prev is not None: _prev() if clock() - started >= float(deadline_seconds): raise RunDeadlineExceeded( f"주행 마감 {deadline_seconds}초가 지났다 — 남은 대상은 " "`time_capped`(run_deadline) 로 남는다") install_stop_check(_check) try: yield budget finally: if prev_budget is None: uninstall_budget() else: install_budget(prev_budget) if deadline_seconds is not None: if prev_check is None: uninstall_stop_check() else: install_stop_check(prev_check) def bind_current_research_budget(fn: Any) -> Any: """팬아웃으로 **예산과 정지 표를 같이** 실어 보낸다. ★`install_budget` 은 스레드 지역이라, 그냥 넘기면 worker 는 예산을 못 본다 — 그러면 상한이 그 자리에서 통째로 no-op 이 된다. """ import functools from app.core.image_call_budget import bind_current_stop_check budget = get_current_budget() armed = is_armed() # ★팔을 든 채 팬아웃했으면 worker 도 든다 — 안 그러면 worker 의 전송은 한 번도 안 세어진다 (PR #82 리뷰) inner = bind_current_stop_check(fn) @functools.wraps(fn) def _run(*args: Any, **kwargs: Any) -> Any: prev = get_current_budget() if budget is not None: install_budget(budget) try: if armed: with research_calls_armed(): return inner(*args, **kwargs) return inner(*args, **kwargs) finally: # ★원래대로 되돌린다 — 안 그러면 그 worker 가 다음 일에도 이 # 예산을 들고 간다(pool 은 스레드를 재사용한다). if prev is None: uninstall_budget() else: install_budget(prev) return _run __all__ = [ "ResearchCallBudget", "is_armed", "research_calls_armed", "research_run_scope", "RunDeadlineExceeded", "ResearchCallBudgetExceeded", "bind_current_research_budget", "get_current_budget", "install_budget", "reserve_current_research_call", "uninstall_budget", ]