Merge 5c76148f71 into 1d52985d03

litellm/caching/caching_handler.py

@@ -738,10 +738,13 @@ class LLMCachingHandler:
             end_time (datetime): The end time of the operation.
             cache_hit (bool): Whether it was a cache hit.
         """
+        from litellm.litellm_core_utils.litellm_logging import evaluation_logging_snapshot
         from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
 
+        receipt_logger: Final = evaluation_logging_snapshot(logging_obj)
+
         GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue(
-            async_coroutine=logging_obj.async_success_handler(
+            async_coroutine=receipt_logger.async_success_handler(
                 result=cached_result,
                 start_time=start_time,
                 end_time=end_time,

litellm/caching/in_memory_cache.py

@@ -168,7 +168,7 @@ class InMemoryCache(BaseCache):
             return
 
         self.cache_dict[key] = value
-        if self.allow_ttl_override(key):  # if ttl is not set, set it to default ttl
+        if kwargs.get("refresh_ttl", False) or self.allow_ttl_override(key):
             if "ttl" in kwargs and kwargs["ttl"] is not None:
                 self.ttl_dict[key] = self._clock() + float(kwargs["ttl"])
                 heapq.heappush(self.expiration_heap, (self.ttl_dict[key], key))

litellm/integrations/lago.py

@@ -3,14 +3,19 @@
 
 import json
 import os
+from collections.abc import Mapping
 from typing import Final, Literal
 
 import httpx
+from pydantic import TypeAdapter
 
 import litellm
 from litellm._logging import verbose_logger
 from litellm._uuid import uuid
 from litellm.integrations.custom_logger import CustomLogger
+from litellm.litellm_core_utils.internal_call_metadata import (
+    get_evaluation_billing_owner_from_kwargs,
+)
 from litellm.llms.custom_httpx.http_handler import (
     HTTPHandler,
     get_async_httpx_client,
@@ -61,6 +66,9 @@ class LagoLogger(CustomLogger):
             raise Exception(f"Missing keys={missing_keys} in environment.")
 
     def _common_logic(self, kwargs: dict, response_obj) -> dict:
+        billing_owner: Final = get_evaluation_billing_owner_from_kwargs(
+            TypeAdapter(Mapping[str, object]).validate_python(kwargs)
+        )
         response_obj.get("id", kwargs.get("litellm_call_id"))
         get_utc_datetime().isoformat()
         cost: Final = kwargs.get("response_cost", None)
@@ -96,7 +104,9 @@ class LagoLogger(CustomLogger):
             else:
                 raise Exception("invalid LAGO_API_CHARGE_BY set")
 
-        if charge_by == "end_user_id":
+        if billing_owner is not None:
+            external_customer_id = billing_owner.user_id
+        elif charge_by == "end_user_id":
             external_customer_id = end_user_id
         elif charge_by == "team_id":
             external_customer_id = team_id

litellm/integrations/shadow_eval_logger.py

@@ -30,7 +30,11 @@ from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY
 from litellm.integrations.custom_logger import CustomLogger
 from litellm.integrations.websearch_interception.tools import is_web_search_tool_responses
 from litellm.litellm_core_utils.core_helpers import get_litellm_metadata_from_kwargs, independent_snapshot
-from litellm.litellm_core_utils.internal_call_metadata import sanitized_forwardable_call_metadata
+from litellm.litellm_core_utils.internal_call_metadata import (
+    EvaluationBillingOwner,
+    evaluation_billing_context,
+    sanitized_forwardable_call_metadata,
+)
 from litellm.litellm_core_utils.llm_judge import (
     default_router_provider,
     extract_text_from_content,
@@ -76,6 +80,7 @@ _EMPTY_METADATA: Final[Mapping[str, object]] = MappingProxyType({})
 _CHAT_REQUEST_ADAPTER: Final = TypeAdapter(Mapping[str, object])
 _CHAT_MESSAGES_ADAPTER: Final = TypeAdapter(tuple[Mapping[str, object], ...])
 _MESSAGE_ITEMS_ADAPTER: Final = TypeAdapter(tuple[object, ...])
+_CREATOR_MODEL_BUDGET: Final = TypeAdapter[Mapping[str, object] | None](Mapping[str, object] | None)
 
 
 def _chat_messages(kwargs: Mapping[str, object]) -> tuple[Mapping[str, object], ...]:
@@ -747,6 +752,7 @@ class ActiveShadowEvalJob(BaseModel):
     baseline_model: str | None = None
     shadow_percentage: float
     judge_model: str
+    created_by: str | None = None
     max_turns: int
     max_budget: float | None = None
     ends_at: datetime
@@ -1082,22 +1088,27 @@ class ShadowEvalLogger(CustomLogger):
             if spend >= job.max_budget:
                 self._record_funnel(job.id, "withheld")
                 return
-        for arm_router in job.arm_router_names:
-            await self._run_shadow_arm(
-                prisma=prisma,
-                job=job,
-                arm_router=arm_router,
-                request_id=request_id,
-                messages=messages,
-                real_text=real_text,
-                real_model=real_model,
-                real_cost=real_cost,
-                real_classifier_cost=real_classifier_cost,
-                real_cache_hit=real_cache_hit,
-                control_tier=control_tier,
-                shadow_params=shadow_params,
-                parent_metadata=parent_metadata,
-            )
+        owner: Final = await _evaluation_billing_owner(prisma, job.created_by)
+        if owner is None:
+            self._record_funnel(job.id, "withheld")
+            return
+        with evaluation_billing_context(owner):
+            for arm_router in job.arm_router_names:
+                await self._run_shadow_arm(
+                    prisma=prisma,
+                    job=job,
+                    arm_router=arm_router,
+                    request_id=request_id,
+                    messages=messages,
+                    real_text=real_text,
+                    real_model=real_model,
+                    real_cost=real_cost,
+                    real_classifier_cost=real_classifier_cost,
+                    real_cache_hit=real_cache_hit,
+                    control_tier=control_tier,
+                    shadow_params=shadow_params,
+                    parent_metadata=parent_metadata,
+                )
 
     async def _run_shadow_arm(
         self,
@@ -1285,6 +1296,7 @@ class ShadowEvalLogger(CustomLogger):
                 model=target_model,
                 messages=[dict(m) for m in messages],  # pyright: ignore[reportArgumentType]  # snapshot of the SDK's own message dicts
                 metadata=shadow_metadata,
+                stream=False,
                 num_retries=0,
                 fallbacks=[],
                 **shadow_params,
@@ -1348,6 +1360,7 @@ class ShadowEvalLogger(CustomLogger):
                 judge_model,
                 judge_messages,  # pyright: ignore[reportArgumentType]  # plain SDK message dicts
                 team_id=_forwarded_team_id(parent_metadata),
+                stream=False,
                 temperature=0,
                 max_tokens=JUDGE_MAX_OUTPUT_TOKENS,
                 response_format=PAIRWISE_JUDGE_RESPONSE_FORMAT,
@@ -1380,3 +1393,32 @@ def _default_prisma_provider() -> "PrismaClient | None":
     except ImportError:
         return None
     return prisma_client
+
+
+async def _evaluation_billing_owner(prisma: "PrismaClient", created_by: str | None) -> EvaluationBillingOwner | None:
+    from litellm.proxy.auth.auth_checks import get_user_object
+    from litellm.proxy.proxy_server import litellm_proxy_admin_name, proxy_logging_obj, user_api_key_cache
+    from litellm.types.proxy.auth.auth_checks import UserNotFoundError
+
+    creator_id: Final = created_by or litellm_proxy_admin_name
+    try:
+        creator: Final = await get_user_object(
+            user_id=creator_id,
+            prisma_client=prisma,
+            user_api_key_cache=user_api_key_cache,
+            user_id_upsert=False,
+            proxy_logging_obj=proxy_logging_obj,
+        )
+    except UserNotFoundError:
+        return None
+    except Exception as e:  # noqa: BLE001  # optional evaluation work must not spend against unverifiable limits
+        verbose_logger.warning("shadow_eval: creator budget unavailable for %s: %s", creator_id, e)
+        return None
+    if creator is None:
+        return None
+    return EvaluationBillingOwner(
+        creator_id,
+        _CREATOR_MODEL_BUDGET.validate_python(creator.model_max_budget),
+        creator.max_budget,
+        creator.spend or 0.0,
+    )

litellm/litellm_core_utils/fallback_utils.py

@@ -60,7 +60,7 @@ async def async_completion_with_fallbacks(**kwargs):
             completion_kwargs = filter_internal_params(completion_kwargs)
 
             response = await litellm.acompletion(
-                **completion_kwargs,
+                **{**completion_kwargs, "fallbacks": []},
                 model=model,
                 litellm_logging_obj=litellm_logging_obj,
             )

litellm/litellm_core_utils/internal_call_metadata.py

@@ -1,9 +1,7 @@
 """Metadata a request forwards to the internal LLM sub-calls it triggers.
 
-Internal features (the auto-router's classifier and embeddings, shadow eval's shadow and
-judge calls) bill real provider spend that nobody typed a prompt for. That spend must land
-on the same key/team/org/user as the request that caused it, so the sub-call carries the
-caller's identity metadata, minus two things that must never be forwarded as-is:
+Internal calls retain the caller's routing identity. Shadow evaluation receipts bill the
+evaluation creator without changing that routing context. Two fields need special handling:
 
 * ``user_api_key_budget_reservation`` (and the reservation nested inside
   ``user_api_key_auth``) belongs to the parent completion. If a sub-call's cost callback
@@ -17,16 +15,118 @@ caller's identity metadata, minus two things that must never be forwarded as-is:
 
 from __future__ import annotations
 
-from collections.abc import Mapping
+from collections.abc import Generator, Mapping
+from contextlib import contextmanager
+from contextvars import ContextVar
+from dataclasses import dataclass
 from types import MappingProxyType
 from typing import Final
 
+from pydantic import TypeAdapter
+
 from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY, NON_INFERENCE_CALL_TYPES
 from litellm.litellm_core_utils.initialize_dynamic_callback_params import initialize_standard_callback_dynamic_params
 from litellm.types.utils import BACKGROUND_RESPONSE_COST_POLL_CALL_ORIGIN, InternalCallOrigin
 
 BUDGET_RESERVATION_METADATA_KEYS: Final = frozenset({"user_api_key_budget_reservation"})
 
+
+@dataclass(frozen=True, slots=True)
+class EvaluationBillingOwner:
+    user_id: str
+    user_model_max_budget: Mapping[str, object] | None = None
+    max_budget: float | None = None
+    spend: float = 0.0
+
+
+EVALUATION_BILLING_OWNER_KEY: Final = "_evaluation_billing_owner"
+EVALUATION_BUDGET_RESERVATION_KEY: Final = "_evaluation_budget_reservation"
+_EVALUATION_BILLING_OWNER: Final[ContextVar[EvaluationBillingOwner | None]] = ContextVar(
+    "evaluation_billing_owner", default=None
+)
+_BILLING_MAPPING: Final = TypeAdapter(Mapping[str, object])
+_BILLING_IDENTITY_FIELDS: Final = frozenset(
+    {"user_api_key", "user_api_end_user_max_budget", "team_id", "team_alias", "agent_id", "billing_agent_id"}
+)
+
+
+def get_evaluation_billing_owner() -> EvaluationBillingOwner | None:
+    return _EVALUATION_BILLING_OWNER.get()
+
+
+@contextmanager
+def evaluation_billing_context(owner: EvaluationBillingOwner | None) -> Generator[None]:
+    token: Final = _EVALUATION_BILLING_OWNER.set(owner)
+    try:
+        yield
+    finally:
+        _EVALUATION_BILLING_OWNER.reset(token)
+
+
+def get_evaluation_billing_owner_from_kwargs(kwargs: Mapping[str, object]) -> EvaluationBillingOwner | None:
+    owner: Final = kwargs.get(EVALUATION_BILLING_OWNER_KEY)
+    return owner if isinstance(owner, EvaluationBillingOwner) else None
+
+
+def _billing_mapping(value: object) -> Mapping[str, object]:
+    return _BILLING_MAPPING.validate_python(value) if isinstance(value, Mapping) else MappingProxyType({})
+
+
+def project_evaluation_billing_kwargs(
+    kwargs: Mapping[str, object],
+) -> dict[str, object]:  # mutable-ok: existing callback consumers require dictionaries
+    owner: Final = get_evaluation_billing_owner_from_kwargs(kwargs)
+    if owner is None:
+        return kwargs if isinstance(kwargs, dict) else dict(kwargs)
+    from litellm.proxy.spend_tracking.evaluation_budget import EvaluationAttempt
+
+    handle: Final = kwargs.get(EVALUATION_BUDGET_RESERVATION_KEY)
+    reservation: Final = handle.total if isinstance(handle, EvaluationAttempt) else None
+
+    def metadata(value: object) -> Mapping[str, object]:
+        return {
+            **{
+                key: None if key.startswith("user_api_key_") or key in _BILLING_IDENTITY_FIELDS else item
+                for key, item in _billing_mapping(value).items()
+            },
+            "user_api_key_user_id": owner.user_id,
+            "user_api_key_user_model_max_budget": owner.user_model_max_budget,
+            "user_api_key_budget_reservation": reservation,
+            "tags": [],
+        }
+
+    def fields(value: object) -> Mapping[str, object]:
+        source: Final = _billing_mapping(value)
+        request: Final = _billing_mapping(source.get("proxy_server_request"))
+        return {
+            **source,
+            "user": owner.user_id,
+            "end_user": None,
+            "user_api_key_end_user_id": None,
+            "agent_id": None,
+            "billing_agent_id": None,
+            "request_tags": [],
+            "request_model_access_groups": (),
+            "metadata": metadata(source.get("metadata")),
+            **({"litellm_metadata": metadata(source["litellm_metadata"])} if source.get("litellm_metadata") else {}),
+            **(
+                {"proxy_server_request": {**request, "body": {**_billing_mapping(request["body"]), "user": None}}}
+                if isinstance(request.get("body"), Mapping)
+                else {}
+            ),
+        }
+
+    return {
+        **fields(kwargs),
+        "litellm_params": fields(kwargs.get("litellm_params")),
+        **(
+            {"standard_logging_object": fields(kwargs["standard_logging_object"])}
+            if kwargs.get("standard_logging_object") is not None
+            else {}
+        ),
+    }
+
+
 MODEL_ACCESS_GROUP_METADATA_KEY: Final = "user_api_key_matched_model_access_groups"
 """Where auth records the model access groups that authorized the request, for the spend writer.
 

litellm/litellm_core_utils/litellm_logging.py

@@ -11,13 +11,14 @@ import sys
 import time
 import traceback
 from collections.abc import Awaitable, Callable, Iterator, Mapping, Sequence
+from dataclasses import dataclass
 from datetime import datetime as dt_object
 from functools import lru_cache
 from types import MappingProxyType, TracebackType
 from typing import TYPE_CHECKING, Any, Final, Literal, Union, cast
 
 from httpx import Response
-from pydantic import BaseModel, JsonValue
+from pydantic import BaseModel, JsonValue, TypeAdapter
 
 import litellm
 from litellm import _custom_logger_compatible_callbacks_literal
@@ -79,8 +80,13 @@ from litellm.litellm_core_utils.core_helpers import (
 from litellm.litellm_core_utils.error_normalization import normalize_error
 from litellm.litellm_core_utils.get_litellm_params import get_litellm_params
 from litellm.litellm_core_utils.internal_call_metadata import (
+    EVALUATION_BILLING_OWNER_KEY,
+    EVALUATION_BUDGET_RESERVATION_KEY,
     MODEL_ACCESS_GROUP_METADATA_KEY,
+    EvaluationBillingOwner,
+    get_evaluation_billing_owner,
     is_unbilled_non_inference_call,
+    project_evaluation_billing_kwargs,
 )
 from litellm.litellm_core_utils.llm_cost_calc.guardrail_cost import (
     cost_breakdown_with_guardrail,
@@ -236,6 +242,7 @@ if TYPE_CHECKING:
     from litellm.litellm_core_utils.llm_cost_calc.utils import BilledTokenRates
     from litellm.llms.base_llm.passthrough.transformation import PassthroughStreamCollector
     from litellm.proxy.hooks.autorouter_baseline_cache import BaselineCacheContext, CapturedBaselineObservation
+    from litellm.proxy.spend_tracking.evaluation_budget import EvaluationAttempt
 try:
     from litellm_enterprise.enterprise_callbacks.callback_controls import (
         EnterpriseCallbackControls,
@@ -553,6 +560,22 @@ def _timestamp_seconds(moment: object) -> float | None:
     return None
 
 
+@dataclass(slots=True)
+class EvaluationBudgetInvocation:
+    reservation: "EvaluationAttempt | None" = None
+
+
+def evaluation_logging_snapshot(logging_obj: "Logging") -> "Logging":
+    if not isinstance(logging_obj.evaluation_billing_owner, EvaluationBillingOwner):
+        return logging_obj
+    receipt: Final = copy.copy(logging_obj)
+    receipt.model_call_details.update(project_evaluation_billing_kwargs(receipt.model_call_details))
+    receipt.litellm_params = TypeAdapter(dict[str, object]).validate_python(
+        receipt.model_call_details["litellm_params"]
+    )
+    return receipt
+
+
 class Logging(LiteLLMLoggingBaseClass):
     global \
         supabaseClient, \
@@ -618,6 +641,9 @@ class Logging(LiteLLMLoggingBaseClass):
         self.call_type = call_type
         self.litellm_call_id = litellm_call_id
         self.litellm_trace_id: str = litellm_trace_id if litellm_trace_id else str(uuid.uuid4())
+        self.evaluation_billing_owner: Final[EvaluationBillingOwner | None] = get_evaluation_billing_owner()
+        self.evaluation_budget_reservation: EvaluationAttempt | None = None
+        self.evaluation_budget_invocation: EvaluationBudgetInvocation | None = None
 
         # Capture the pre-call *value* (not a contextvars.Token) so restoration works
         # even if this attempt's own logging ends up dispatched onto a different
@@ -714,6 +740,8 @@ class Logging(LiteLLMLoggingBaseClass):
             "litellm_params": litellm_params,
             "applied_guardrails": applied_guardrails,
             "model": model,
+            EVALUATION_BILLING_OWNER_KEY: self.evaluation_billing_owner,
+            EVALUATION_BUDGET_RESERVATION_KEY: self.evaluation_budget_reservation,
         }
 
         # Set by proxy request handlers to defer spend-log fire until after
@@ -945,6 +973,8 @@ class Logging(LiteLLMLoggingBaseClass):
                 "standard_callback_dynamic_params": self.standard_callback_dynamic_params,
                 **self.optional_params,
                 **additional_params,
+                EVALUATION_BILLING_OWNER_KEY: self.evaluation_billing_owner,
+                EVALUATION_BUDGET_RESERVATION_KEY: self.evaluation_budget_reservation,
             }
         )
 
@@ -2197,7 +2227,7 @@ class Logging(LiteLLMLoggingBaseClass):
             return
 
         executor.submit(
-            self.success_handler,
+            evaluation_logging_snapshot(self).success_handler,
             result,
             start_time=start_time,
             end_time=end_time,
@@ -2216,6 +2246,22 @@ class Logging(LiteLLMLoggingBaseClass):
         if isinstance(usage, Usage):
             self.record_partial_usage_for_failure(usage, self._response_cost_calculator(result=assembled) or 0.0)
 
+    def recover_failure_cost(self, result: object) -> float:
+        if isinstance(result, ModelResponse):
+            self.record_assembled_response_for_failure(result)
+        elif isinstance(result, ResponsesAPIResponse):
+            self.record_assembled_response_for_failure(self._translate_responses_api_response_to_model_response(result))
+            self.model_call_details["response_cost"] = self._response_cost_calculator(result=result)
+        elif result is not None and self.call_type == CallTypes.anthropic_messages.value:
+            self.record_assembled_response_for_failure(self._handle_anthropic_messages_response_logging(result))
+        return TypeAdapter(float).validate_python(self.model_call_details.get("response_cost") or 0.0)
+
+    async def _settle_evaluation_budget(self) -> None:
+        if self.evaluation_budget_reservation is not None:
+            await self.evaluation_budget_reservation.settle(
+                TypeAdapter(float).validate_python(self.model_call_details.get("response_cost") or 0.0)
+            )
+
     async def dispatch_failure_handlers(
         self,
         exception: Exception,
@@ -2244,7 +2290,7 @@ class Logging(LiteLLMLoggingBaseClass):
             await self.async_failure_handler(exception, traceback_exception)
         finally:
             if self._should_run_sync_failure_callbacks_for_async_calls():
-                executor.submit(self.failure_handler, exception, traceback_exception)
+                executor.submit(evaluation_logging_snapshot(self).failure_handler, exception, traceback_exception)
 
     def should_run_logging(
         self,
@@ -2505,6 +2551,8 @@ class Logging(LiteLLMLoggingBaseClass):
         standard_logging_object: StandardLoggingPayload | None = None,
         build_logging_payload: bool = True,
     ):
+        if self.evaluation_billing_owner is not None:
+            self.model_call_details.update(project_evaluation_billing_kwargs(self.model_call_details))
         try:
             if start_time is None:
                 start_time = self.start_time
@@ -3395,6 +3443,9 @@ class Logging(LiteLLMLoggingBaseClass):
                 # print standard logging payload
                 if (standard_logging_payload := self.model_call_details.get("standard_logging_object")) is not None:
                     emit_standard_logging_payload(standard_logging_payload)
+        if not self.stream or "async_complete_streaming_response" in self.model_call_details:
+            await self._settle_evaluation_budget()
+
         callbacks: Final = self.get_combined_callback_list(
             dynamic_success_callbacks=self.dynamic_async_success_callbacks,
             global_callbacks=litellm._async_success_callback,
@@ -3605,6 +3656,8 @@ class Logging(LiteLLMLoggingBaseClass):
             verbose_logger.debug("Error in _handle_callback_failure: %s", e)
 
     def _failure_handler_helper_fn(self, exception, traceback_exception, start_time=None, end_time=None):
+        if self.evaluation_billing_owner is not None and hasattr(self, "model_call_details"):
+            self.model_call_details.update(project_evaluation_billing_kwargs(self.model_call_details))
         if start_time is None:
             start_time = self.start_time
         if end_time is None:
@@ -3639,6 +3692,9 @@ class Logging(LiteLLMLoggingBaseClass):
             metadata: Final = self.model_call_details["litellm_params"].get("metadata", {}) or {}
             metadata.update(exception.headers)
 
+        if self.evaluation_billing_owner is not None:
+            self.model_call_details.update(project_evaluation_billing_kwargs(self.model_call_details))
+
         ## STANDARDIZED LOGGING PAYLOAD
 
         self.model_call_details["standard_logging_object"] = get_standard_logging_object_payload(
@@ -3920,6 +3976,7 @@ class Logging(LiteLLMLoggingBaseClass):
             start_time=start_time,
             end_time=end_time,
         )
+        await self._settle_evaluation_budget()
 
         callbacks: Final = self.get_combined_callback_list(
             dynamic_success_callbacks=self.dynamic_async_failure_callbacks,
@@ -4002,7 +4059,7 @@ class Logging(LiteLLMLoggingBaseClass):
             return
 
         executor.submit(
-            self.success_handler,
+            evaluation_logging_snapshot(self).success_handler,
             result,
             start_time,
             end_time,
@@ -6537,8 +6594,12 @@ def get_standard_logging_object_payload(
         _model_id: Final = metadata.get("model_info", {}).get("id", "")
         _model_group: Final = metadata.get("model_group", "")
 
-        request_tags: Final = StandardLoggingPayloadSetup._get_request_tags(
-            litellm_params=litellm_params, proxy_server_request=proxy_server_request
+        request_tags: Final = (
+            []
+            if logging_obj.evaluation_billing_owner is not None
+            else StandardLoggingPayloadSetup._get_request_tags(
+                litellm_params=litellm_params, proxy_server_request=proxy_server_request
+            )
         )
         request_model_access_groups: Final = request_model_access_groups_from_litellm_params(litellm_params)
 

litellm/litellm_core_utils/llm_judge.py

@@ -118,7 +118,7 @@ async def judge_acompletion(
 ) -> ModelResponse:
     """Dispatch a judge call through the proxy's router when the judge model is a
     configured deployment (DB-stored credentials work), through the SDK for
-    provider-qualified public names. The router path never retries or falls back:
+    provider-qualified public names. Judge calls never retry or fall back:
     a failed judge call is the caller's counted failure, not a spend multiplier.
     Sampling preferences are advisory: models that removed sampling params (e.g.
     claude-sonnet-5) drop them instead of rejecting the judge call.
@@ -135,4 +135,6 @@ async def judge_acompletion(
             drop_params=True,
             **params,
         )
-    return await litellm.acompletion(model=judge_model, messages=messages, num_retries=0, drop_params=True, **params)
+    return await litellm.acompletion(
+        model=judge_model, messages=messages, num_retries=0, fallbacks=[], drop_params=True, **params
+    )

litellm/llms/anthropic/pass_through/messages/handler.py

@@ -7,10 +7,12 @@
 
 import asyncio
 import contextvars
-from collections.abc import AsyncIterator, Coroutine, Iterator
+from collections.abc import AsyncIterator, Coroutine, Iterator, Mapping
 from functools import partial
 from typing import Any, Final, cast
 
+from pydantic import TypeAdapter
+
 import litellm
 from litellm.litellm_core_utils.exception_mapping_utils import exception_type
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
@@ -106,6 +108,32 @@ def _deployment_supports_cache_control_ttl(model_info: object) -> bool:
     return isinstance(model_info, dict) and model_info.get("cache_control_ttl") is True
 
 
+def get_anthropic_messages_provider_config(
+    model: str, custom_llm_provider: str | None, model_info: object
+) -> BaseAnthropicMessagesConfig | None:
+    from litellm.types.utils import LlmProviders
+
+    config: Final = (
+        ProviderConfigManager.get_provider_anthropic_messages_config(model, LlmProviders(custom_llm_provider))
+        if custom_llm_provider is not None and custom_llm_provider in (provider.value for provider in LlmProviders)
+        else None
+    )
+    if config is not None or not _deployment_passes_through_anthropic_messages(model_info):
+        return config
+    from litellm.llms.openai_like.messages.transformation import OpenAILikeAnthropicMessagesConfig
+
+    return OpenAILikeAnthropicMessagesConfig(cache_control_ttl=_deployment_supports_cache_control_ttl(model_info))
+
+
+def anthropic_messages_uses_bridge(model: str, request: Mapping[str, object]) -> bool:
+    resolved_model, provider, _, _ = litellm.get_llm_provider(
+        model=model,
+        custom_llm_provider=TypeAdapter[str | None](str | None).validate_python(request.get("custom_llm_provider")),
+        api_base=TypeAdapter[str | None](str | None).validate_python(request.get("api_base")),
+    )
+    return get_anthropic_messages_provider_config(resolved_model, provider, request.get("model_info")) is None
+
+
 ####### ENVIRONMENT VARIABLES ###################
 # Initialize any necessary instances or variables here
 base_llm_http_handler = BaseLLMHTTPHandler()
@@ -452,8 +480,6 @@ def anthropic_messages_handler(
     Args:
         container: Container config with skills for code execution
     """
-    from litellm.types.utils import LlmProviders
-
     # Sanitize empty text blocks so the sync entry point
     # (litellm.messages.create -> anthropic_messages_handler) gets the same
     # protection as the async wrapper. The async wrapper already sanitized and
@@ -561,23 +587,9 @@ def anthropic_messages_handler(
                 **kwargs,
             )
 
-    anthropic_messages_provider_config: BaseAnthropicMessagesConfig | None = None
-
-    if custom_llm_provider is not None and custom_llm_provider in [provider.value for provider in LlmProviders]:
-        anthropic_messages_provider_config = ProviderConfigManager.get_provider_anthropic_messages_config(
-            model=model,
-            provider=litellm.LlmProviders(custom_llm_provider),
-        )
-    if anthropic_messages_provider_config is None and _deployment_passes_through_anthropic_messages(
-        kwargs.get("model_info")
-    ):
-        from litellm.llms.openai_like.messages.transformation import (
-            OpenAILikeAnthropicMessagesConfig,
-        )
-
-        anthropic_messages_provider_config = OpenAILikeAnthropicMessagesConfig(
-            cache_control_ttl=_deployment_supports_cache_control_ttl(kwargs.get("model_info")),
-        )
+    anthropic_messages_provider_config: Final = get_anthropic_messages_provider_config(
+        model, custom_llm_provider, TypeAdapter(object).validate_python(kwargs.get("model_info"))
+    )
     if anthropic_messages_provider_config is None:
         # Route to Responses API for OpenAI / Azure, chat/completions for everything else.
         if kwargs.get("compaction") is None and _should_route_to_responses_api(

litellm/main.py

@@ -643,9 +643,11 @@ async def acompletion(
             api_base=kwargs.get("api_base") or base_url,
         )
 
-    fallbacks = fallbacks or litellm.model_fallbacks
-    if fallbacks is not None:
-        response = await async_completion_with_fallbacks(**completion_kwargs, kwargs={"fallbacks": fallbacks, **kwargs})
+    resolved_fallbacks: Final = litellm.model_fallbacks if fallbacks is None else fallbacks
+    if resolved_fallbacks:
+        response = await async_completion_with_fallbacks(
+            **completion_kwargs, kwargs={**kwargs, "fallbacks": resolved_fallbacks}
+        )
         if response is None:
             raise Exception(
                 "No response from fallbacks. Got none. Turn on `litellm.set_verbose=True` to see more details."
@@ -5460,10 +5462,10 @@ def completion(
         elif num_retries is not None:
             max_retries = num_retries
         logging: Final[LiteLLMLoggingObj] = cast(LiteLLMLoggingObj, litellm_logging_obj)
-        fallbacks = fallbacks or litellm.model_fallbacks
-        if fallbacks is not None:
+        resolved_fallbacks: Final = litellm.model_fallbacks if fallbacks is None else fallbacks
+        if resolved_fallbacks:
             return completion_with_fallbacks(  # pyright: ignore[reportReturnType]  # fallback runner is untyped; resolves to ModelResponse|CustomStreamWrapper at runtime
-                **args
+                **{**args, "kwargs": {**kwargs, "fallbacks": resolved_fallbacks}}
             )
         if model_list is not None:
             deployments: Final = [m["litellm_params"] for m in model_list if m["model_name"] == model]

litellm/proxy/auth/user_api_key_auth.py

@@ -19,6 +19,7 @@ import fastapi
 import orjson
 from fastapi import HTTPException, Request, WebSocket, status
 from fastapi.security.api_key import APIKeyHeader
+from pydantic import TypeAdapter
 from starlette.exceptions import WebSocketException
 
 import litellm
@@ -39,6 +40,7 @@ from litellm.integrations.otel.model.config import is_otel_v2_enabled
 from litellm.integrations.otel.runtime import phase_event, phase_span, seed_request_identity
 from litellm.litellm_core_utils.dd_tracing import tracer
 from litellm.litellm_core_utils.dot_notation_indexing import get_nested_value
+from litellm.litellm_core_utils.internal_call_metadata import get_evaluation_billing_owner
 from litellm.proxy._types import *
 from litellm.proxy.agent_endpoints.auth.agent_caller import agent_caller_from_headers
 from litellm.proxy.auth.auth_checks import (
@@ -3224,13 +3226,17 @@ async def _reserve_budget_after_common_checks(
     request: Request | None = None,
 ) -> None:
     user_api_key_auth_obj.budget_reservation = None
-    if not skip_budget_checks and general_settings.get("disable_budget_reservation") is not True:
+    if (
+        not skip_budget_checks
+        and general_settings.get("disable_budget_reservation") is not True
+        and get_evaluation_billing_owner() is None
+    ):
         from litellm.proxy.spend_tracking.budget_reservation import (
             reserve_budget_for_request,
         )
 
         user_api_key_auth_obj.budget_reservation = await reserve_budget_for_request(
-            request_body=request_data,
+            request_body=TypeAdapter(dict[str, object]).validate_python(request_data),
             route=route,
             llm_router=llm_router,
             valid_token=user_api_key_auth_obj,

litellm/proxy/hooks/model_max_budget_limiter.py

@@ -470,7 +470,13 @@ class _PROXY_VirtualKeyModelMaxBudgetLimiter(RouterBudgetLimiting):
             model=model,
             resolved=resolved,
         )
-        current_spend: Final = _as_spend(await self._cached_spend(spend_key))
+        from litellm.proxy.spend_tracking.evaluation_budget import model_budget_spend
+
+        current_spend: Final = (
+            await model_budget_spend(self.dual_cache, spend_key)
+            if entity_type == Litellm_EntityType.USER
+            else _as_spend(await self._cached_spend(spend_key))
+        )
         if legacy_spend_key is None or legacy_spend_key == spend_key:
             return current_spend
         return current_spend + _as_spend(await self._cached_spend(legacy_spend_key))
@@ -499,6 +505,14 @@ class _PROXY_VirtualKeyModelMaxBudgetLimiter(RouterBudgetLimiting):
 
         Example: key=sk-1234567890, model=gpt-4o, max_budget=100, time_period=1d
         """
+        from litellm.litellm_core_utils.internal_call_metadata import EVALUATION_BUDGET_RESERVATION_KEY
+        from litellm.proxy.spend_tracking.evaluation_budget import EvaluationAttempt
+
+        if (
+            isinstance(attempt := kwargs.get(EVALUATION_BUDGET_RESERVATION_KEY), EvaluationAttempt)
+            and attempt.model is not None
+        ):
+            return
         verbose_proxy_logger.debug("in RouterBudgetLimiting.async_log_success_event")
         standard_logging_payload: Final[StandardLoggingPayload | None] = kwargs.get("standard_logging_object", None)
         if standard_logging_payload is None:

litellm/proxy/hooks/proxy_track_cost_callback.py

@@ -4,6 +4,8 @@ from collections.abc import Callable, Mapping, Sequence
 from datetime import datetime
 from typing import TYPE_CHECKING, Any, Final, Protocol, cast
 
+from pydantic import TypeAdapter
+
 import litellm
 from litellm._logging import verbose_proxy_logger
 from litellm.batches.batch_utils import batch_cost_is_final
@@ -15,7 +17,14 @@ from litellm.litellm_core_utils.core_helpers import (
     get_litellm_metadata_from_kwargs,
     get_metadata_variable_name_from_kwargs,
 )
-from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
+from litellm.litellm_core_utils.internal_call_metadata import (
+    EVALUATION_BILLING_OWNER_KEY,
+    EVALUATION_BUDGET_RESERVATION_KEY,
+    get_evaluation_billing_owner,
+    get_evaluation_billing_owner_from_kwargs,
+    project_evaluation_billing_kwargs,
+)
+from litellm.litellm_core_utils.litellm_logging import Logging, StandardLoggingPayloadSetup
 from litellm.litellm_core_utils.llm_cost_calc.guardrail_cost import guardrail_information_cost
 from litellm.proxy._types import UserAPIKeyAuth
 from litellm.proxy.auth.auth_checks import (
@@ -273,19 +282,34 @@ class _ProxyDBLogger(CustomLogger):
         recovered_response_cost: Final = recovered_stream_cost + guardrail_information_cost(
             existing_metadata.get("standard_logging_guardrail_information")
         )
+        logger: Final = _litellm_logging_obj if isinstance(_litellm_logging_obj, Logging) else None
+        request_receipt: Final = TypeAdapter(dict[str, object]).validate_python(request_data)
+        owner: Final = (
+            get_evaluation_billing_owner_from_kwargs(request_receipt)
+            or (logger.evaluation_billing_owner if logger is not None else None)
+            or get_evaluation_billing_owner()
+        )
+        billing_auth: Final = UserAPIKeyAuth(user_id=owner.user_id) if owner is not None else user_api_key_dict
+        receipt: Final = project_evaluation_billing_kwargs(
+            {
+                **request_receipt,
+                EVALUATION_BILLING_OWNER_KEY: owner,
+                EVALUATION_BUDGET_RESERVATION_KEY: logger.evaluation_budget_reservation if logger is not None else None,
+            }
+        )
 
         await self._spend_writer().update_database(
-            token=LiteLLMProxyRequestSetup.get_logged_api_key(user_api_key_dict),
+            token=LiteLLMProxyRequestSetup.get_logged_api_key(billing_auth),
             response_cost=recovered_response_cost,
-            user_id=user_api_key_dict.user_id,
-            end_user_id=user_api_key_dict.end_user_id,
-            team_id=user_api_key_dict.team_id,
-            kwargs=request_data,
+            user_id=billing_auth.user_id,
+            end_user_id=billing_auth.end_user_id,
+            team_id=billing_auth.team_id,
+            kwargs=receipt,
             completion_response=original_exception,
             start_time=actual_start_time,
             end_time=datetime.now(),
-            org_id=user_api_key_dict.org_id,
-            project_id=user_api_key_dict.project_id,
+            org_id=billing_auth.org_id,
+            project_id=billing_auth.project_id,
         )
 
     async def _PROXY_track_cost_callback(

litellm/proxy/management_endpoints/auto_router_endpoints.py

@@ -1684,10 +1684,10 @@ async def start_shadow_eval(
     eval spend, the shadow and judge calls' own cost, reaches max_budget dollars, the
     job's window ends, or the job is stopped, so one target running out of budget does
     not end sampling for the others; sampling changes propagate to pods within about 10
-    seconds. Shadow and judge calls bill to the sampled request's own identity but are
+    seconds. Shadow and judge calls bill to the admin who started the job and are
     excluded from request counts and auto-router adoption metrics.
     """
-    from litellm.proxy.proxy_server import llm_router, prisma_client
+    from litellm.proxy.proxy_server import litellm_proxy_admin_name, llm_router, prisma_client
 
     _require_admin_writer(user_api_key_dict, "start a shadow eval")
     if prisma_client is None:
@@ -1805,7 +1805,7 @@ async def start_shadow_eval(
         "shadow_percentage": data.shadow_percentage,
         "max_turns": SHADOW_EVAL_TURN_VALVE,
         "max_budget": data.max_budget,
-        "created_by": user_api_key_dict.user_id,
+        "created_by": user_api_key_dict.user_id or litellm_proxy_admin_name,
         "created_at": now,
         "ends_at": ends_at,
     }

litellm/proxy/native_compaction.py

@@ -13,6 +13,10 @@ from litellm.litellm_core_utils.initialize_dynamic_callback_params import (
     inherit_message_logging_privacy,
     initialize_standard_callback_dynamic_params,
 )
+from litellm.litellm_core_utils.internal_call_metadata import (
+    evaluation_billing_context,
+    get_evaluation_billing_owner,
+)
 from litellm.llms.custom_httpx.asgi_handler import get_async_asgi_client
 from litellm.proxy.litellm_pre_call_utils import UNTRUSTED_REQUEST_HEADER_CONTROL_FIELDS
 from litellm.router_strategy.complexity_router.context_compaction import (
@@ -40,6 +44,7 @@ async def with_proxy_compaction_executor(call: Awaitable[_ResultT], request: Req
         protocol: Literal["chat", "messages"], payload: Mapping[str, object], parent_model: str | None = None
     ) -> Mapping[str, object]:
         logging_disabled: Final = initialize_standard_callback_dynamic_params().get("turn_off_message_logging") is True
+        billing_owner: Final = get_evaluation_billing_owner()
 
         async def dispatch() -> Mapping[str, object]:
             scope: Final = _JSON_OBJECT.validate_python(request.scope)
@@ -55,6 +60,7 @@ async def with_proxy_compaction_executor(call: Awaitable[_ResultT], request: Req
             with (
                 native_compaction_call(parent_model, str(payload["model"])),
                 inherit_message_logging_privacy(logging_disabled),
+                evaluation_billing_context(billing_owner),
             ):
                 with get_async_asgi_client(
                     app=_ASGI_APP.validate_python(scope["app"]),

litellm/proxy/proxy_server.py

@@ -3376,15 +3376,21 @@ async def _reconcile_budget_reservation_for_counter_update(
     budget_reservation: dict | None,
     response_cost: float | None,
 ) -> _ReservationCounterUpdate:
-    if budget_reservation is None or budget_reservation.get("finalized") is True:
-        return _ReservationCounterUpdate()
-
     from litellm.proxy.spend_tracking.budget_reservation import (
         get_reserved_counter_keys,
         invalidate_budget_reservation_counters,
         reconcile_budget_reservation,
     )
 
+    if budget_reservation is not None and budget_reservation.get("externally_settled") is True:
+        return _ReservationCounterUpdate(
+            reserved_counter_keys=TypeAdapter(frozenset[str]).validate_python(
+                get_reserved_counter_keys(budget_reservation)
+            )
+        )
+    if budget_reservation is None or budget_reservation.get("finalized") is True:
+        return _ReservationCounterUpdate()
+
     reserved_counter_keys: Final = get_reserved_counter_keys(budget_reservation=budget_reservation)
     try:
         pending: Final = await reconcile_budget_reservation(

litellm/proxy/spend_tracking/budget_reservation.py

@@ -111,7 +111,11 @@ def get_reserved_counter_keys(budget_reservation: dict | None) -> set:
 _lease_renewals: Final[set[asyncio.Task[None]]] = set()  # mutable-ok: asyncio only weak-refs pending tasks
 
 
-def _start_reservation_lease_renewal(budget_reservation: Mapping[str, object], counter_keys: frozenset[str]) -> None:
+def _start_reservation_lease_renewal(
+    budget_reservation: Mapping[str, object],
+    counter_keys: frozenset[str],
+    request_task: asyncio.Task[object] | None = None,
+) -> None:
     """A reservation lives inside spend counter keys that expire on their Redis TTL. Renew the TTL
     while the request is in flight so a request longer than the TTL does not drop its
     reservation and admit concurrent requests against the DB floor on any worker."""
@@ -124,7 +128,7 @@ def _start_reservation_lease_renewal(budget_reservation: Mapping[str, object], c
             budget_reservation=budget_reservation,
             counter_keys=counter_keys,
             interval=spend_counter_cache.redis_cache.default_ttl / 2,
-            request_task=asyncio.current_task(),
+            request_task=request_task or asyncio.current_task(),
         )
     )
     _lease_renewals.add(task)
@@ -144,7 +148,11 @@ async def _renew_reservation_lease(
     deadline: Final = time.monotonic() + litellm.request_timeout
     while time.monotonic() < deadline:
         await asyncio.sleep(interval)
-        if budget_reservation.get("finalized") is True or (request_task is not None and request_task.done()):
+        if (
+            budget_reservation.get("finalized") is True
+            or budget_reservation.get("externally_settled") is True
+            or (request_task is not None and request_task.done())
+        ):
             return
         for counter_key in counter_keys:
             await refresh_spend_counter_ttl(counter_key=counter_key)
@@ -248,7 +256,7 @@ def _is_unbilled_route(route: str) -> bool:
 
 
 async def reserve_budget_for_request(
-    request_body: dict,
+    request_body: dict[str, object],  # mutable-ok: existing reservation helpers accept dictionaries
     route: str,
     llm_router: Router | None,
     valid_token: UserAPIKeyAuth | None,
@@ -262,7 +270,8 @@ async def reserve_budget_for_request(
     apply_user_budget_to_team_keys: bool = False,
     fail_closed_budget_enforcement: bool = False,
     raw_body: bytes | None = None,
-) -> dict | None:
+    request_task: asyncio.Task[object] | None = None,
+) -> dict[str, object] | None:  # mutable-ok: shared reservation is finalized by the spend writer
     if valid_token is None or not RouteChecks.is_llm_api_route(route=route):
         return None
     if _is_unbilled_route(route):
@@ -334,7 +343,7 @@ async def reserve_budget_for_request(
         llm_router=llm_router,
         input_token_counts=input_token_counts,
     )
-    budget_reservation: Final = {
+    budget_reservation: Final[dict[str, object]] = {  # mutable-ok: shared finalization state
         "reserved_cost": reservation_cost,
         "entries": applied_entries,
         "finalized": False,
@@ -345,6 +354,7 @@ async def reserve_budget_for_request(
     _start_reservation_lease_renewal(
         budget_reservation=budget_reservation,
         counter_keys=frozenset(get_reserved_counter_keys(budget_reservation=budget_reservation)),
+        request_task=request_task,
     )
     return budget_reservation
 
@@ -357,8 +367,13 @@ async def reconcile_budget_reservation(
 ) -> tuple[PendingSpendIncrement, ...]:
     """Settle every reserved counter on ``actual_cost``. With ``apply_consistent`` False the adjustments for
     counters that still hold the reservation are returned instead of written, so the caller can pipeline them with
-    its own increments and then call ``stamp_budget_reservation_actual_cost``."""
-    if not budget_reservation or budget_reservation.get("finalized") is True:
+    its own increments and then call ``stamp_budget_reservation_actual_cost``. External owners settle independently;
+    their entries remain available for the spend writer to skip already charged counters."""
+    if (
+        not budget_reservation
+        or budget_reservation.get("finalized") is True
+        or budget_reservation.get("externally_settled") is True
+    ):
         return ()
 
     reserved_cost: Final = float(budget_reservation.get("reserved_cost") or 0.0)
@@ -377,7 +392,7 @@ async def reconcile_budget_reservation(
 def stamp_budget_reservation_actual_cost(budget_reservation: dict | None, actual_cost: float | None) -> None:
     """Record that every reserved counter now holds ``actual_cost``, once the adjustments handed back by
     ``reconcile_budget_reservation(apply_consistent=False)`` have been written."""
-    if not budget_reservation:
+    if not budget_reservation or budget_reservation.get("externally_settled") is True:
         return
     reserved_cost: Final = float(budget_reservation.get("reserved_cost") or 0.0)
     actual: Final = float(actual_cost or 0.0)
@@ -1302,7 +1317,7 @@ def _coerce_datetime(value: object) -> datetime | None:
 
 
 def estimate_request_max_cost(
-    request_body: dict,
+    request_body: dict[str, object],  # mutable-ok: existing cost helpers accept dictionaries
     route: str,
     llm_router: Router | None,
     input_token_counts: Mapping[str, int] | None = None,
@@ -1324,7 +1339,7 @@ def estimate_request_max_cost(
 
 
 def estimate_request_input_cost(
-    request_body: dict,
+    request_body: dict[str, object],  # mutable-ok: existing cost helpers accept dictionaries
     route: str,
     llm_router: Router | None,
     input_token_counts: Mapping[str, int] | None = None,

litellm/proxy/spend_tracking/evaluation_budget.py

@@ -0,0 +1,383 @@
+from __future__ import annotations
+
+import asyncio
+import math
+import time
+import uuid
+from collections.abc import Awaitable, Mapping, Sequence
+from dataclasses import dataclass, field
+from functools import lru_cache
+from itertools import product
+from string import ascii_letters, digits
+from threading import Lock
+from typing import Final, Literal, Protocol, runtime_checkable
+
+from pydantic import ConfigDict, TypeAdapter
+from redis.crc import key_slot
+
+import litellm
+from litellm._internal_context import with_service_target
+from litellm._logging import verbose_proxy_logger
+from litellm.caching.caching import DualCache
+from litellm.litellm_core_utils.duration_parser import duration_in_seconds
+from litellm.litellm_core_utils.internal_call_metadata import EvaluationBillingOwner
+from litellm.proxy._types import Litellm_EntityType, LiteLLM_UserTable, UserAPIKeyAuth
+from litellm.proxy.hooks.model_max_budget_limiter import (
+    model_budget_spend_cache_key,
+    model_budget_start_time_cache_key,
+    resolve_model_budget,
+)
+from litellm.proxy.spend_tracking.budget_reservation import (
+    estimate_request_input_cost,
+    estimate_request_max_cost,
+    reconcile_budget_reservation,  # pyright: ignore[reportUnknownVariableType]  # legacy reservation entries are untyped
+    reserve_budget_for_request,
+)
+from litellm.router_utils.common_utils import resolve_model_group_alias
+from litellm.types.utils import API_ROUTE_TO_CALL_TYPES, CallTypes
+
+_NUMBER: Final = TypeAdapter(float)
+_MAPPING: Final = TypeAdapter(Mapping[str, object])
+_REQUEST: Final = TypeAdapter(dict[str, object])
+_HOLDS: Final = TypeAdapter(Mapping[str, float])
+_LEASE_SECONDS: Final = 60
+_LOCAL_LOCK: Final = Lock()
+_LEASES: Final[set[asyncio.Task[None]]] = set()  # mutable-ok: asyncio weakly references pending tasks
+_HOLD_SCRIPT: Final = """
+local actual = tonumber(redis.call('GET', KEYS[2]) or '0')
+if not actual then return redis.error_reply('Invalid model budget spend') end
+local clock = redis.call('TIME')
+local now = tonumber(clock[1]) + tonumber(clock[2]) / 1000000
+local total = 0
+for _, member in ipairs(redis.call('ZRANGEBYSCORE', KEYS[1], '(' .. now, '+inf')) do
+    if member ~= ARGV[2] then total = total + tonumber(string.match(member, ':([^:]+)$')) end
+end
+if ARGV[1] == 'renew' then
+    local expiry = redis.call('ZSCORE', KEYS[1], ARGV[2])
+    if not expiry or tonumber(expiry) <= now then return redis.error_reply('Evaluation reservation expired') end
+end
+if ARGV[1] == 'reserve' or ARGV[1] == 'renew' then
+    total = total + tonumber(string.match(ARGV[2], ':([^:]+)$'))
+end
+if ARGV[1] == 'reserve' and ARGV[6] ~= '' then
+    local estimate = tonumber(string.match(ARGV[2], ':([^:]+)$'))
+    if actual + total > tonumber(ARGV[6]) or actual + total - estimate >= tonumber(ARGV[6]) then
+        return tostring(actual + total)
+    end
+end
+if ARGV[1] == 'settle' and tonumber(ARGV[4]) > 0 then
+    actual = tonumber(redis.call('INCRBYFLOAT', KEYS[2], ARGV[4]))
+    redis.call('EXPIRE', KEYS[2], ARGV[5])
+end
+redis.call('ZREMRANGEBYSCORE', KEYS[1], '-inf', now)
+if ARGV[1] == 'reserve' or ARGV[1] == 'renew' then
+    redis.call('ZADD', KEYS[1], now + tonumber(ARGV[3]), ARGV[2])
+    redis.call('EXPIRE', KEYS[1], ARGV[3])
+elseif ARGV[1] == 'settle' then redis.call('ZREM', KEYS[1], ARGV[2]) end
+return tostring(actual + total)
+"""
+
+
+@runtime_checkable
+class _NumericCache(Protocol):
+    def get_cache(self, key: str) -> object: ...
+    def set_cache(self, key: str, value: object, *, ttl: int) -> object: ...
+    def delete_cache(self, key: str) -> object: ...
+    def increment_cache(self, key: str, value: float, *, ttl: int, refresh_ttl: bool = False) -> float: ...
+
+
+@runtime_checkable
+class _Script(Protocol):
+    async def __call__(self, *, keys: Sequence[str], args: Sequence[str | int | float]) -> object: ...
+
+
+_LOCAL: Final = TypeAdapter(_NumericCache, config=ConfigDict(arbitrary_types_allowed=True))
+_SCRIPT: Final = TypeAdapter(_Script, config=ConfigDict(arbitrary_types_allowed=True))
+
+
+@lru_cache(maxsize=4096)
+def _model_hold_key(effective_key: str) -> str:
+    prefix: Final = f"{effective_key}:evaluation_holds:"
+    tagged: Final = f"{prefix}{{{effective_key}}}"
+    slot: Final = key_slot(effective_key.encode())
+    if key_slot(tagged.encode()) == slot:
+        return tagged
+    candidates: Final = (prefix + "".join(chars) for chars in product(ascii_letters + digits + "-_", repeat=3))
+    return next(candidate for candidate in candidates if key_slot(candidate.encode()) == slot)
+
+
+@with_service_target("model_budgets")
+async def model_budget_spend(
+    cache: DualCache,
+    spend_key: str,
+    *,
+    operation: Literal["read", "reserve", "renew", "settle"] = "read",
+    member: str = "",
+    adjustment: float = 0.0,
+    ttl: int = 1,
+    limit: float | None = None,
+) -> float:
+    if cache.redis_cache is not None:
+        key: Final = cache.redis_cache.check_and_fix_namespace(spend_key)
+        script: Final = _SCRIPT.validate_python(cache.redis_cache.async_register_script(_HOLD_SCRIPT))
+        return _NUMBER.validate_python(
+            await script(
+                keys=(_model_hold_key(key), key),
+                args=(operation, member, _LEASE_SECONDS, adjustment, ttl, "" if limit is None else limit),
+            )
+        )
+    local: Final = _LOCAL.validate_python(cache.in_memory_cache)
+    hold_key: Final = _model_hold_key(spend_key)
+    with _LOCAL_LOCK:
+        actual: Final = _NUMBER.validate_python(local.get_cache(spend_key) or 0.0)
+        now: Final = cache.in_memory_cache._clock()  # pyright: ignore[reportPrivateUsage]  # lease expiry uses the cache's injected clock
+        active: Final = {
+            token: expiry
+            for token, expiry in _HOLDS.validate_python(local.get_cache(hold_key) or {}).items()
+            if expiry > now
+        }
+        if operation == "renew" and member not in active:
+            raise RuntimeError("Evaluation reservation expired")
+        updated: Final = {
+            **{token: expiry for token, expiry in active.items() if token != member},
+            **({member: now + _LEASE_SECONDS} if operation in ("reserve", "renew") else {}),
+        }
+        held: Final = sum(float(token.rsplit(":", 1)[1]) for token in updated)
+        proposed: Final = actual + held
+        if (
+            operation == "reserve"
+            and limit is not None
+            and (proposed > limit or proposed - float(member.rsplit(":", 1)[1]) >= limit)
+        ):
+            return proposed
+        settled: Final = (
+            local.increment_cache(spend_key, adjustment, ttl=ttl, refresh_ttl=True)
+            if operation == "settle" and adjustment
+            else actual
+        )
+        local.delete_cache(hold_key)
+        if updated:
+            local.set_cache(hold_key, updated, ttl=_LEASE_SECONDS)
+        return settled + held
+
+
+async def _complete(operation: Awaitable[EvaluationAttempt | None]) -> EvaluationAttempt | None:
+    task: Final = asyncio.ensure_future(operation)
+    try:
+        return await asyncio.shield(task)
+    except asyncio.CancelledError:
+        drained: Final = asyncio.gather(task, return_exceptions=True)
+        while not drained.done():
+            try:
+                await asyncio.shield(drained)
+            except asyncio.CancelledError:
+                continue
+        if not task.cancelled() and task.exception() is None and (attempt := task.result()) is not None:
+            await attempt.settle(0.0)
+        raise
+
+
+@dataclass(frozen=True, slots=True)
+class _ModelBudget:
+    spend_key: str
+    start_key: str
+    duration: int
+    limit: float
+
+
+@with_service_target("model_budgets")
+async def _model_window(cache: DualCache, model: _ModelBudget) -> int:
+    if cache.redis_cache is not None:
+        script: Final = _SCRIPT.validate_python(
+            cache.redis_cache.async_register_script(
+                "local clock = redis.call('TIME'); local now = tonumber(clock[1]) + tonumber(clock[2]) / 1000000; "
+                "local start = tonumber(redis.call('GET', KEYS[1]) or now); "
+                "if now - start >= tonumber(ARGV[1]) then start = now end; "
+                "local ttl = math.max(1, math.ceil(tonumber(ARGV[1]) - (now - start))); "
+                "redis.call('SET', KEYS[1], start, 'EX', ttl); return ttl"
+            )
+        )
+        return math.ceil(_NUMBER.validate_python(await script(keys=(model.start_key,), args=(model.duration,))))
+    local: Final = _LOCAL.validate_python(cache.in_memory_cache)
+    now: Final = time.time()
+    cached: Final = local.get_cache(model.start_key)
+    previous: Final = _NUMBER.validate_python(now if cached is None else cached)
+    start: Final = now if now - previous >= model.duration else previous
+    ttl: Final = max(1, math.ceil(model.duration - (now - start)))
+    local.delete_cache(model.start_key)
+    local.set_cache(model.start_key, start, ttl=ttl)
+    return ttl
+
+
+@dataclass(slots=True)
+class EvaluationAttempt:
+    cache: DualCache
+    model: _ModelBudget | None
+    total: dict[str, object] | None = None  # mutable-ok: existing spend writer owns this reservation format
+    member: str = ""
+    input_cost: float = 0.0
+    known_cost: float = 0.0
+    model_applied_cost: float | None = None
+    lock: asyncio.Lock = field(default_factory=asyncio.Lock)
+
+    async def settle(self, cost: float, cancelled: bool = False) -> None:
+        await _complete(self._settle(max(cost, self.input_cost if cancelled else 0.0)))
+
+    async def _settle(self, cost: float) -> None:
+        async with self.lock:
+            self.known_cost = max(cost, self.known_cost)
+            try:
+                await reconcile_budget_reservation(
+                    {**self.total, "finalized": False, "externally_settled": False} if self.total is not None else None,
+                    self.known_cost,
+                    finalize=False,
+                )
+                if self.total is not None:
+                    self.total["externally_settled"] = True
+            finally:
+                if self.model is not None:
+                    delta: Final = max(self.known_cost - (self.model_applied_cost or 0.0), 0.0)
+                    ttl: Final = await _model_window(self.cache, self.model) if delta else 1
+                    await model_budget_spend(
+                        self.cache,
+                        self.model.spend_key,
+                        operation="settle",
+                        member=self.member,
+                        adjustment=delta,
+                        ttl=ttl,
+                    )
+                    self.model_applied_cost = self.known_cost
+
+    async def renew(self, request_task: asyncio.Task[object] | None) -> None:
+        deadline: Final = time.monotonic() + 2 * litellm.request_timeout
+        while self.model_applied_cost is None and self.model is not None:
+            await asyncio.sleep(_LEASE_SECONDS / 2)
+            async with self.lock:
+                if self.model_applied_cost is not None:
+                    return
+                try:
+                    if time.monotonic() >= deadline:
+                        raise TimeoutError("Evaluation reservation exceeded request timeout")
+                    await model_budget_spend(self.cache, self.model.spend_key, operation="renew", member=self.member)
+                except Exception:  # noqa: BLE001  # an active call must stop when its budget hold cannot be renewed
+                    verbose_proxy_logger.exception("Unable to renew evaluation budget reservation")
+                    if request_task is not None and not request_task.done():
+                        request_task.cancel()
+                    return
+
+
+async def reserve_evaluation_budget(
+    owner: EvaluationBillingOwner,
+    request: Mapping[str, object],
+    call_type: str,
+) -> EvaluationAttempt | None:
+    return await _complete(_reserve(owner, request, call_type, asyncio.current_task()))
+
+
+async def _reserve(
+    owner: EvaluationBillingOwner,
+    request: Mapping[str, object],
+    call_type: str,
+    request_task: asyncio.Task[object] | None,
+) -> EvaluationAttempt | None:
+    from litellm.proxy import proxy_server as proxy
+
+    llm_router: Final = proxy.llm_router
+    metadata: Final = _MAPPING.validate_python(request.get("litellm_metadata") or request.get("metadata") or {})
+    logical_model: Final = str(metadata.get("model_group") or request["model"])
+    resolved: Final = resolve_model_budget(logical_model, owner.user_model_max_budget or {})
+    model: Final = (
+        _ModelBudget(
+            model_budget_spend_cache_key(
+                Litellm_EntityType.USER, owner.user_id, resolved.budget_model, resolved.budget_config.budget_duration
+            ),
+            model_budget_start_time_cache_key(
+                Litellm_EntityType.USER, owner.user_id, resolved.budget_model, resolved.budget_config.budget_duration
+            ),
+            duration_in_seconds(str(resolved.budget_config.budget_duration)),
+            resolved.budget_config.max_budget,
+        )
+        if resolved is not None
+        and resolved.budget_config.max_budget is not None
+        and math.isfinite(resolved.budget_config.max_budget)
+        and resolved.budget_config.max_budget >= 0
+        else None
+    )
+    total_budget: Final = owner.max_budget is not None and math.isfinite(owner.max_budget)
+    if not total_budget and model is None:
+        return None
+    if total_budget:
+        current: Final = await proxy.get_current_spend(
+            counter_key=f"spend:user:{owner.user_id}", fallback_spend=owner.spend, max_budget=owner.max_budget
+        )
+        if current >= _NUMBER.validate_python(owner.max_budget):
+            raise litellm.BudgetExceededError(
+                current_cost=current,
+                max_budget=_NUMBER.validate_python(owner.max_budget),
+                entity_type=Litellm_EntityType.USER.value,
+                entity_id=owner.user_id,
+            )
+    route: Final = next(route for route, types in API_ROUTE_TO_CALL_TYPES.items() if CallTypes(call_type) in types)
+    model_info: Final = _MAPPING.validate_python(request.get("model_info") or metadata.get("model_info") or {})
+    deployment_id: Final = model_info.get("id")
+    deployment: Final = (
+        llm_router.get_deployment(deployment_id) if llm_router is not None and isinstance(deployment_id, str) else None
+    )
+    pricing_model: Final = (
+        deployment.model_name
+        if deployment is not None
+        else (
+            resolve_model_group_alias(llm_router.model_group_alias, logical_model) or logical_model
+            if llm_router is not None
+            else logical_model
+        )
+    )
+    body: Final = _REQUEST.validate_python(
+        {**request, "model": pricing_model, "metadata": {}, "litellm_metadata": {}, "tags": []}
+    )
+    attempt: Final = EvaluationAttempt(cache=proxy.model_max_budget_limiter.dual_cache, model=model)
+    try:
+        attempt.total = await reserve_budget_for_request(
+            request_body=body,
+            route=route,
+            llm_router=llm_router,
+            valid_token=UserAPIKeyAuth(user_id=owner.user_id),
+            team_object=None,
+            user_object=LiteLLM_UserTable(user_id=owner.user_id, max_budget=owner.max_budget, spend=owner.spend),
+            prisma_client=proxy.prisma_client,
+            user_api_key_cache=proxy.user_api_key_cache,
+            proxy_logging_obj=proxy.proxy_logging_obj,
+            fail_closed_budget_enforcement=True,
+            request_task=request_task,
+        )
+        estimate: Final = (
+            _NUMBER.validate_python(attempt.total["reserved_cost"])
+            if attempt.total is not None
+            else estimate_request_max_cost(body, route, llm_router)
+        )
+        if estimate is None or not math.isfinite(estimate) or estimate < 0:
+            raise ValueError("Evaluation budget cannot be checked for an unpriced model")
+        attempt.member = f"{uuid.uuid4()}:{estimate}"
+        attempt.input_cost = (
+            _NUMBER.validate_python(attempt.total["input_cost"])
+            if attempt.total is not None
+            else estimate_request_input_cost(body, route, llm_router) or 0.0
+        )
+        if model is not None:
+            spend: Final = await model_budget_spend(
+                attempt.cache, model.spend_key, operation="reserve", member=attempt.member, limit=model.limit
+            )
+            if spend > model.limit or spend - estimate >= model.limit:
+                raise litellm.BudgetExceededError(
+                    current_cost=spend - estimate,
+                    max_budget=model.limit,
+                    entity_type=Litellm_EntityType.USER.value,
+                    entity_id=owner.user_id,
+                )
+            lease: Final = asyncio.create_task(attempt.renew(request_task))
+            _LEASES.add(lease)
+            lease.add_done_callback(_LEASES.discard)
+    except Exception:
+        await attempt.settle(0.0)
+        raise
+    return attempt

litellm/router_strategy/budget_limiter.py

@@ -21,6 +21,7 @@ anthropic:
 import asyncio
 import builtins
 import logging
+import math
 from collections.abc import Mapping, Sequence
 from datetime import datetime, timedelta, timezone
 from itertools import groupby
@@ -590,7 +591,7 @@ class RouterBudgetLimiting(CustomLogger):
         else:
             # Within existing window - increment spend
             remaining_time: Final = ttl_seconds - (current_time - budget_start)
-            ttl_for_increment: Final = int(remaining_time)
+            ttl_for_increment: Final = max(1, math.ceil(remaining_time))
 
             await self._increment_spend_in_current_window(
                 spend_key=spend_key, response_cost=response_cost, ttl=ttl_for_increment

litellm/utils.py

@@ -45,7 +45,7 @@ from httpx import Proxy
 from httpx._utils import get_environment_proxies
 from openai.lib import _parsing, _pydantic
 from openai.types.chat.completion_create_params import ResponseFormat
-from pydantic import BaseModel
+from pydantic import BaseModel, TypeAdapter
 
 import litellm
 import litellm.litellm_core_utils
@@ -1379,11 +1379,14 @@ def _schedule_async_success_logging(
     first-wins rule: the innermost wrapper's provider-shaped result is the one the spend log
     reads usage from, and a later wrapper never swaps in its client-shaped translation.
     """
+    from litellm.litellm_core_utils.litellm_logging import evaluation_logging_snapshot
+
+    receipt_logger: Final = evaluation_logging_snapshot(logging_obj)
 
     def _enqueue_async_logging() -> None:
         asyncio.create_task(
             _client_async_logging_helper(
-                logging_obj=logging_obj,
+                logging_obj=receipt_logger,
                 result=result,
                 start_time=start_time,
                 end_time=end_time,
@@ -1965,11 +1968,26 @@ def client(original_function):
 
     @wraps(original_function)
     async def wrapper_async(*args, **kwargs):
+        from litellm.litellm_core_utils.internal_call_metadata import (
+            EVALUATION_BUDGET_RESERVATION_KEY,
+            EvaluationBillingOwner,
+            get_evaluation_billing_owner,
+        )
+        from litellm.litellm_core_utils.litellm_logging import EvaluationBudgetInvocation, Logging
+        from litellm.llms.anthropic.pass_through.messages.handler import anthropic_messages_uses_bridge
+
         print_args_passed_to_litellm(original_function, args, kwargs)
         start_time: Final = datetime.datetime.now()
         result = None
         _update_response_metadata: Final[_ResponseMetadataUpdater] = litellm_utils.update_response_metadata
         logging_obj: LiteLLMLoggingObject | None = kwargs.get("litellm_logging_obj", None)
+        evaluation_invocation: Final = (
+            EvaluationBudgetInvocation()
+            if get_evaluation_billing_owner() is not None
+            or isinstance(logging_obj, Logging)
+            and logging_obj.evaluation_billing_owner is not None
+            else None
+        )
         LLMCachingHandler: Final = _get_cached_llm_caching_handler()
         _llm_caching_handler: Final[LLMCachingHandler] = LLMCachingHandler(
             original_function=original_function,
@@ -1982,7 +2000,21 @@ def client(original_function):
             kwargs["litellm_call_id"] = str(uuid.uuid4())
 
         model: Final[str | None] = args[0] if len(args) > 0 else kwargs.get("model", None)
-        is_completion_with_fallbacks: Final = kwargs.get("fallbacks") is not None
+        is_evaluation_bridge: Final = (
+            evaluation_invocation is not None
+            and TypeAdapter(str).validate_python(call_type) == CallTypes.anthropic_messages.value
+            and anthropic_messages_uses_bridge(
+                TypeAdapter(str).validate_python(model or ""), TypeAdapter(Mapping[str, object]).validate_python(kwargs)
+            )
+        )
+        is_completion_with_fallbacks: Final = call_type in (
+            CallTypes.acompletion.value,
+            CallTypes.atext_completion.value,
+        ) and bool(
+            TypeAdapter(object).validate_python(
+                kwargs.get("fallbacks") if kwargs.get("fallbacks") is not None else litellm.model_fallbacks
+            )
+        )
         kwargs.pop("_is_litellm_internal_call", None)  # discard if injected
         _is_litellm_internal_call: Final = is_internal_call.get()
         _deployment_call_end_time: datetime.datetime | None = None
@@ -1993,6 +2025,36 @@ def client(original_function):
 
             # Type assertion: logging_obj is guaranteed to be non-None after function_setup
             assert logging_obj is not None, "logging_obj should not be None after function_setup"
+            if (
+                evaluation_invocation is not None
+                and isinstance(logging_obj, Logging)
+                and isinstance(logging_obj.evaluation_billing_owner, EvaluationBillingOwner)
+                and not _is_litellm_internal_call
+                and logging_obj.evaluation_budget_invocation is None
+            ):
+                if not is_completion_with_fallbacks and not is_evaluation_bridge:
+                    logging_obj.evaluation_budget_invocation = evaluation_invocation
+                logging_obj.evaluation_budget_reservation = None
+                prior_receipt: Final = TypeAdapter(Mapping[str, object]).validate_python(logging_obj.model_call_details)
+                logging_obj.model_call_details = {
+                    key: value
+                    for key, value in prior_receipt.items()
+                    if not key.startswith("has_logged_")
+                    and key not in ("response_cost", "standard_logging_object", "combined_usage_object", "cache_hit")
+                }
+                logging_obj.model_call_details[EVALUATION_BUDGET_RESERVATION_KEY] = None
+            if (
+                evaluation_invocation is not None
+                and not _is_litellm_internal_call
+                and _is_streaming_request(
+                    TypeAdapter(dict[str, object]).validate_python(kwargs), TypeAdapter(str).validate_python(call_type)
+                )
+            ):
+                raise litellm.BadRequestError(
+                    message="Shadow evaluation requires a non-streaming response",
+                    model=TypeAdapter(str).validate_python(model or ""),
+                    llm_provider="",
+                )
             if not _is_litellm_internal_call:
                 bind_budget_reservation_to_callbacks(logging_obj.litellm_params)
 
@@ -2089,6 +2151,30 @@ def client(original_function):
                 and _caching_handler_response.embedding_uncached_input is not None
                 else kwargs
             )
+            streaming_requested: Final = _is_streaming_request(kwargs=kwargs, call_type=call_type)
+            if (
+                evaluation_invocation is not None
+                and isinstance(logging_obj, Logging)
+                and isinstance(logging_obj.evaluation_billing_owner, EvaluationBillingOwner)
+                and logging_obj.evaluation_budget_invocation is evaluation_invocation
+            ):
+                if streaming_requested:
+                    raise litellm.BadRequestError(
+                        message="Shadow evaluation requires a non-streaming response",
+                        model=model or "",
+                        llm_provider="",
+                    )
+                from litellm.proxy.spend_tracking.evaluation_budget import reserve_evaluation_budget
+
+                evaluation_invocation.reservation = await reserve_evaluation_budget(
+                    logging_obj.evaluation_billing_owner,
+                    TypeAdapter(dict[str, object]).validate_python(
+                        {**call_kwargs, "model": model, "messages": logging_obj.messages}
+                    ),
+                    TypeAdapter(str).validate_python(call_type),
+                )
+                logging_obj.evaluation_budget_reservation = evaluation_invocation.reservation
+                logging_obj.model_call_details[EVALUATION_BUDGET_RESERVATION_KEY] = evaluation_invocation.reservation
             try:
                 result = await original_function(*args, **call_kwargs)
             except Exception as deployment_error:
@@ -2104,7 +2190,6 @@ def client(original_function):
                 raise
             end_time = datetime.datetime.now()
 
-            streaming_requested: Final = _is_streaming_request(kwargs=kwargs, call_type=call_type)
             if streaming_requested or _is_converted_stream_result(result):
                 logging_obj.stream = True
                 logging_obj.model_call_details["stream"] = True
@@ -2168,7 +2253,7 @@ def client(original_function):
                     result=result,
                     start_time=start_time,
                     end_time=end_time,
-                    is_completion_with_fallbacks=is_completion_with_fallbacks,
+                    is_completion_with_fallbacks=is_completion_with_fallbacks or is_evaluation_bridge,
                     is_litellm_internal_call=_is_litellm_internal_call,
                 )
                 return _llm_caching_handler._combine_cached_embedding_response_with_api_result(
@@ -2191,12 +2276,24 @@ def client(original_function):
                 result=result,
                 start_time=start_time,
                 end_time=end_time,
-                is_completion_with_fallbacks=is_completion_with_fallbacks,
+                is_completion_with_fallbacks=is_completion_with_fallbacks or is_evaluation_bridge,
                 is_litellm_internal_call=_is_litellm_internal_call,
             )
 
             return result
-        except Exception as e:
+        except (Exception, asyncio.CancelledError) as e:
+            if (
+                evaluation_invocation is not None
+                and isinstance(logging_obj, Logging)
+                and logging_obj.evaluation_budget_invocation is evaluation_invocation
+                and evaluation_invocation.reservation is not None
+            ):
+                await evaluation_invocation.reservation.settle(
+                    logging_obj.recover_failure_cost(TypeAdapter(object).validate_python(result)),
+                    cancelled=isinstance(e, asyncio.CancelledError),
+                )
+            if isinstance(e, asyncio.CancelledError):
+                raise
             traceback_exception: Final = traceback.format_exc()
             # Reuse the timestamp taken right when the deployment call itself failed, before
             # the failure hook ran, so a slow callback doesn't inflate the reported duration.
@@ -2216,7 +2313,10 @@ def client(original_function):
 
             call_type = original_function.__name__
             num_retries, kwargs = _get_wrapper_num_retries(kwargs=kwargs, exception=e)
-            if call_type == CallTypes.acompletion.value:
+            sdk_retries_enabled: Final = not isinstance(logging_obj, Logging) or not isinstance(
+                logging_obj.evaluation_billing_owner, EvaluationBillingOwner
+            )
+            if call_type == CallTypes.acompletion.value and sdk_retries_enabled:
                 context_window_fallback_dict: Final = kwargs.get("context_window_fallback_dict", {})
 
                 _is_litellm_router_call = "model_group" in (
@@ -2251,7 +2351,7 @@ def client(original_function):
                         kwargs["model"] = context_window_fallback_dict[model]
                     result = await original_function(*args, **kwargs)
                     return result
-            elif call_type == CallTypes.aresponses.value:
+            elif call_type == CallTypes.aresponses.value and sdk_retries_enabled:
                 _is_litellm_router_call = "model_group" in (
                     kwargs.get("metadata") or {}
                 )  # check if call from litellm.router/proxy
@@ -2282,6 +2382,12 @@ def client(original_function):
             raise e
 
         finally:
+            if (
+                evaluation_invocation is not None
+                and isinstance(logging_obj, Logging)
+                and logging_obj.evaluation_budget_invocation is evaluation_invocation
+            ):
+                logging_obj.evaluation_budget_invocation = None
             # Restore trace_id/session_id contextvars to their pre-call value once
             # this call (in this asyncio Task) is fully done - see
             # request_correlation_in_logs. Unlike wrapper()'s sync path, it's safe to

tests/unit/conftest.py

@@ -45,6 +45,7 @@ import litellm.router as litellm_router_module  # noqa: E402  # same import-time
 import litellm.utils as litellm_utils_module  # noqa: E402  # same import-time dependency
 from litellm._logging import ALL_LOGGERS  # noqa: E402  # same import-time dependency
 from litellm.anthropic_beta_headers_manager import reload_beta_headers_config  # noqa: E402  # same import-time dependency
+from litellm.caching.caching import DualCache  # noqa: E402  # same import-time dependency
 from litellm.litellm_core_utils.prompt_templates import factory as prompt_factory_module  # noqa: E402  # same import-time dependency
 from litellm.litellm_core_utils.prompt_templates import (  # noqa: E402  # same import-time dependency
     image_handling as image_handling_module,
@@ -332,3 +333,20 @@ def pytest_sessionfinish() -> None:
         _close_handler_if_needed(getattr(litellm, name, None))
     _run_coroutine_if_needed(close_litellm_async_clients())
     enable_socket()
+
+
+@pytest.fixture
+def evaluation_spend_cache(monkeypatch: pytest.MonkeyPatch) -> DualCache:
+    from litellm.proxy import proxy_server
+    from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
+    from litellm.proxy.hooks.model_max_budget_limiter import _PROXY_VirtualKeyModelMaxBudgetLimiter
+
+    cache: Final = DualCache()
+    monkeypatch.setattr(proxy_server, "spend_counter_cache", cache)
+    monkeypatch.setattr(proxy_server, "user_api_key_cache", UserApiKeyCache())
+    monkeypatch.setattr(proxy_server, "prisma_client", None)
+    monkeypatch.setattr(proxy_server, "general_settings", {})
+    monkeypatch.setattr(proxy_server, "llm_router", None)
+    monkeypatch.setattr(proxy_server, "model_max_budget_limiter", _PROXY_VirtualKeyModelMaxBudgetLimiter(cache))
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    return cache

tests/unit/integrations/test_lago.py

@@ -0,0 +1,43 @@
+import json
+from datetime import datetime
+from typing import Final
+
+import pytest
+import respx
+
+from litellm.integrations.lago import LagoLogger
+from litellm.litellm_core_utils.internal_call_metadata import (
+    EVALUATION_BILLING_OWNER_KEY,
+    EvaluationBillingOwner,
+    project_evaluation_billing_kwargs,
+)
+
+
+@pytest.mark.parametrize("charge_by", ("user_id", "team_id", "end_user_id"))
+@pytest.mark.parametrize("evaluation", (False, True))
+def test_evaluation_receipts_charge_the_creator_regardless_of_lago_charge_by(
+    monkeypatch: pytest.MonkeyPatch, charge_by: str, evaluation: bool
+) -> None:
+    for name, value in {
+        "LAGO_API_KEY": "test",
+        "LAGO_API_BASE": "https://lago.invalid",
+        "LAGO_API_EVENT_CODE": "usage",
+        "LAGO_API_CHARGE_BY": charge_by,
+    }.items():
+        monkeypatch.setenv(name, value)
+    receipt: Final = project_evaluation_billing_kwargs(
+        {
+            EVALUATION_BILLING_OWNER_KEY: EvaluationBillingOwner("creator") if evaluation else None,
+            "response_cost": 0.25,
+            "litellm_params": {
+                "metadata": {"user_api_key_user_id": "user_id", "user_api_key_team_id": "team_id"},
+                "proxy_server_request": {"body": {"user": "end_user_id"}},
+            },
+        }
+    )
+    with respx.mock(assert_all_called=True) as transport:
+        endpoint: Final = transport.post("https://lago.invalid/api/v1/events").respond(200)
+        LagoLogger().log_success_event(receipt, {}, datetime(2026, 1, 1), datetime(2026, 1, 1))
+        sent: Final = json.loads(endpoint.calls.last.request.content)["event"]
+    assert sent["external_subscription_id"] == ("creator" if evaluation else charge_by)
+    assert sent["properties"]["response_cost"] == 0.25

tests/unit/integrations/test_shadow_eval_logger.py

@@ -11,7 +11,7 @@ import pytest
 from pydantic import ValidationError
 
 from litellm.caching.in_memory_cache import InMemoryCache
-from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY
+from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY, LITELLM_PROXY_ADMIN_NAME
 from litellm.integrations.shadow_eval_logger import (
     _MAX_CONCURRENT_SHADOW_TASKS,
     _MAX_ERROR_CHARS,
@@ -27,6 +27,7 @@ from litellm.integrations.shadow_eval_logger import (
     _unmask_preference,
     request_guardrail_fingerprint,
 )
+from litellm.proxy._types import LiteLLM_UserTable
 from litellm.types.guardrails import GuardrailEventHooks
 from litellm.types.utils import (
     SHADOW_EVAL_JUDGE_CALL_ORIGIN,
@@ -52,6 +53,7 @@ def _job(**overrides) -> ActiveShadowEvalJob:
         router_name="my-router",
         shadow_percentage=100.0,
         judge_model="judge-model",
+        created_by="evaluation-admin",
         max_turns=200,
         ends_at=datetime.now(timezone.utc) + timedelta(days=1),
         attempts=0,
@@ -74,6 +76,7 @@ def _prisma(jobs=(), attempt_counts=(), attempt_costs=()) -> MagicMock:
         ]
     )
     prisma.db.litellm_shadowevalattempt.create = AsyncMock()
+    prisma.db.litellm_usertable.find_unique = AsyncMock(return_value=LiteLLM_UserTable(user_id="evaluation-admin"))
     return prisma
 
 
@@ -90,6 +93,7 @@ def _job_record(job: ActiveShadowEvalJob, target_type="key", target_id="key-hash
         baseline_model=job.baseline_model,
         shadow_percentage=job.shadow_percentage,
         judge_model=job.judge_model,
+        created_by=job.created_by,
         max_turns=job.max_turns,
         max_budget=job.max_budget,
         ends_at=job.ends_at,
@@ -402,7 +406,7 @@ class TestSurfaceNormalization:
         assert shadow_call["tools"][0]["type"] == "function"
         assert shadow_call["tools"][0]["function"]["name"] == "get_weather"
         assert "stop_sequences" not in shadow_call
-        assert "stream" not in shadow_call
+        assert shadow_call["stream"] is False
         assert shadow_call["metadata"][INTERNAL_CALL_ORIGIN_METADATA_KEY] == SHADOW_EVAL_ROUTER_CALL_ORIGIN
 
     async def test_responses_arm_translates_wire_body_params_and_drops_surface_only_keys(self):
@@ -1029,7 +1033,7 @@ class TestSuccessHookSkipChain:
 
         shadow_call = router.acompletion.call_args_list[0].kwargs
         assert shadow_call["temperature"] == 0.5
-        assert "stream" not in shadow_call
+        assert shadow_call["stream"] is False
         create = prisma.db.litellm_shadowevalattempt.create
         create.assert_awaited_once()
         row = create.call_args.kwargs["data"]
@@ -1993,6 +1997,7 @@ class TestShadowPipeline:
         shadow_call = router.acompletion.call_args_list[0].kwargs
         judge_call = router.acompletion.call_args_list[1].kwargs
         for call in (shadow_call, judge_call):
+            assert call["stream"] is False
             assert call["num_retries"] == 0
             assert call["fallbacks"] == []
             assert call["metadata"]["user_api_key_hash"] == "key-hash"
@@ -2521,3 +2526,29 @@ class TestSamplingFunnel:
 
         assert logger._test_funnel == []
         prisma.db.litellm_shadowevalattempt.create.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("evaluation_spend_cache")
+@pytest.mark.parametrize(
+    ("created_by", "failure"),
+    (
+        ("missing-creator", None),
+        ("unreadable-creator", RuntimeError("creator unavailable")),
+        (LITELLM_PROXY_ADMIN_NAME, None),
+        (None, None),
+    ),
+)
+async def test_unavailable_evaluation_creator_withholds_without_calling_a_provider(
+    created_by: str | None, failure: Exception | None
+) -> None:
+    prisma: Final = _prisma()
+    prisma.db.litellm_usertable.find_unique = AsyncMock(return_value=None, side_effect=failure)
+    router: Final = _router()
+    job: Final = _job(created_by=created_by)
+    logger: Final = _logger(router=router, prisma=prisma, jobs=(job,))
+    await logger.async_log_success_event(_success_kwargs(), RESPONSE, None, None)
+    await _drain(logger)
+    assert router.acompletion.call_count == 0
+    assert logger._test_funnel == [("job-1", "withheld")]
+    prisma.db.litellm_shadowevalattempt.create.assert_not_awaited()

tests/unit/litellm_core_utils/test_litellm_logging.py

@@ -9165,3 +9165,228 @@ def test_signoz_dispatch_requires_an_endpoint(monkeypatch):
         logging_module._in_memory_loggers.clear()
         monkeypatch.delenv("LITELLM_OTEL_V2", raising=False)
         is_otel_v2_enabled.cache_clear()
+
+
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("evaluation_spend_cache")
+@pytest.mark.parametrize("dispatch", ("cache", "success", "failure"))
+async def test_queued_evaluation_callbacks_keep_their_receipt_after_logger_reuse(
+    monkeypatch: pytest.MonkeyPatch, dispatch: Literal["cache", "success", "failure"]
+) -> None:
+    from concurrent.futures import ThreadPoolExecutor
+    from queue import Queue
+    from threading import Event
+
+    from pydantic import TypeAdapter
+
+    from litellm.integrations.s3 import S3Logger
+    from litellm.litellm_core_utils import litellm_logging, thread_pool_executor
+    from litellm.litellm_core_utils.internal_call_metadata import EvaluationBillingOwner, evaluation_billing_context
+    from litellm.types.utils import Usage
+
+    receipts: Final = Queue[Mapping[str, object]]()
+    payload: Final = TypeAdapter(Mapping[str, object])
+
+    def upload(**request: object) -> None:
+        body: Final = request["Body"]
+        assert isinstance(body, str)
+        receipts.put_nowait(payload.validate_json(body))
+
+    def failure(kwargs: Mapping[str, object], response: object, start: object, end: object) -> None:
+        receipts.put_nowait(payload.validate_python(kwargs["standard_logging_object"]))
+
+    with patch("boto3.client", return_value=MagicMock(put_object=upload)):
+        sink: Final = S3Logger(s3_bucket_name="evaluation-receipts")
+    monkeypatch.setattr(litellm_logging, "s3Logger", sink)
+    monkeypatch.setattr(litellm, "success_callback", ["s3"])
+    monkeypatch.setattr(litellm, "failure_callback", [failure])
+    now: Final = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
+    result: Final = ModelResponse(
+        model="openai/evaluation-receipt-model", usage=Usage(prompt_tokens=1, completion_tokens=1, total_tokens=2)
+    )
+    with evaluation_billing_context(EvaluationBillingOwner("creator")):
+        logger: Final = LitellmLogging(
+            model="openai/evaluation-receipt-model",
+            messages=[{"role": "user", "content": "hello"}],
+            stream=False,
+            call_type="acompletion",
+            start_time=now,
+            litellm_call_id="first-call",
+            function_id="first-call",
+        )
+    logger.update_environment_variables(
+        litellm_params={"acompletion": True, "metadata": {"model_group": "first-group"}}, optional_params={}
+    )
+    logger.record_partial_usage_for_failure(Usage(prompt_tokens=1, completion_tokens=1, total_tokens=2), 0.3)
+    release: Final = Event()
+    with ThreadPoolExecutor(max_workers=1) as worker:
+        worker.submit(release.wait)
+        monkeypatch.setattr(litellm_logging, "executor", worker)
+        monkeypatch.setattr(thread_pool_executor, "executor", worker)
+        try:
+            if dispatch == "cache":
+                logger.handle_sync_success_callbacks_for_async_calls(result, now, now, cache_hit=True)
+            elif dispatch == "success":
+                await logger.dispatch_success_handlers(result, start_time=now, end_time=now, cache_hit=False)
+            else:
+                await logger.dispatch_failure_handlers(RuntimeError("first failure"), "first failure")
+            later_params: Final = {"acompletion": True, "metadata": {"model_group": "later-group"}}
+            later: Final = {
+                **logger.model_call_details,
+                "model": "later-model",
+                "response_cost": 9.0,
+                "cache_hit": False,
+                "exception": RuntimeError("later failure"),
+                "litellm_params": later_params,
+            }
+            logger.model_call_details = later
+            logger.model = "later-model"
+            logger.litellm_params = later_params
+            untouched: Final = later.copy()
+        finally:
+            release.set()
+    assert receipts.qsize() == 1
+    receipt: Final = receipts.get_nowait()
+    assert receipt["model_group"] == "first-group"
+    assert receipt["response_cost"] == pytest.approx(0.0 if dispatch == "cache" else 0.3)
+    assert receipt["cache_hit"] is (dispatch == "cache")
+    assert payload.validate_python(receipt["metadata"])["user_api_key_user_id"] == "creator"
+    assert receipt["error_str"] == ("first failure" if dispatch == "failure" else None)
+    assert logger.model_call_details == untouched
+
+
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("evaluation_spend_cache")
+@pytest.mark.parametrize("api", ("chat", "messages"))
+async def test_deferred_evaluation_receipts_preserve_routing_metadata_on_real_router_reuse(
+    monkeypatch: pytest.MonkeyPatch, api: Literal["chat", "messages"]
+) -> None:
+    from concurrent.futures import ThreadPoolExecutor
+    from queue import Queue
+    from threading import Event
+
+    import respx
+    from pydantic import TypeAdapter
+
+    from litellm.integrations.s3 import S3Logger
+    from litellm.litellm_core_utils import litellm_logging, thread_pool_executor
+    from litellm.litellm_core_utils.internal_call_metadata import EvaluationBillingOwner, evaluation_billing_context
+    from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
+    from litellm.proxy import proxy_server
+    from litellm.types.utils import Usage
+
+    sync_receipts: Final = Queue[Mapping[str, object]]()
+    async_receipts: Final = asyncio.Queue[Mapping[str, object]]()
+    payload: Final = TypeAdapter(Mapping[str, object])
+
+    def upload(**request: object) -> None:
+        body: Final = request["Body"]
+        assert isinstance(body, str)
+        sync_receipts.put_nowait(payload.validate_json(body))
+
+    async def capture(kwargs: Mapping[str, object], response: object, start: object, end: object) -> None:
+        async_receipts.put_nowait(payload.validate_python(kwargs["standard_logging_object"]))
+
+    with patch("boto3.client", return_value=MagicMock(put_object=upload)):
+        sink: Final = S3Logger(s3_bucket_name="evaluation-receipts")
+    monkeypatch.setattr(litellm_logging, "s3Logger", sink)
+    monkeypatch.setattr(litellm, "success_callback", ["s3"])
+    model: Final = "hosted_vllm/evaluation-receipt"
+    monkeypatch.setitem(
+        litellm.model_cost,
+        model,
+        {
+            "input_cost_per_token": 0.001,
+            "output_cost_per_token": 0.002,
+            "max_input_tokens": 1000,
+            "max_output_tokens": 1000,
+            "litellm_provider": "hosted_vllm",
+            "mode": "chat",
+        },
+    )
+    router: Final = litellm.Router(
+        model_list=[
+            {
+                "model_name": group,
+                "litellm_params": {
+                    "model": model,
+                    "api_key": "test",
+                    "api_base": "https://evaluation.invalid/v1",
+                    "input_cost_per_token": input_cost,
+                    "output_cost_per_token": output_cost,
+                },
+                "model_info": {"id": group + "-id"},
+            }
+            for group, input_cost, output_cost in (("first-group", 0.001, 0.002), ("later-group", 0.003, 0.004))
+        ],
+        num_retries=0,
+    )
+    monkeypatch.setattr(proxy_server, "llm_router", router)
+    create: Final = router.acompletion if api == "chat" else router.aanthropic_messages
+    metadata: Final = {"user_api_key_user_id": "caller", "user_api_key_team_id": "caller-team"}
+    release: Final = Event()
+    with (
+        respx.mock(assert_all_called=True) as transport,
+        evaluation_billing_context(EvaluationBillingOwner("creator", max_budget=1)),
+        ThreadPoolExecutor(max_workers=1) as worker,
+    ):
+        worker.submit(release.wait)
+        monkeypatch.setattr(litellm_logging, "executor", worker)
+        monkeypatch.setattr(thread_pool_executor, "executor", worker)
+        upstream: Final = transport.post("https://evaluation.invalid/v1/chat/completions").respond(
+            200,
+            json=ModelResponse(
+                model="evaluation-receipt",
+                choices=[{"message": {"role": "assistant", "content": "ok"}}],
+                usage=Usage(prompt_tokens=10, completion_tokens=1, total_tokens=11),
+            ).model_dump(),
+        )
+        logger: Final = LitellmLogging(
+            model=model,
+            messages=[],
+            stream=False,
+            call_type="acompletion" if api == "chat" else "anthropic_messages",
+            start_time=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
+            litellm_call_id="reused-evaluation",
+            function_id="test",
+            dynamic_async_success_callbacks=[capture],
+        )
+        logger._defer_async_logging = True
+        request: Final = {
+            "messages": [{"role": "user", "content": "hello"}],
+            "max_tokens": 10,
+            "metadata" if api == "chat" else "litellm_metadata": metadata,
+            "litellm_logging_obj": logger,
+            "fallbacks": [],
+        }
+        try:
+            await create(model="first-group", **request)
+            first: Final = logger._enqueue_deferred_logging
+            assert first is not None
+            logger._enqueue_deferred_logging = None
+            await create(model="later-group", **request)
+            second: Final = logger._enqueue_deferred_logging
+            assert second is not None
+            first()
+            second()
+            received: Final = (
+                await asyncio.wait_for(async_receipts.get(), 10),
+                await asyncio.wait_for(async_receipts.get(), 10),
+            )
+            await GLOBAL_LOGGING_WORKER.flush()
+        finally:
+            release.set()
+    assert upstream.call_count == 2 and async_receipts.empty() and sync_receipts.qsize() == 2
+    for receipts in (received, (sync_receipts.get_nowait(), sync_receipts.get_nowait())):
+        assert {receipt["model_group"]: receipt["model_id"] for receipt in receipts} == {
+            "first-group": "first-group-id",
+            "later-group": "later-group-id",
+        }
+        assert {receipt["model_group"]: receipt["response_cost"] for receipt in receipts} == pytest.approx(
+            {"first-group": 10 * 0.001 + 0.002, "later-group": 10 * 0.003 + 0.004}
+        )
+        for receipt in receipts:
+            assert payload.validate_python(receipt["metadata"])["user_api_key_user_id"] == "creator"
+            assert payload.validate_python(receipt["metadata"])["user_api_key_team_id"] is None
+    assert metadata["model_group"] == "later-group"
+    assert metadata["user_api_key_user_id"] == "caller" and metadata["user_api_key_team_id"] == "caller-team"

tests/unit/litellm_core_utils/test_llm_judge.py

@@ -1,12 +1,15 @@
 """Unit tests for the shared LLM-judge primitives: verdict parsing, router resolution, dispatch."""
 
 import json
-from unittest.mock import AsyncMock, MagicMock
+from typing import Final
+from unittest.mock import AsyncMock
 
 import pytest
+import respx
 
 import litellm
 from litellm.litellm_core_utils.llm_judge import (
+    default_router_provider,
     extract_text_from_content,
     judge_acompletion,
     judge_target,
@@ -106,6 +109,30 @@ async def test_judge_acompletion_falls_back_to_sdk_for_unconfigured_model(monkey
     assert sdk.call_args.kwargs["drop_params"] is True
 
 
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("evaluation_spend_cache")
+async def test_sdk_judge_failure_does_not_inherit_a_global_fallback(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(litellm, "model_fallbacks", ["openai/judge-fallback"])
+    with respx.mock(assert_all_called=False) as transport:
+        primary: Final = transport.post("https://judge.test/v1/chat/completions", json__model="judge-primary").respond(
+            400, json={"error": {"message": "judge rejected", "type": "invalid_request_error"}}
+        )
+        fallback: Final = transport.post(
+            "https://judge.test/v1/chat/completions", json__model="judge-fallback"
+        ).respond(200, json=litellm.ModelResponse().model_dump())
+        with pytest.raises(litellm.BadRequestError, match="judge rejected"):
+            await judge_acompletion(
+                default_router_provider(),
+                "openai/judge-primary",
+                [{"role": "user", "content": "judge this"}],
+                api_key="test",
+                api_base="https://judge.test/v1",
+                max_tokens=10,
+            )
+        assert primary.call_count == 1
+        assert not fallback.called
+
+
 @pytest.mark.parametrize(
     "model,expected",
     [

tests/unit/proxy/spend_tracking/test_evaluation_budget.py

@@ -0,0 +1,242 @@
+import asyncio
+import time
+from collections.abc import Mapping
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Final
+from unittest.mock import patch
+
+import pytest
+import respx
+
+import litellm
+from litellm.caching.caching import DualCache
+from litellm.caching.in_memory_cache import InMemoryCache
+from litellm.litellm_core_utils.internal_call_metadata import EvaluationBillingOwner, evaluation_billing_context
+from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
+from litellm.proxy import proxy_server
+from litellm.proxy.spend_tracking.budget_reservation import estimate_request_max_cost
+from litellm.proxy.spend_tracking.evaluation_budget import (
+    EvaluationAttempt,
+    _complete,
+    model_budget_spend,
+    reserve_evaluation_budget,
+)
+from litellm.types.utils import ModelResponse, Usage
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("scope", ("total", "model", "both"))
+async def test_concurrent_evaluations_settle_once_without_spending_another_attempt(
+    evaluation_spend_cache: DualCache, monkeypatch: pytest.MonkeyPatch, scope: str
+) -> None:
+    monkeypatch.setattr(evaluation_spend_cache, "in_memory_cache", InMemoryCache(clock=lambda: 0.0))
+    model: Final = "openai/evaluation-budget-test"
+    monkeypatch.setitem(litellm.model_cost, model, {"input_cost_per_token": 0.001, "output_cost_per_token": 0.002})
+    request: Final = {"model": model, "messages": [{"role": "user", "content": "hello"}], "max_tokens": 10}
+    estimate: Final = estimate_request_max_cost(request, "/chat/completions", None)
+    assert estimate is not None and estimate > 0
+    owner: Final = EvaluationBillingOwner(
+        "creator",
+        {model: {"max_budget": estimate * 1.5, "budget_duration": "1d"}} if scope != "total" else None,
+        max_budget=estimate * 1.5 if scope != "model" else None,
+    )
+    attempts: Final = await asyncio.gather(
+        *(reserve_evaluation_budget(owner, request, "acompletion") for _ in range(2)), return_exceptions=True
+    )
+    assert sum(isinstance(result, litellm.BudgetExceededError) for result in attempts) == 1
+    admitted: Final = next(result for result in attempts if not isinstance(result, BaseException))
+    assert admitted is not None
+    await admitted.settle(estimate / 4)
+    later: Final = await reserve_evaluation_budget(owner, request, "acompletion")
+    assert later is not None
+    for cost in (estimate / 4, 0, estimate / 2):
+        await admitted.settle(cost)
+        await proxy_server.increment_spend_counters(
+            token=None, team_id=None, user_id="creator", response_cost=cost, budget_reservation=admitted.total
+        )
+    await later.settle(0)
+    await proxy_server.model_max_budget_limiter.async_log_success_event(
+        {
+            "litellm_params": {"metadata": {"user_api_key_user_model_max_budget": owner.user_model_max_budget}},
+            "standard_logging_object": {
+                "model": model,
+                "response_cost": 0.01,
+                "metadata": {"user_api_key_user_id": owner.user_id},
+            },
+        },
+        None,
+        None,
+        None,
+    )
+    keys: Final = {"total": "spend:user:creator", "model": f"user_model_spend:creator:{model}:1d"}
+    for kind in keys if scope == "both" else (scope,):
+        assert await evaluation_spend_cache.async_get_cache(keys[kind]) == pytest.approx(
+            estimate / 2 + (0.01 if kind == "model" else 0)
+        )
+
+
+@pytest.mark.asyncio
+async def test_auto_router_prices_its_selected_deployment_and_bills_the_creator(
+    evaluation_spend_cache: DualCache, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    rates: Final = {"input_cost_per_token": 0.001, "output_cost_per_token": 0.002}
+    router: Final = litellm.Router(
+        model_list=[
+            {
+                "model_name": "leaf",
+                "litellm_params": {"model": "openai/private-evaluation", "api_key": "test", **rates},
+                "model_info": {"id": "selected", "max_input_tokens": 1000, "max_output_tokens": 10},
+            },
+            {
+                "model_name": "router",
+                "litellm_params": {
+                    "model": "auto_router/complexity_router",
+                    "complexity_router_config": {
+                        "classifier_type": "heuristic",
+                        "tiers": dict.fromkeys(("SIMPLE", "MEDIUM", "COMPLEX", "REASONING"), "leaf"),
+                    },
+                },
+            },
+        ],
+        num_retries=0,
+    )
+    monkeypatch.setattr(proxy_server, "llm_router", router)
+    owner: Final = EvaluationBillingOwner("creator", {"router": {"max_budget": 1, "budget_duration": "1d"}}, 1)
+    receipts: Final[asyncio.Queue[Mapping[str, object]]] = asyncio.Queue()
+    source: Final = {"user_api_key_user_id": "source", "user_api_key_team_id": "team", "agent_id": "agent"}
+
+    async def capture(kwargs: Mapping[str, object], response: object, start: datetime, end: datetime) -> None:
+        receipts.put_nowait(kwargs)
+
+    with respx.mock(assert_all_called=True) as transport:
+        transport.post("https://api.openai.com/v1/chat/completions").respond(
+            200,
+            json=ModelResponse(
+                model="private-evaluation",
+                choices=[{"message": {"role": "assistant", "content": "ok"}}],
+                usage=Usage(prompt_tokens=10, completion_tokens=2, total_tokens=12),
+            ).model_dump(),
+        )
+
+        with evaluation_billing_context(owner):
+            await router.acompletion(
+                model="router",
+                messages=[{"role": "user", "content": "hello"}],
+                max_tokens=10,
+                metadata=source,
+                fallbacks=[],
+                success_callback=[capture],
+            )
+        receipt: Final = await asyncio.wait_for(receipts.get(), 10)
+        await GLOBAL_LOGGING_WORKER.flush()
+    assert (receipt["user"], receipt["agent_id"], receipt["request_tags"]) == ("creator", None, [])
+    assert source["user_api_key_user_id"] == "source" and source["agent_id"] == "agent"
+    actual: Final = 10 * rates["input_cost_per_token"] + 2 * rates["output_cost_per_token"]
+    assert receipt["response_cost"] == pytest.approx(actual)
+    for key in ("spend:user:creator", "user_model_spend:creator:router:1d"):
+        assert await evaluation_spend_cache.async_get_cache(key) == pytest.approx(actual)
+
+
+@dataclass(slots=True)
+class _Clock:
+    seconds: float = 0.0
+
+
+@pytest.mark.asyncio
+async def test_expired_model_hold_cannot_release_or_renew_another_requests_budget() -> None:
+    clock: Final = _Clock()
+    cache: Final = DualCache(in_memory_cache=InMemoryCache(clock=lambda: clock.seconds))
+    await model_budget_spend(cache, "budget", operation="reserve", member="old:0.2")
+    clock.seconds = 30
+    assert await model_budget_spend(cache, "budget", operation="reserve", member="new:0.3") == pytest.approx(0.5)
+    clock.seconds = 61
+    assert await model_budget_spend(cache, "budget") == pytest.approx(0.3)
+    assert await model_budget_spend(cache, "budget", operation="settle", member="old:0.2") == pytest.approx(0.3)
+    with pytest.raises(RuntimeError, match="Evaluation reservation expired"):
+        await model_budget_spend(cache, "budget", operation="renew", member="old:0.2")
+    await model_budget_spend(cache, "budget", operation="renew", member="new:0.3")
+    clock.seconds = 92
+    assert await model_budget_spend(cache, "budget") == pytest.approx(0.3)
+    assert await model_budget_spend(cache, "budget", operation="settle", member="new:0.3") == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("phase", ("admission", "settlement"))
+async def test_repeated_cancellation_drains_budget_operations_before_returning(
+    evaluation_spend_cache: DualCache, monkeypatch: pytest.MonkeyPatch, phase: str
+) -> None:
+    model: Final = "openai/cancelled-evaluation"
+    monkeypatch.setitem(litellm.model_cost, model, {"input_cost_per_token": 0.001, "output_cost_per_token": 0.002})
+    request: Final = {"model": model, "messages": [{"role": "user", "content": "hello"}], "max_tokens": 10}
+    owner: Final = EvaluationBillingOwner("creator", {model: {"max_budget": 1, "budget_duration": "1d"}}, 1)
+    existing: Final = await reserve_evaluation_budget(owner, request, "acompletion") if phase == "settlement" else None
+    entered: Final = asyncio.Event()
+    proceed: Final = asyncio.Event()
+
+    async def delayed_write() -> EvaluationAttempt | None:
+        attempt: Final = existing or await reserve_evaluation_budget(owner, request, "acompletion")
+        assert attempt is not None
+        entered.set()
+        await proceed.wait()
+        if existing is None:
+            return attempt
+        await attempt.settle(0.005)
+        return None
+
+    pending: Final = asyncio.create_task(_complete(delayed_write()))
+    await asyncio.wait_for(entered.wait(), 5)
+    pending.cancel()
+    await asyncio.sleep(0)
+    pending.cancel()
+    assert not pending.done()
+    proceed.set()
+    with pytest.raises(asyncio.CancelledError):
+        await pending
+    expected: Final = 0.005 if phase == "settlement" else 0
+    assert await evaluation_spend_cache.async_get_cache("spend:user:creator") == pytest.approx(expected)
+    assert await model_budget_spend(evaluation_spend_cache, f"user_model_spend:creator:{model}:1d") == pytest.approx(
+        expected
+    )
+
+
+@pytest.mark.asyncio
+async def test_evaluation_settlement_preserves_spend_until_the_shared_budget_window_ends(
+    evaluation_spend_cache: DualCache, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    clock: Final = _Clock()
+    epoch: Final = 1000.0
+    monkeypatch.setattr(time, "time", lambda: epoch + 0.2 + clock.seconds)
+    monkeypatch.setattr(evaluation_spend_cache, "in_memory_cache", InMemoryCache(clock=lambda: clock.seconds))
+    model: Final = "openai/shared-window"
+    monkeypatch.setitem(litellm.model_cost, model, {"input_cost_per_token": 0.001, "output_cost_per_token": 0.002})
+    owner: Final = EvaluationBillingOwner("creator", {model: {"max_budget": 1, "budget_duration": "10s"}})
+    key: Final = f"user_model_spend:creator:{model}:10s"
+    evaluation_spend_cache.in_memory_cache.set_cache(key, 0.1, ttl=1)
+    evaluation_spend_cache.in_memory_cache.set_cache(f"user_model_budget_start_time:creator:{model}:10s", epoch, ttl=10)
+    attempt: Final = await reserve_evaluation_budget(
+        owner, {"model": model, "messages": [{"role": "user", "content": "hello"}], "max_tokens": 10}, "acompletion"
+    )
+    assert attempt is not None
+    await attempt.settle(0.02)
+    with patch("litellm.router_strategy.budget_limiter.datetime") as wall_clock:
+        wall_clock.now.return_value = datetime.fromtimestamp(epoch + 0.2, timezone.utc)
+        await proxy_server.model_max_budget_limiter.async_log_success_event(
+            {
+                "litellm_params": {"metadata": {"user_api_key_user_model_max_budget": owner.user_model_max_budget}},
+                "standard_logging_object": {
+                    "model": model,
+                    "response_cost": 0.01,
+                    "metadata": {"user_api_key_user_id": owner.user_id},
+                },
+            },
+            None,
+            None,
+            None,
+        )
+    clock.seconds = 2
+    assert await model_budget_spend(evaluation_spend_cache, key) == pytest.approx(0.13)
+    clock.seconds = 9.9
+    assert await model_budget_spend(evaluation_spend_cache, key) == pytest.approx(0.13)
+    clock.seconds = 10.1
+    assert await model_budget_spend(evaluation_spend_cache, key) == 0

tests/unit/test_utils.py

@@ -29,13 +29,16 @@ from litellm._logging import (
     trace_id_var,
     verbose_logger,
 )
-from litellm.caching.caching import Cache
+from litellm.caching.caching import Cache, DualCache
 from litellm.caching.caching_handler import _PENDING_CACHE_WRITES
 from litellm.caching.in_memory_cache import InMemoryCache
 from litellm.constants import DEFAULT_MOCK_RESPONSE_COMPLETION_TOKEN_COUNT
 from litellm.integrations.custom_guardrail import CustomGuardrail
 from litellm.integrations.custom_logger import CustomLogger
 from litellm.litellm_core_utils.get_litellm_params import get_litellm_params
+from litellm.litellm_core_utils.internal_call_metadata import EvaluationBillingOwner, evaluation_billing_context
+from litellm.litellm_core_utils.litellm_logging import Logging
+from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
 from litellm.litellm_core_utils.thread_pool_executor import executor as logging_executor
 from litellm.llms.base_llm.base_model_iterator import MockResponseIterator
 from litellm.proxy.utils import is_valid_api_key
@@ -6558,3 +6561,342 @@ def test_function_setup_never_logs_the_ocr_data_uri_payload() -> None:
 
     assert logged == [{"role": "user", "content": f"data:application/pdf;base64 ({len(payload)} chars)"}]
     assert payload not in str(logged)
+
+
+@pytest.fixture
+def evaluation_model(monkeypatch: pytest.MonkeyPatch) -> str:
+    model: Final = "hosted_vllm/evaluation-test"
+    for deployment in (model, model + "-fallback", "anthropic/evaluation-native"):
+        monkeypatch.setitem(
+            litellm.model_cost,
+            deployment,
+            {
+                "input_cost_per_token": 0.001,
+                "output_cost_per_token": 0.002,
+                "max_input_tokens": 1000,
+                "max_output_tokens": 1000,
+                "litellm_provider": deployment.split("/", 1)[0],
+                "mode": "chat",
+            },
+        )
+    return model
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("api", ("chat", "messages", "messages_responses", "embedding"))
+@pytest.mark.parametrize("scope", ("total", "model"))
+async def test_evaluation_budget_blocks_paid_leaf_calls_before_transport(
+    evaluation_spend_cache: DualCache, evaluation_model: str, api: str, scope: str
+) -> None:
+    model: Final = "openai/gpt-5.6-luna" if api == "messages_responses" else evaluation_model
+    owner: Final = EvaluationBillingOwner(
+        "creator",
+        {model: {"max_budget": 0, "budget_duration": "1d"}} if scope == "model" else None,
+        max_budget=0 if scope == "total" else None,
+    )
+    create: Final = {
+        "chat": litellm.acompletion,
+        "messages": litellm.anthropic.messages.acreate,
+        "messages_responses": litellm.anthropic.messages.acreate,
+        "embedding": litellm.aembedding,
+    }[api]
+    body: Final = (
+        {"input": ["hello"]}
+        if api == "embedding"
+        else {"messages": [{"role": "user", "content": "hello"}], "max_tokens": 10}
+    )
+    with respx.mock(assert_all_called=False) as transport, evaluation_billing_context(owner):
+        upstream: Final = transport.post(url__startswith="https://evaluation.invalid/").respond(500)
+        with pytest.raises(litellm.BudgetExceededError):
+            await create(
+                model=model,
+                api_base="https://evaluation.invalid/v1",
+                api_key="test",
+                metadata={"internal_call_origin": "autorouter_classifier"},
+                fallbacks=[],
+                num_retries=0,
+                **body,
+            )
+        assert upstream.call_count == 0
+    assert (await evaluation_spend_cache.async_get_cache("spend:user:creator") or 0) == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("global_fallbacks", (False, True))
+async def test_messages_fallback_checks_its_own_budget_before_a_second_provider_call(
+    evaluation_spend_cache: DualCache,
+    evaluation_model: str,
+    monkeypatch: pytest.MonkeyPatch,
+    global_fallbacks: bool,
+) -> None:
+    fallback: Final = evaluation_model + "-fallback"
+    monkeypatch.setattr(litellm, "model_fallbacks", [fallback] if global_fallbacks else None)
+    owner: Final = EvaluationBillingOwner("creator", {fallback: {"max_budget": 0, "budget_duration": "1d"}}, 1)
+    with respx.mock(assert_all_called=True) as transport, evaluation_billing_context(owner):
+        upstream: Final = transport.post("https://evaluation.invalid/v1/chat/completions").respond(
+            500, json={"error": {"message": "unavailable", "type": "server_error"}}
+        )
+        with pytest.raises(Exception, match="Max budget: 0"):
+            await litellm.anthropic.messages.acreate(
+                model=evaluation_model,
+                messages=[{"role": "user", "content": "hi"}],
+                max_tokens=10,
+                fallbacks=None if global_fallbacks else [fallback],
+                api_base="https://evaluation.invalid/v1",
+                api_key="test",
+                num_retries=0,
+                max_retries=0,
+            )
+        assert upstream.call_count == 1
+    assert (await evaluation_spend_cache.async_get_cache("spend:user:creator") or 0) == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("global_fallbacks", (False, True))
+async def test_native_messages_keep_creator_admission_when_chat_fallbacks_are_configured(
+    evaluation_spend_cache: DualCache,
+    evaluation_model: str,
+    monkeypatch: pytest.MonkeyPatch,
+    global_fallbacks: bool,
+) -> None:
+    model: Final = "anthropic/evaluation-native"
+    monkeypatch.setattr(litellm, "model_fallbacks", [evaluation_model] if global_fallbacks else None)
+    owner: Final = EvaluationBillingOwner("creator", {model: {"max_budget": 0, "budget_duration": "1d"}}, 1)
+    with respx.mock(assert_all_called=False) as transport, evaluation_billing_context(owner):
+        upstream: Final = transport.post(url__startswith="https://evaluation.invalid/").respond(500)
+        with pytest.raises(litellm.BudgetExceededError):
+            await litellm.anthropic.messages.acreate(
+                model=model,
+                messages=[{"role": "user", "content": "hello"}],
+                max_tokens=10,
+                api_base="https://evaluation.invalid/v1",
+                api_key="test",
+                num_retries=0,
+                fallbacks=None if global_fallbacks else [evaluation_model],
+            )
+        assert upstream.call_count == 0
+    assert (await evaluation_spend_cache.async_get_cache("spend:user:creator") or 0) == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("api", ("chat", "messages"))
+async def test_delayed_evaluation_receipts_survive_logger_reuse_and_a_cache_hit(
+    evaluation_spend_cache: DualCache, evaluation_model: str, monkeypatch: pytest.MonkeyPatch, api: str
+) -> None:
+    monkeypatch.setattr(litellm, "cache", Cache(type="local"))
+    owner: Final = EvaluationBillingOwner("creator", max_budget=1)
+    receipts: Final[asyncio.Queue[Mapping[str, object]]] = asyncio.Queue()
+    requests: Final[asyncio.Queue[httpx.Request]] = asyncio.Queue()
+
+    async def capture(kwargs: Mapping[str, object], response: object, start: datetime, end: datetime) -> None:
+        receipts.put_nowait(kwargs)
+
+    def upstream(request: httpx.Request) -> httpx.Response:
+        requests.put_nowait(request)
+        return httpx.Response(
+            200,
+            json=ModelResponse(
+                model="evaluation-test",
+                choices=[{"message": {"role": "assistant", "content": "ok"}}],
+                usage=Usage(prompt_tokens=10, completion_tokens=requests.qsize(), total_tokens=10 + requests.qsize()),
+            ).model_dump(),
+        )
+
+    with respx.mock(assert_all_called=True) as transport, evaluation_billing_context(owner):
+        transport.post("https://evaluation.invalid/v1/chat/completions").mock(side_effect=upstream)
+        logger: Final = Logging(
+            model=evaluation_model,
+            messages=[],
+            stream=False,
+            call_type="acompletion",
+            start_time=datetime.now(),
+            litellm_call_id="evaluation",
+            function_id="test",
+            dynamic_async_success_callbacks=[capture],
+        )
+        logger._defer_async_logging = True
+        create: Final = litellm.acompletion if api == "chat" else litellm.anthropic.messages.acreate
+        request: Final = {
+            "model": evaluation_model,
+            "messages": [{"role": "user", "content": "one"}],
+            "max_tokens": 10,
+            "api_base": "https://evaluation.invalid/v1",
+            "api_key": "test",
+            "num_retries": 0,
+            "litellm_logging_obj": logger,
+            "fallbacks": [],
+        }
+        await create(**request)
+        first: Final = logger._enqueue_deferred_logging
+        assert first is not None
+        logger._enqueue_deferred_logging = None
+        await asyncio.gather(*tuple(_PENDING_CACHE_WRITES))
+        await create(**request)
+        cached: Final = await asyncio.wait_for(receipts.get(), 10)
+        assert cached["response_cost"] == 0 and requests.qsize() == 1
+        assert await evaluation_spend_cache.async_get_cache("spend:user:creator") > 0
+        await create(**{**request, "messages": [{"role": "user", "content": "two"}]})
+        second: Final = logger._enqueue_deferred_logging
+        assert second is not None
+        first()
+        second()
+        received: Final = (await asyncio.wait_for(receipts.get(), 10), await asyncio.wait_for(receipts.get(), 10))
+        await GLOBAL_LOGGING_WORKER.flush()
+    assert requests.qsize() == 2 and receipts.empty()
+    assert {receipt["response_cost"] for receipt in received} == {10 * 0.001 + 0.002, 10 * 0.001 + 2 * 0.002}
+    assert tuple(receipt["user"] for receipt in received) == ("creator", "creator")
+    assert await evaluation_spend_cache.async_get_cache("spend:user:creator") == pytest.approx(20 * 0.001 + 3 * 0.002)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("api", ("chat", "messages"))
+async def test_cancelled_evaluation_releases_output_hold_and_keeps_input_cost(
+    evaluation_spend_cache: DualCache, evaluation_model: str, api: str
+) -> None:
+    entered: Final = asyncio.Event()
+    owner: Final = EvaluationBillingOwner("creator", {evaluation_model: {"max_budget": 1, "budget_duration": "1d"}}, 1)
+
+    async def upstream(request: httpx.Request) -> httpx.Response:
+        entered.set()
+        return await asyncio.Future[httpx.Response]()
+
+    with respx.mock(assert_all_called=False) as transport, evaluation_billing_context(owner):
+        transport.post("https://evaluation.invalid/v1/chat/completions").mock(side_effect=upstream)
+        create: Final = litellm.acompletion if api == "chat" else litellm.anthropic.messages.acreate
+        pending: Final = asyncio.create_task(
+            create(
+                model=evaluation_model,
+                messages=[{"role": "user", "content": "hello"}],
+                max_tokens=10,
+                api_base="https://evaluation.invalid/v1",
+                api_key="test",
+                fallbacks=[],
+                num_retries=0,
+            )
+        )
+        await asyncio.wait_for(entered.wait(), 5)
+        held: Final = await evaluation_spend_cache.async_get_cache("spend:user:creator")
+        pending.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await pending
+    settled: Final = await evaluation_spend_cache.async_get_cache("spend:user:creator")
+    assert 0 < settled < held
+    assert await evaluation_spend_cache.async_get_cache(f"user_model_spend:creator:{evaluation_model}:1d") == settled
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("api", ("chat", "messages"))
+@pytest.mark.parametrize("intercept_stream", (False, True))
+async def test_evaluation_streams_are_rejected_before_paid_transport(
+    evaluation_spend_cache: DualCache,
+    evaluation_model: str,
+    monkeypatch: pytest.MonkeyPatch,
+    api: str,
+    intercept_stream: bool,
+) -> None:
+    from litellm.integrations.code_interpreter_interception.handler import CodeInterpreterInterceptionLogger
+
+    monkeypatch.setattr(litellm, "callbacks", [CodeInterpreterInterceptionLogger()] if intercept_stream else [])
+    create: Final = litellm.acompletion if api == "chat" else litellm.anthropic.messages.acreate
+    with (
+        respx.mock(assert_all_called=False) as transport,
+        evaluation_billing_context(EvaluationBillingOwner("creator", max_budget=1)),
+    ):
+        upstream: Final = transport.post("https://evaluation.invalid/v1/chat/completions").respond(500)
+        with pytest.raises(litellm.BadRequestError, match="Shadow evaluation requires a non-streaming response"):
+            await create(
+                model=evaluation_model,
+                messages=[{"role": "user", "content": "hello"}],
+                max_tokens=10,
+                api_base="https://evaluation.invalid/v1",
+                api_key="test",
+                fallbacks=[],
+                stream=True,
+                tools=[{"type": "code_interpreter"}],
+            )
+    assert upstream.call_count == 0
+    assert await evaluation_spend_cache.async_get_cache("spend:user:creator") is None
+
+
+@pytest.mark.asyncio
+async def test_internal_evaluation_streams_do_not_claim_budget(
+    evaluation_spend_cache: DualCache, evaluation_model: str
+) -> None:
+    token: Final = is_internal_call.set(True)
+    try:
+        with evaluation_billing_context(EvaluationBillingOwner("creator", max_budget=0)):
+            response: Final = await litellm.acompletion(
+                model=evaluation_model,
+                messages=[{"role": "user", "content": "hello"}],
+                max_tokens=10,
+                stream=True,
+                mock_response="ok",
+            )
+            assert isinstance(response, CustomStreamWrapper)
+            await response.aclose()
+    finally:
+        is_internal_call.reset(token)
+    assert await evaluation_spend_cache.async_get_cache("spend:user:creator") is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("budget", (0, 1))
+@pytest.mark.parametrize("api", ("chat", "messages"))
+async def test_global_fallback_leaves_enforce_creator_budget_and_emit_one_paid_receipt(
+    evaluation_spend_cache: DualCache, evaluation_model: str, monkeypatch: pytest.MonkeyPatch, budget: int, api: str
+) -> None:
+    fallback: Final = evaluation_model + "-fallback"
+    monkeypatch.setattr(litellm, "model_fallbacks", [fallback])
+    receipts: Final[asyncio.Queue[Mapping[str, object]]] = asyncio.Queue()
+    owner: Final = EvaluationBillingOwner(
+        "creator",
+        {model: {"max_budget": 1, "budget_duration": "1d"} for model in (evaluation_model, fallback)},
+        max_budget=budget,
+    )
+
+    async def capture(kwargs: Mapping[str, object], response: object, start: datetime, end: datetime) -> None:
+        receipts.put_nowait(kwargs)
+
+    with (
+        respx.mock(assert_all_called=False) as transport,
+        evaluation_billing_context(owner),
+    ):
+        upstream: Final = transport.post("https://evaluation.invalid/v1/chat/completions").mock(
+            side_effect=[
+                httpx.Response(500, json={"error": {"message": "unavailable", "type": "server_error"}}),
+                httpx.Response(
+                    200,
+                    json=ModelResponse(
+                        model="evaluation-test-fallback",
+                        choices=[{"message": {"role": "assistant", "content": "ok"}}],
+                        usage=Usage(prompt_tokens=10, completion_tokens=2, total_tokens=12),
+                    ).model_dump(),
+                ),
+            ],
+        )
+        create: Final = litellm.acompletion if api == "chat" else litellm.anthropic.messages.acreate
+        request: Final = create(
+            model=evaluation_model,
+            messages=[{"role": "user", "content": "hello"}],
+            max_tokens=10,
+            api_base="https://evaluation.invalid/v1",
+            api_key="test",
+            num_retries=0,
+            max_retries=0,
+            success_callback=[capture],
+        )
+        if budget == 0:
+            with pytest.raises(Exception, match="Max budget: 0"):
+                await request
+        else:
+            await request
+            receipt: Final = await asyncio.wait_for(receipts.get(), 10)
+            assert (receipt["user"], receipt["response_cost"]) == ("creator", 10 * 0.001 + 2 * 0.002)
+        await GLOBAL_LOGGING_WORKER.flush()
+    assert upstream.call_count == budget * 2
+    assert receipts.empty()
+    assert (await evaluation_spend_cache.async_get_cache("spend:user:creator") or 0) == pytest.approx(budget * 0.014)
+    assert (await evaluation_spend_cache.async_get_cache(f"user_model_spend:creator:{evaluation_model}:1d") or 0) == 0
+    assert (
+        await evaluation_spend_cache.async_get_cache(f"user_model_spend:creator:{fallback}:1d") or 0
+    ) == pytest.approx(budget * 0.014)

ui/litellm-dashboard/src/app/(dashboard)/cost-optimization/_components/ShadowEvalStartForm.tsx

@@ -38,9 +38,9 @@ const DIRECTION_OPTIONS: readonly { value: ShadowEvalDirection; label: string }[
 
 const START_FORM_DESCRIPTION: Record<ShadowEvalDirection, string> = {
   forward:
-    "Duplicates a sampled slice of the selected targets' traffic (keys, teams, or users) through the auto-router and has an LLM judge compare both answers blind. Each target gets its own spend budget. The router's answers are never served to users; judge calls bill to the sampled traffic's own identity.",
+    "Duplicates a sampled slice of the selected targets' traffic (keys, teams, or users) through the auto-router and has an LLM judge compare both answers blind. Each target gets its own spend budget. The router's answers are never served to users; shadow and judge calls bill to the initiating admin.",
   reverse:
-    "Duplicates a sampled slice of the traffic the auto-router already serves against a fixed baseline model and has an LLM judge compare both answers blind. Each target gets its own spend budget. The baseline's answers are never served to users; judge calls bill to the sampled traffic's own identity.",
+    "Duplicates a sampled slice of the traffic the auto-router already serves against a fixed baseline model and has an LLM judge compare both answers blind. Each target gets its own spend budget. The baseline's answers are never served to users; shadow and judge calls bill to the initiating admin.",
 };
 
 const DURATION_OPTIONS = [

ui/litellm-dashboard/src/lib/http/schema.d.ts

@@ -1475,7 +1475,7 @@ export interface paths {
          *     eval spend, the shadow and judge calls' own cost, reaches max_budget dollars, the
          *     job's window ends, or the job is stopped, so one target running out of budget does
          *     not end sampling for the others; sampling changes propagate to pods within about 10
-         *     seconds. Shadow and judge calls bill to the sampled request's own identity but are
+         *     seconds. Shadow and judge calls bill to the admin who started the job and are
          *     excluded from request counts and auto-router adoption metrics.
          */
         post: operations["start_shadow_eval_auto_router_shadow_eval_start_post"];