Merge pull request #29541 from BerriAI/litellm_cherrypick_1_86_3

chore(release): backport pending 1.84.x/1.85.x fixes into stable/1.86.x and cut 1.86.3

litellm/litellm_core_utils/litellm_logging.py

@@ -1613,6 +1613,90 @@ class Logging(LiteLLMLoggingBaseClass):
     ) -> Optional[float]:
         return self._response_cost_calculator(result=result, cache_hit=cache_hit)
 
+    @staticmethod
+    def _is_sync_litellm_request(litellm_params: dict) -> bool:
+        """True for sync SDK entrypoints (``completion``), false for async (``acompletion``, etc.)."""
+        return (
+            litellm_params.get(CallTypes.acompletion.value, False) is not True
+            and litellm_params.get(CallTypes.aresponses.value, False) is not True
+            and litellm_params.get(CallTypes.aembedding.value, False) is not True
+            and litellm_params.get(CallTypes.aimage_generation.value, False) is not True
+            and litellm_params.get(CallTypes.atranscription.value, False) is not True
+        )
+
+    def _is_assembled_stream_success(self, result=None) -> bool:
+        """Final assembled stream export (not a per-chunk success call).
+
+        Per-chunk callers pass a ``ModelResponseStream`` (or ``None``); the
+        final assembled response is any other non-``None`` value (typically a
+        ``ModelResponse``). Treating a chunk as the assembled response would
+        prematurely set the ``has_dispatched_final_stream_success`` dedup
+        guard and silently suppress the real final stream log.
+        """
+        if self.stream is not True:
+            return False
+        if result is not None and not isinstance(result, ModelResponseStream):
+            return True
+        return (
+            "async_complete_streaming_response" in self.model_call_details
+            or self.model_call_details.get("complete_streaming_response") is not None
+        )
+
+    async def dispatch_success_handlers(
+        self,
+        result=None,
+        start_time=None,
+        end_time=None,
+        cache_hit=None,
+        prefer_async_handlers: bool = False,
+        **kwargs,
+    ) -> None:
+        """Route success logging to async and/or sync handlers for this request.
+
+        ``prefer_async_handlers`` only bypasses the sync-SDK-only shortcut (e.g.
+        ``async for`` on a stream from ``completion()``). Legacy string callbacks
+        still run via ``executor.submit(success_handler)`` when configured.
+        """
+        from litellm.litellm_core_utils.thread_pool_executor import executor
+
+        if self._is_assembled_stream_success(result):
+            if self.model_call_details.get("has_dispatched_final_stream_success"):
+                return
+            self.model_call_details["has_dispatched_final_stream_success"] = True
+
+        litellm_params = self.model_call_details.get("litellm_params", {}) or {}
+        sync_sdk = self._is_sync_litellm_request(litellm_params)
+        passthrough = self.call_type == CallTypes.pass_through.value
+        if sync_sdk and not prefer_async_handlers and not passthrough:
+            self.success_handler(
+                result,
+                start_time=start_time,
+                end_time=end_time,
+                cache_hit=cache_hit,
+                **kwargs,
+            )
+            return
+
+        await self.async_success_handler(
+            result,
+            start_time=start_time,
+            end_time=end_time,
+            cache_hit=cache_hit,
+            **kwargs,
+        )
+
+        if not self._should_run_sync_callbacks_for_async_calls():
+            return
+
+        executor.submit(
+            self.success_handler,
+            result,
+            start_time=start_time,
+            end_time=end_time,
+            cache_hit=cache_hit,
+            **kwargs,
+        )
+
     def should_run_logging(
         self,
         event_type: Literal[
@@ -2032,13 +2116,7 @@ class Logging(LiteLLMLoggingBaseClass):
             standard_logging_object=kwargs.get("standard_logging_object", None),
         )
         litellm_params = self.model_call_details.get("litellm_params", {})
-        is_sync_request = (
-            litellm_params.get(CallTypes.acompletion.value, False) is not True
-            and litellm_params.get(CallTypes.aresponses.value, False) is not True
-            and litellm_params.get(CallTypes.aembedding.value, False) is not True
-            and litellm_params.get(CallTypes.aimage_generation.value, False) is not True
-            and litellm_params.get(CallTypes.atranscription.value, False) is not True
-        )
+        is_sync_request = self._is_sync_litellm_request(litellm_params)
         try:
             ## BUILD COMPLETE STREAMED RESPONSE
             complete_streaming_response: Optional[
@@ -2494,9 +2572,11 @@ class Logging(LiteLLMLoggingBaseClass):
         print_verbose(
             "Logging Details LiteLLM-Async Success Call, cache_hit={}".format(cache_hit)
         )
-        if not self.should_run_logging(
+        if not self._is_assembled_stream_success(
+            result
+        ) and not self.should_run_logging(
             event_type="async_success"
-        ):  # prevent double logging
+        ):  # prevent double logging (non-streaming)
             return
 
         ## CALCULATE COST FOR BATCH JOBS
@@ -2946,13 +3026,7 @@ class Logging(LiteLLMLoggingBaseClass):
         ):  # prevent double logging
             return
         litellm_params = self.model_call_details.get("litellm_params", {})
-        is_sync_request = (
-            litellm_params.get(CallTypes.acompletion.value, False) is not True
-            and litellm_params.get(CallTypes.aresponses.value, False) is not True
-            and litellm_params.get(CallTypes.aembedding.value, False) is not True
-            and litellm_params.get(CallTypes.aimage_generation.value, False) is not True
-            and litellm_params.get(CallTypes.atranscription.value, False) is not True
-        )
+        is_sync_request = self._is_sync_litellm_request(litellm_params)
 
         try:
             start_time, end_time = self._failure_handler_helper_fn(

litellm/litellm_core_utils/prompt_templates/factory.py

@@ -1233,6 +1233,7 @@ def infer_protocol_value(
 
 def _gemini_tool_call_invoke_helper(
     function_call_params: ChatCompletionToolCallFunctionChunk,
+    tool_call_id: Optional[str] = None,
 ) -> Optional[VertexFunctionCall]:
     name = function_call_params.get("name", "") or ""
     arguments = function_call_params.get("arguments", "")
@@ -1248,6 +1249,10 @@ def _gemini_tool_call_invoke_helper(
         name=name,
         args=arguments_dict,
     )
+    if tool_call_id:
+        clean_id = tool_call_id.split(THOUGHT_SIGNATURE_SEPARATOR, 1)[0]
+        if clean_id:
+            function_call["id"] = clean_id
     return function_call
 
 
@@ -1339,6 +1344,7 @@ def _get_dummy_thought_signature() -> str:
 def convert_to_gemini_tool_call_invoke(
     message: ChatCompletionAssistantMessage,
     model: Optional[str] = None,
+    custom_llm_provider: Optional[str] = None,
 ) -> List[VertexPartType]:
     """
     OpenAI tool invokes:
@@ -1384,12 +1390,26 @@ def convert_to_gemini_tool_call_invoke(
         tool_calls = message.get("tool_calls", None)
         function_call = message.get("function_call", None)
 
+        from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
+            VertexGeminiConfig,
+        )
+
+        forward_tool_call_id = bool(
+            model
+            and VertexGeminiConfig._forward_gemini_function_call_id(
+                model, custom_llm_provider
+            )
+        )
+
         if tool_calls is not None:
             for idx, tool in enumerate(tool_calls):
                 if "function" in tool:
                     gemini_function_call: Optional[VertexFunctionCall] = (
                         _gemini_tool_call_invoke_helper(
-                            function_call_params=tool["function"]
+                            function_call_params=tool["function"],
+                            tool_call_id=(
+                                tool.get("id") if forward_tool_call_id else None
+                            ),
                         )
                     )
                     if gemini_function_call is not None:
@@ -1429,10 +1449,6 @@ def convert_to_gemini_tool_call_invoke(
                     thought_signature = provider_fields.get("thought_signature")
 
                 # If no signature found and model is gemini-3, use dummy signature
-                from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
-                    VertexGeminiConfig,
-                )
-
                 if (
                     not thought_signature
                     and model
@@ -1462,6 +1478,8 @@ def convert_to_gemini_tool_call_invoke(
 def convert_to_gemini_tool_call_result(  # noqa: PLR0915
     message: Union[ChatCompletionToolMessage, ChatCompletionFunctionMessage],
     last_message_with_tool_calls: Optional[dict],
+    model: Optional[str] = None,
+    custom_llm_provider: Optional[str] = None,
 ) -> Union[VertexPartType, List[VertexPartType]]:
     """
     OpenAI message with a tool result looks like:
@@ -1602,6 +1620,23 @@ def convert_to_gemini_tool_call_result(  # noqa: PLR0915
             ):
                 name = tool.get("function", {}).get("name", "")
 
+    # Echo the OpenAI tool_call_id on functionResponse (strip thought-signature suffix).
+    # Only Google AI Studio Gemini 3+ accepts `id` on function_response parts.
+    # Vertex AI and older Gemini models reject the field with HTTP 400.
+    from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
+        VertexGeminiConfig,
+    )
+
+    gemini_call_id: Optional[str] = None
+    if model and VertexGeminiConfig._forward_gemini_function_call_id(
+        model, custom_llm_provider
+    ):
+        raw_tool_call_id = message.get("tool_call_id")
+        if raw_tool_call_id and isinstance(raw_tool_call_id, str):
+            stripped_id = raw_tool_call_id.split(THOUGHT_SIGNATURE_SEPARATOR, 1)[0]
+            if stripped_id:
+                gemini_call_id = stripped_id
+
     if not name:
         raise Exception(
             "Missing corresponding tool call for tool response message. Received - message={}, last_message_with_tool_calls={}".format(
@@ -1632,6 +1667,8 @@ def convert_to_gemini_tool_call_result(  # noqa: PLR0915
         name=name,
         response=response_data,  # type: ignore
     )
+    if gemini_call_id:
+        _function_response["id"] = gemini_call_id
 
     # Create part with function_response, and optionally inline_data for images (Computer Use)
     _part: VertexPartType = {"function_response": _function_response}

litellm/litellm_core_utils/streaming_handler.py

@@ -1808,8 +1808,10 @@ class CustomStreamWrapper:
                     processed_chunk, None, None, cache_hit
                 )
             )
-        ## SYNC LOGGING
-        self.logging_obj.success_handler(processed_chunk, None, None, cache_hit)
+        ## SYNC LOGGING — only for sync SDK entrypoints; async proxy paths export via async_success_handler
+        litellm_params = self.logging_obj.model_call_details.get("litellm_params", {})
+        if self.logging_obj._is_sync_litellm_request(litellm_params):
+            self.logging_obj.success_handler(processed_chunk, None, None, cache_hit)
 
     def finish_reason_handler(self):
         model_response = self.model_response_creator()
@@ -2206,23 +2208,19 @@ class CustomStreamWrapper:
                         cache_hit,
                     )
                 else:
+                    # prefer_async_handlers routes CustomLogger to async_success_handler
+                    # when consumers use ``async for`` on sync-SDK streams. Legacy string
+                    # callbacks still run via executor.submit inside dispatch_success_handlers.
                     asyncio.create_task(
-                        self.logging_obj.async_success_handler(
+                        self.logging_obj.dispatch_success_handlers(
                             complete_streaming_response,
                             cache_hit=cache_hit,
                             start_time=None,
                             end_time=None,
+                            prefer_async_handlers=True,
                         )
                     )
 
-                    executor.submit(
-                        self.logging_obj.success_handler,
-                        complete_streaming_response,
-                        cache_hit=cache_hit,
-                        start_time=None,
-                        end_time=None,
-                    )
-
                 raise StopAsyncIteration  # Re-raise StopIteration
             else:
                 self.sent_last_chunk = True

litellm/llms/gemini/chat/transformation.py

@@ -151,4 +151,8 @@ class GoogleAIStudioGeminiConfig(VertexGeminiConfig):
                             except Exception:
                                 # If conversion fails, leave as is and let the API handle it
                                 pass
-        return _gemini_convert_messages_with_history(messages=messages, model=model)
+        return _gemini_convert_messages_with_history(
+            messages=messages,
+            model=model,
+            custom_llm_provider="gemini",
+        )

litellm/llms/vertex_ai/context_caching/transformation.py

@@ -174,7 +174,9 @@ def transform_openai_messages_to_gemini_context_caching(
     )
 
     transformed_messages = _gemini_convert_messages_with_history(
-        messages=new_messages, model=model
+        messages=new_messages,
+        model=model,
+        custom_llm_provider=custom_llm_provider,
     )
 
     model_name = "models/{}".format(model)

litellm/llms/vertex_ai/gemini/transformation.py

@@ -311,6 +311,7 @@ def check_if_part_exists_in_parts(
 def _gemini_convert_messages_with_history(  # noqa: PLR0915
     messages: List[AllMessageValues],
     model: Optional[str] = None,
+    custom_llm_provider: Optional[str] = None,
 ) -> List[ContentType]:
     """
     Converts given messages from OpenAI format to Gemini format
@@ -548,7 +549,9 @@ def _gemini_convert_messages_with_history(  # noqa: PLR0915
                     or assistant_msg.get("function_call") is not None
                 ):  # support assistant tool invoke conversion
                     gemini_tool_call_parts = convert_to_gemini_tool_call_invoke(
-                        assistant_msg, model=model
+                        assistant_msg,
+                        model=model,
+                        custom_llm_provider=custom_llm_provider,
                     )
                     ## check if gemini_tool_call already exists in assistant_content
                     for gemini_tool_call_part in gemini_tool_call_parts:
@@ -607,7 +610,10 @@ def _gemini_convert_messages_with_history(  # noqa: PLR0915
                 and messages[msg_i]["role"] in tool_call_message_roles
             ):
                 _part = convert_to_gemini_tool_call_result(
-                    messages[msg_i], last_message_with_tool_calls  # type: ignore
+                    messages[msg_i],  # type: ignore
+                    last_message_with_tool_calls,  # type: ignore
+                    model=model,
+                    custom_llm_provider=custom_llm_provider,
                 )
                 msg_i += 1
                 # Handle both single part and list of parts (for Computer Use with images)

litellm/llms/vertex_ai/gemini/vertex_and_google_ai_studio_gemini.py

@@ -280,6 +280,8 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
         - gemini-3-pro-preview
         - gemini-3-flash
         - gemini-3-flash-preview (Gemini 3 Flash)
+        - gemini-3.1-pro-preview, gemini-3.1-flash, gemini-3.1-flash-lite-preview
+        - gemini-3.5-flash
         - Any future Gemini 3.x models
         """
         # Check for Gemini 3 models
@@ -287,6 +289,20 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
             return True
         return False
 
+    @staticmethod
+    def _forward_gemini_function_call_id(
+        model: str, custom_llm_provider: Optional[str] = None
+    ) -> bool:
+        """
+        Whether to include `id` on function_call / function_response parts.
+
+        Gemini 3+ on Google AI Studio accepts (and returns) `id` for strict
+        tool-call matching. Vertex AI rejects the field with HTTP 400.
+        """
+        if custom_llm_provider != "gemini":
+            return False
+        return VertexGeminiConfig._is_gemini_3_or_newer(model)
+
     def _supports_penalty_parameters(self, model: str) -> bool:
         # Gemini 3 models do not support penalty parameters
         if VertexGeminiConfig._is_gemini_3_or_newer(model):
@@ -300,6 +316,7 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
         supported_params = [
             "temperature",
             "top_p",
+            "top_k",
             "max_tokens",
             "max_completion_tokens",
             "stream",
@@ -363,6 +380,66 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
         """
         return Tools(googleSearch={})
 
+    @staticmethod
+    def _search_tool_keys() -> set:
+        return {
+            VertexToolName.GOOGLE_SEARCH.value,
+            VertexToolName.GOOGLE_SEARCH_RETRIEVAL.value,
+            VertexToolName.ENTERPRISE_WEB_SEARCH.value,
+            VertexToolName.URL_CONTEXT.value,
+            "google_search",
+            "google_search_retrieval",
+            "enterprise_web_search",
+            "urlContext",
+        }
+
+    @classmethod
+    def _drop_search_tools_mixed_with_functions(cls, optional_params: dict) -> None:
+        """
+        Drop search tools from optional_params when mixed with function declarations
+        and include_server_side_tool_invocations is not enabled.
+
+        Runs after map_openai_params merges tools and web_search_options so both
+        code paths (single _map_function call vs split tools + web_search_options)
+        get the same conflict resolution.
+        """
+        if optional_params.get("include_server_side_tool_invocations"):
+            return
+
+        tools = optional_params.get("tools")
+        if not isinstance(tools, list) or not tools:
+            return
+
+        search_tool_keys = cls._search_tool_keys()
+        has_function_declarations = any(
+            isinstance(tool, dict) and tool.get("function_declarations")
+            for tool in tools
+        )
+        if not has_function_declarations:
+            return
+
+        has_search_tools = any(
+            isinstance(tool, dict) and any(key in tool for key in search_tool_keys)
+            for tool in tools
+        )
+        if not has_search_tools:
+            return
+
+        verbose_logger.warning(
+            "Vertex AI does not support mixing function declarations with "
+            "search tools (googleSearch, enterpriseWebSearch, urlContext, "
+            "googleSearchRetrieval) in the same request. Dropping search "
+            "tools and keeping function declarations. To use search tools, "
+            "send a request without function calling tools."
+        )
+        optional_params["tools"] = [
+            tool
+            for tool in tools
+            if not (
+                isinstance(tool, dict) and any(key in tool for key in search_tool_keys)
+            )
+        ]
+
     def _map_service_tier_param(self, value: str, optional_params: dict) -> None:
         """
         Map OpenAI service_tier (string) to Gemini serviceTier.
@@ -884,9 +961,10 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
             GeminiThinkingConfig with thinkingLevel and includeThoughts
         """
         # Check if this is gemini-3-flash which supports MINIMAL thinking level
-        # Covers gemini-3-flash, gemini-3-flash-preview, gemini-3.1-flash, gemini-3.1-flash-lite-preview, etc.
+        # Covers gemini-3-flash, gemini-3-flash-preview, gemini-3.1-flash, gemini-3.1-flash-lite-preview,
+        # gemini-3.5-flash, and any future 3.x-flash variants.
         is_gemini3flash = model and (
-            "gemini-3-flash" in model.lower() or "gemini-3.1-flash" in model.lower()
+            "flash" in model.lower() and "gemini-3" in model.lower()
         )
         is_gemini31pro = model and ("gemini-3.1-pro-preview" in model.lower())
         if reasoning_effort == "minimal":
@@ -982,8 +1060,7 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
                     # Follow provider defaults unless explicitly opted into legacy behavior.
                     if litellm.enable_gemini_default_thinking_level_low is True:
                         is_gemini3flash = (
-                            "gemini-3-flash-preview" in model.lower()
-                            or "gemini-3-flash" in model.lower()
+                            "gemini-3" in model.lower() and "flash" in model.lower()
                         )
                         params["thinkingLevel"] = (
                             "minimal" if is_gemini3flash else "low"
@@ -1077,6 +1154,7 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
         model: str,
         drop_params: bool,
     ) -> Dict:
+        gemini_sampling_params_warned: bool = False
         for param, value in non_default_params.items():
             if param == "temperature":
                 if VertexGeminiConfig._is_gemini_3_or_newer(model):
@@ -1086,9 +1164,41 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
                             "can cause infinite loops, degraded reasoning performance, and failure on complex tasks. "
                             "Strongly recommended to use temperature = 1.0 (default)."
                         )
+                    if not gemini_sampling_params_warned:
+                        verbose_logger.warning(
+                            "DeprecationWarning: `temperature`, `top_p`, and `top_k` continue to "
+                            f"function for Gemini 3+ ({model}) but are planned for removal in a "
+                            "future release. Move sampling guidance into the `system` "
+                            "instructions instead."
+                        )
+                        gemini_sampling_params_warned = True
                 optional_params["temperature"] = value
             elif param == "top_p":
+                if (
+                    VertexGeminiConfig._is_gemini_3_or_newer(model)
+                    and not gemini_sampling_params_warned
+                ):
+                    verbose_logger.warning(
+                        "DeprecationWarning: `temperature`, `top_p`, and `top_k` continue to "
+                        f"function for Gemini 3+ ({model}) but are planned for removal in a "
+                        "future release. Move sampling guidance into the `system` "
+                        "instructions instead."
+                    )
+                    gemini_sampling_params_warned = True
                 optional_params["top_p"] = value
+            elif param == "top_k":
+                if (
+                    VertexGeminiConfig._is_gemini_3_or_newer(model)
+                    and not gemini_sampling_params_warned
+                ):
+                    verbose_logger.warning(
+                        "DeprecationWarning: `temperature`, `top_p`, and `top_k` continue to "
+                        f"function for Gemini 3+ ({model}) but are planned for removal in a "
+                        "future release. Move sampling guidance into the `system` "
+                        "instructions instead."
+                    )
+                    gemini_sampling_params_warned = True
+                optional_params["top_k"] = value
             elif (
                 param == "stream" and value is True
             ):  # sending stream = False, can cause it to get passed unchecked and raise issues
@@ -1139,11 +1249,14 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
                 if _tool_choice_value is not None:
                     optional_params["tool_choice"] = _tool_choice_value
             elif param == "parallel_tool_calls":
-                if value is False and not (
-                    drop_params or litellm.drop_params
-                ):  # if drop params is True, then we should just ignore this
-                    self.validate_parallel_tool_calls(value, non_default_params)
-                else:
+                tools_list = non_default_params.get(
+                    "tools", non_default_params.get("functions")
+                )
+                num_tools = len(tools_list) if isinstance(tools_list, list) else 0
+                # Gemini does not support parallel_tool_calls=False with multiple
+                # tools. Drop the param instead of failing — Responses API clients
+                # often send parallel_tool_calls=false by default.
+                if not (value is False and num_tools > 1):
                     optional_params["parallel_tool_calls"] = value
             elif param == "seed":
                 optional_params["seed"] = value
@@ -1216,6 +1329,8 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
             if "temperature" not in optional_params:
                 optional_params["temperature"] = 1.0
 
+        self._drop_search_tools_mixed_with_functions(optional_params)
+
         return optional_params
 
     def get_mapped_special_auth_params(self) -> dict:
@@ -1588,6 +1703,10 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
                 }
                 # Extract thought signature if present
                 thought_signature = part.get("thoughtSignature")
+                # Gemini 3.5+ returns a stable `id` per function call to enable
+                # strict response matching. Preserve it as the OpenAI
+                # tool_call_id so it can be echoed back unchanged.
+                gemini_call_id = part["functionCall"].get("id")
 
                 if is_function_call is True:
                     function_dict: Dict[str, Any] = dict(_function_chunk)
@@ -1605,6 +1724,11 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
                         "function": _function_chunk,
                         "index": cumulative_tool_call_idx,
                     }
+                    # Gemini 3.5+ returns a stable native `id`; prefer it over
+                    # the synthetic call_<uuid> so the same value can be echoed
+                    # back on the matching `functionResponse`.
+                    if gemini_call_id:
+                        _tool_response_chunk["id"] = gemini_call_id
                     # Embed thought signature in ID for OpenAI client compatibility
                     if thought_signature:
                         _tool_response_chunk["provider_specific_fields"] = {  # type: ignore
@@ -2535,7 +2659,11 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
     def _transform_messages(
         self, messages: List[AllMessageValues], model: Optional[str] = None
     ) -> List[ContentType]:
-        return _gemini_convert_messages_with_history(messages=messages, model=model)
+        return _gemini_convert_messages_with_history(
+            messages=messages,
+            model=model,
+            custom_llm_provider="vertex_ai",
+        )
 
     def get_error_class(
         self, error_message: str, status_code: int, headers: Union[Dict, httpx.Headers]

litellm/model_prices_and_context_window_backup.json

@@ -15611,6 +15611,64 @@
         },
         "web_search_billing_unit": "per_query"
     },
+    "vertex_ai/gemini-3.5-flash": {
+        "cache_read_input_token_cost": 1.5e-07,
+        "input_cost_per_token": 1.5e-06,
+        "input_cost_per_audio_token": 1e-06,
+        "litellm_provider": "vertex_ai",
+        "max_audio_length_hours": 8.4,
+        "max_audio_per_prompt": 1,
+        "max_images_per_prompt": 3000,
+        "max_input_tokens": 1048576,
+        "max_output_tokens": 65535,
+        "max_pdf_size_mb": 30,
+        "max_tokens": 65535,
+        "max_video_length": 1,
+        "max_videos_per_prompt": 10,
+        "mode": "chat",
+        "output_cost_per_reasoning_token": 9e-06,
+        "output_cost_per_token": 9e-06,
+        "source": "https://cloud.google.com/vertex-ai/generative-ai/pricing",
+        "supported_endpoints": [
+            "/v1/chat/completions",
+            "/v1/completions",
+            "/v1/batch"
+        ],
+        "supported_modalities": [
+            "text",
+            "image",
+            "audio",
+            "video"
+        ],
+        "supported_output_modalities": [
+            "text"
+        ],
+        "supports_audio_input": true,
+        "supports_function_calling": true,
+        "supports_parallel_function_calling": true,
+        "supports_pdf_input": true,
+        "supports_prompt_caching": true,
+        "supports_reasoning": true,
+        "supports_response_schema": true,
+        "supports_system_messages": true,
+        "supports_tool_choice": true,
+        "supports_url_context": true,
+        "supports_video_input": true,
+        "supports_vision": true,
+        "supports_web_search": true,
+        "supports_native_streaming": true,
+        "input_cost_per_token_priority": 2.7e-06,
+        "input_cost_per_audio_token_priority": 1.8e-06,
+        "output_cost_per_token_priority": 1.62e-05,
+        "cache_read_input_token_cost_priority": 2.7e-07,
+        "supports_service_tier": true,
+        "search_context_cost_per_query": {
+            "search_context_size_low": 0.014,
+            "search_context_size_medium": 0.014,
+            "search_context_size_high": 0.014
+        },
+        "web_search_billing_unit": "per_query"
+    },
     "vertex_ai/gemini-3.1-pro-preview": {
         "cache_read_input_token_cost": 2e-07,
         "cache_read_input_token_cost_above_200k_tokens": 4e-07,
@@ -16988,6 +17046,67 @@
         },
         "web_search_billing_unit": "per_query"
     },
+    "gemini/gemini-3.5-flash": {
+        "cache_read_input_token_cost": 1.5e-07,
+        "input_cost_per_audio_token": 1e-06,
+        "input_cost_per_token": 1.5e-06,
+        "litellm_provider": "gemini",
+        "max_audio_length_hours": 8.4,
+        "max_audio_per_prompt": 1,
+        "max_images_per_prompt": 3000,
+        "max_input_tokens": 1048576,
+        "max_output_tokens": 65535,
+        "max_pdf_size_mb": 30,
+        "max_tokens": 65535,
+        "max_video_length": 1,
+        "max_videos_per_prompt": 10,
+        "mode": "chat",
+        "output_cost_per_reasoning_token": 9e-06,
+        "output_cost_per_token": 9e-06,
+        "rpm": 2000,
+        "source": "https://ai.google.dev/pricing/gemini-3",
+        "supported_endpoints": [
+            "/v1/chat/completions",
+            "/v1/completions",
+            "/v1/batch"
+        ],
+        "supported_modalities": [
+            "text",
+            "image",
+            "audio",
+            "video"
+        ],
+        "supported_output_modalities": [
+            "text"
+        ],
+        "supports_audio_output": false,
+        "supports_audio_input": true,
+        "supports_function_calling": true,
+        "supports_parallel_function_calling": true,
+        "supports_pdf_input": true,
+        "supports_prompt_caching": true,
+        "supports_reasoning": true,
+        "supports_response_schema": true,
+        "supports_system_messages": true,
+        "supports_tool_choice": true,
+        "supports_url_context": true,
+        "supports_video_input": true,
+        "supports_vision": true,
+        "supports_web_search": true,
+        "supports_native_streaming": true,
+        "tpm": 800000,
+        "input_cost_per_token_priority": 2.7e-06,
+        "input_cost_per_audio_token_priority": 1.8e-06,
+        "output_cost_per_token_priority": 1.62e-05,
+        "cache_read_input_token_cost_priority": 2.7e-07,
+        "supports_service_tier": true,
+        "search_context_cost_per_query": {
+            "search_context_size_low": 0.014,
+            "search_context_size_medium": 0.014,
+            "search_context_size_high": 0.014
+        },
+        "web_search_billing_unit": "per_query"
+    },
     "gemini/gemini-3.1-pro-preview": {
         "cache_read_input_token_cost": 2e-07,
         "cache_read_input_token_cost_above_200k_tokens": 4e-07,
@@ -17173,6 +17292,65 @@
         },
         "web_search_billing_unit": "per_query"
     },
+    "gemini-3.5-flash": {
+        "cache_read_input_token_cost": 1.5e-07,
+        "input_cost_per_audio_token": 1e-06,
+        "input_cost_per_token": 1.5e-06,
+        "litellm_provider": "vertex_ai-language-models",
+        "max_audio_length_hours": 8.4,
+        "max_audio_per_prompt": 1,
+        "max_images_per_prompt": 3000,
+        "max_input_tokens": 1048576,
+        "max_output_tokens": 65535,
+        "max_pdf_size_mb": 30,
+        "max_tokens": 65535,
+        "max_video_length": 1,
+        "max_videos_per_prompt": 10,
+        "mode": "chat",
+        "output_cost_per_reasoning_token": 9e-06,
+        "output_cost_per_token": 9e-06,
+        "source": "https://ai.google.dev/pricing/gemini-3",
+        "supported_endpoints": [
+            "/v1/chat/completions",
+            "/v1/completions",
+            "/v1/batch"
+        ],
+        "supported_modalities": [
+            "text",
+            "image",
+            "audio",
+            "video"
+        ],
+        "supported_output_modalities": [
+            "text"
+        ],
+        "supports_audio_output": false,
+        "supports_audio_input": true,
+        "supports_function_calling": true,
+        "supports_parallel_function_calling": true,
+        "supports_pdf_input": true,
+        "supports_prompt_caching": true,
+        "supports_reasoning": true,
+        "supports_response_schema": true,
+        "supports_system_messages": true,
+        "supports_tool_choice": true,
+        "supports_url_context": true,
+        "supports_video_input": true,
+        "supports_vision": true,
+        "supports_web_search": true,
+        "supports_native_streaming": true,
+        "input_cost_per_token_priority": 2.7e-06,
+        "input_cost_per_audio_token_priority": 1.8e-06,
+        "output_cost_per_token_priority": 1.62e-05,
+        "cache_read_input_token_cost_priority": 2.7e-07,
+        "supports_service_tier": true,
+        "search_context_cost_per_query": {
+            "search_context_size_low": 0.014,
+            "search_context_size_medium": 0.014,
+            "search_context_size_high": 0.014
+        },
+        "web_search_billing_unit": "per_query"
+    },
     "gemini/gemini-2.5-pro-preview-tts": {
         "cache_read_input_token_cost": 1.25e-07,
         "cache_read_input_token_cost_above_200k_tokens": 2.5e-07,

litellm/proxy/_types.py

@@ -2713,13 +2713,16 @@ class UserAPIKeyAuth(
         1. Regular API keys from LiteLLM DB
         2. JWT tokens used for connecting to LiteLLM API
         """
-        if api_key.startswith("sk-"):
-            return hash_token(api_key)
+        normalized = api_key
+        if normalized[:7].lower() == "bearer ":
+            normalized = normalized[7:]
+        if normalized.startswith("sk-"):
+            return hash_token(normalized)
         from litellm.proxy.auth.handle_jwt import JWTHandler
 
-        if JWTHandler.is_jwt(token=api_key):
-            return f"hashed-jwt-{hash_token(token=api_key)}"
-        return api_key
+        if JWTHandler.is_jwt(token=normalized):
+            return f"hashed-jwt-{hash_token(token=normalized)}"
+        return normalized
 
     @classmethod
     def get_litellm_internal_health_check_user_api_key_auth(cls) -> "UserAPIKeyAuth":

litellm/proxy/common_request_processing.py

@@ -1266,7 +1266,7 @@ class ProxyBaseLLMRequestProcessing:
                 # (ProxyLogging._fire_deferred_stream_logging) fires the
                 # closure after the full streaming pipeline finishes.
                 # The closure runs non-apply_guardrail hooks on the
-                # assembled response, then fires both logging handlers.
+                # assembled response, then fires success logging.
                 # Only for CustomStreamWrapper — raw async generators from
                 # passthrough routes bypass CSW and would orphan the closure.
                 from litellm.litellm_core_utils.streaming_handler import (
@@ -1387,33 +1387,18 @@ class ProxyBaseLLMRequestProcessing:
                     logging_obj._on_deferred_stream_complete = None  # type: ignore[union-attr]
                     try:
                         asyncio.create_task(
-                            logging_obj.async_success_handler(
+                            logging_obj.dispatch_success_handlers(
                                 response,
                                 cache_hit=None,
                                 start_time=None,
                                 end_time=None,
+                                prefer_async_handlers=True,
                             )
                         )
                     except Exception as e:
                         verbose_proxy_logger.exception(
                             "Error in orphaned streaming async logging: %s", e
                         )
-                    try:
-                        from litellm.litellm_core_utils.thread_pool_executor import (
-                            executor as _exc,
-                        )
-
-                        _exc.submit(
-                            logging_obj.success_handler,
-                            response,
-                            cache_hit=None,
-                            start_time=None,
-                            end_time=None,
-                        )
-                    except Exception as e:
-                        verbose_proxy_logger.exception(
-                            "Error in orphaned streaming sync logging: %s", e
-                        )
 
         # Always return the client-requested model name (not provider-prefixed internal identifiers)
         # for OpenAI-compatible responses.
@@ -1615,7 +1600,7 @@ class ProxyBaseLLMRequestProcessing:
     ) -> None:
         """
         Run non-streaming post-call guardrail hooks on an assembled streaming
-        response, then fire both async and sync logging handlers.
+        response, then fire success logging via ``dispatch_success_handlers``.
 
         Called by ProxyLogging._fire_deferred_stream_logging after the full
         streaming pipeline (including unified_guardrail end-of-stream blocks)
@@ -1631,8 +1616,6 @@ class ProxyBaseLLMRequestProcessing:
         Extracted as a static method so tests can call the production
         implementation directly rather than reimplementing the closure.
         """
-        from litellm.litellm_core_utils.thread_pool_executor import executor
-
         _response = assembled_response
         try:
             from litellm.proxy.proxy_server import llm_router as _global_llm_router
@@ -1691,31 +1674,23 @@ class ProxyBaseLLMRequestProcessing:
             )
         finally:
             try:
+                # Proxy streaming always runs in async context and proxy spend
+                # logging is async-only; force async dispatch so DB/spend
+                # callbacks fire regardless of the call-type heuristic in
+                # _is_sync_litellm_request (which only recognizes a subset of
+                # async markers stored in litellm_params).
                 asyncio.create_task(
-                    captured_logging_obj.async_success_handler(
+                    captured_logging_obj.dispatch_success_handlers(
                         _response,
                         cache_hit=cache_hit,
                         start_time=None,
                         end_time=None,
+                        prefer_async_handlers=True,
                     )
                 )
             except Exception as e:
                 verbose_proxy_logger.exception(
-                    "Error in deferred streaming async logging: %s",
-                    e,
-                )
-
-            try:
-                executor.submit(
-                    captured_logging_obj.success_handler,
-                    _response,
-                    cache_hit=cache_hit,
-                    start_time=None,
-                    end_time=None,
-                )
-            except Exception as e:
-                verbose_proxy_logger.exception(
-                    "Error in deferred streaming sync logging: %s",
+                    "Error in deferred streaming success logging: %s",
                     e,
                 )
 

litellm/proxy/common_utils/reset_budget_job.py

@@ -414,6 +414,72 @@ class ResetBudgetJob:
         )
         return [LiteLLM_EndUserTable(**row.dict()) for row in rows]
 
+    async def _write_key_reset_updates(
+        self, updated_keys: List[LiteLLM_VerificationToken]
+    ) -> None:
+        """
+        Write per-row {spend, budget_reset_at} updates for keys.
+
+        Avoids the batched full-model update path, which trips
+        prisma.errors.DataError on any row carrying object_permission_id or
+        budget_limits (see #27730). Both fields are rejected by Prisma's
+        update input type for LiteLLM_VerificationToken, and the failure
+        aborts the entire batch — silently leaving spend over the cap and
+        budget_reset_at unchanged forever.
+        """
+        batcher = self.prisma_client.db.batch_()
+        for k in updated_keys:
+            token = getattr(k, "token", None)
+            if token is None:
+                continue
+            batcher.litellm_verificationtoken.update(
+                where={"token": token},
+                data={"spend": 0, "budget_reset_at": k.budget_reset_at},
+            )
+        await batcher.commit()
+
+    async def _write_user_reset_updates(
+        self, updated_users: List[LiteLLM_UserTable]
+    ) -> None:
+        """
+        Write per-row {spend, budget_reset_at} updates for users.
+
+        Mirrors _write_key_reset_updates — avoids the full-model update path
+        that trips Prisma's DataError on rows carrying unrecognised fields
+        (see #27730).
+        """
+        batcher = self.prisma_client.db.batch_()
+        for u in updated_users:
+            user_id = getattr(u, "user_id", None)
+            if user_id is None:
+                continue
+            batcher.litellm_usertable.update(
+                where={"user_id": user_id},
+                data={"spend": 0, "budget_reset_at": u.budget_reset_at},
+            )
+        await batcher.commit()
+
+    async def _write_team_reset_updates(
+        self, updated_teams: List[LiteLLM_TeamTable]
+    ) -> None:
+        """
+        Write per-row {spend, budget_reset_at} updates for teams.
+
+        Mirrors _write_key_reset_updates — avoids the full-model update path
+        that trips Prisma's DataError on rows carrying unrecognised fields
+        (see #27730).
+        """
+        batcher = self.prisma_client.db.batch_()
+        for t in updated_teams:
+            team_id = getattr(t, "team_id", None)
+            if team_id is None:
+                continue
+            batcher.litellm_teamtable.update(
+                where={"team_id": team_id},
+                data={"spend": 0, "budget_reset_at": t.budget_reset_at},
+            )
+        await batcher.commit()
+
     async def reset_budget_for_litellm_keys(self):
         """
         Resets the budget for all the litellm keys
@@ -455,11 +521,7 @@ class ResetBudgetJob:
                 )
 
                 if updated_keys:
-                    await self.prisma_client.update_data(
-                        query_type="update_many",
-                        data_list=updated_keys,
-                        table_name="key",
-                    )
+                    await self._write_key_reset_updates(updated_keys=updated_keys)
                     for k in updated_keys:
                         token = getattr(k, "token", None)
                         if token:
@@ -544,11 +606,7 @@ class ResetBudgetJob:
                     "Updated users %s", json.dumps(updated_users, indent=4, default=str)
                 )
                 if updated_users:
-                    await self.prisma_client.update_data(
-                        query_type="update_many",
-                        data_list=updated_users,
-                        table_name="user",
-                    )
+                    await self._write_user_reset_updates(updated_users=updated_users)
                     for u in updated_users:
                         user_id = getattr(u, "user_id", None)
                         if user_id:
@@ -641,11 +699,7 @@ class ResetBudgetJob:
                     "Updated teams %s", json.dumps(updated_teams, indent=4, default=str)
                 )
                 if updated_teams:
-                    await self.prisma_client.update_data(
-                        query_type="update_many",
-                        data_list=updated_teams,
-                        table_name="team",
-                    )
+                    await self._write_team_reset_updates(updated_teams=updated_teams)
                     for t in updated_teams:
                         team_id = getattr(t, "team_id", None)
                         if team_id:
@@ -816,49 +870,16 @@ class ResetBudgetJob:
         """
         In-place, updates spend=0, and sets budget_reset_at to current_time + budget_duration
 
-        Common logic for resetting budget for a team, user, or key
+        Common logic for resetting budget for a team, user, or key.
+
+        Spend-counter invalidation happens in the caller, AFTER the DB write
+        commits. Zeroing the counter here would open a bypass window when the
+        DB write fails: get_current_spend reads 0 from Redis while the DB
+        still holds the pre-reset value, admitting requests past the cap.
         """
         try:
             item.spend = 0.0
-
-            # Reset the cross-pod spend counter.
-            # Reset Redis directly (not via DualCache) so a Redis failure
-            # doesn't silently leave a stale counter that get_current_spend
-            # would read as authoritative, permanently blocking the user.
-            from litellm.proxy.proxy_server import spend_counter_cache
-
-            counter_key = None
-            if item_type == "key" and hasattr(item, "token") and item.token is not None:  # type: ignore[union-attr]
-                counter_key = f"spend:key:{item.token}"  # type: ignore[union-attr]
-            elif (
-                item_type == "team"
-                and hasattr(item, "team_id")
-                and item.team_id is not None  # type: ignore[union-attr]
-            ):
-                counter_key = f"spend:team:{item.team_id}"  # type: ignore[union-attr]
-
-            if counter_key is not None:
-                # Always reset in-memory (local fallback)
-                spend_counter_cache.in_memory_cache.set_cache(
-                    key=counter_key, value=0.0
-                )
-                # Explicitly reset Redis with warning on failure
-                if spend_counter_cache.redis_cache is not None:
-                    try:
-                        await spend_counter_cache.redis_cache.async_set_cache(
-                            key=counter_key, value=0.0
-                        )
-                    except Exception as redis_err:
-                        verbose_proxy_logger.warning(
-                            "Failed to reset spend counter in Redis for %s key=%s: %s. "
-                            "Budget may be over-enforced until counter expires.",
-                            item_type,
-                            counter_key,
-                            redis_err,
-                        )
-
             if hasattr(item, "budget_duration") and item.budget_duration is not None:
-                # Get standardized reset time based on budget duration
                 from litellm.proxy.common_utils.timezone_utils import (
                     get_budget_reset_time,
                 )

litellm/proxy/db/spend_counter_reseed.py

@@ -178,15 +178,28 @@ class SpendCounterReseed:
             if db_spend is None:
                 return None
             # Warm even when 0 so subsequent reads hit cache, not DB.
+            #
+            # Seed via SET NX (cross-pod safe): only one pod initializes the
+            # Redis key with db_spend; concurrent seeders read the winner's
+            # value. INCRBYFLOAT-of-db_spend from N pods would multiply the
+            # counter (N x db_spend) and trigger spurious budget alerts.
+            current_value: float = float(db_spend)
             try:
                 if spend_counter_cache.redis_cache is not None:
-                    current_value = (
-                        await spend_counter_cache.redis_cache.async_increment(
-                            key=counter_key,
-                            value=db_spend,
-                            refresh_ttl=True,
-                        )
+                    seeded = await spend_counter_cache.redis_cache.async_set_cache(
+                        key=counter_key,
+                        value=db_spend,
+                        nx=True,
                     )
+                    if seeded:
+                        current_value = float(db_spend)
+                    else:
+                        cached = await spend_counter_cache.redis_cache.async_get_cache(
+                            key=counter_key
+                        )
+                        current_value = (
+                            float(cached) if cached is not None else float(db_spend)
+                        )
                     spend_counter_cache.in_memory_cache.set_cache(
                         key=counter_key,
                         value=current_value,
@@ -202,7 +215,7 @@ class SpendCounterReseed:
                 )
                 if require_cache_warm:
                     raise
-            return db_spend
+            return current_value
 
     @staticmethod
     async def window_from_spend_logs(

litellm/proxy/pass_through_endpoints/streaming_handler.py

@@ -1,13 +1,12 @@
 import asyncio
 from datetime import datetime
-from typing import List, Optional
+from typing import List, Optional, Tuple
 
 import httpx
 
 import litellm
 from litellm._logging import verbose_proxy_logger
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
-from litellm.litellm_core_utils.thread_pool_executor import executor
 from litellm.proxy._types import PassThroughEndpointLoggingResultValues
 from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
 from litellm.types.passthrough_endpoints.pass_through_endpoints import EndpointType
@@ -115,30 +114,81 @@ class PassThroughStreamingHandler:
         - OpenAI
         """
         try:
-            all_chunks = PassThroughStreamingHandler._convert_raw_bytes_to_str_lines(
-                raw_bytes
+            (
+                standard_logging_response_object,
+                kwargs,
+            ) = PassThroughStreamingHandler._build_passthrough_logging_result(
+                litellm_logging_obj=litellm_logging_obj,
+                passthrough_success_handler_obj=passthrough_success_handler_obj,
+                url_route=url_route,
+                request_body=request_body,
+                endpoint_type=endpoint_type,
+                start_time=start_time,
+                raw_bytes=raw_bytes,
+                end_time=end_time,
+                model=model,
             )
-            standard_logging_response_object: Optional[
-                PassThroughEndpointLoggingResultValues
-            ] = None
-            kwargs: dict = {}
-            if endpoint_type == EndpointType.ANTHROPIC:
-                anthropic_passthrough_logging_handler_result = AnthropicPassthroughLoggingHandler._handle_logging_anthropic_collected_chunks(
-                    litellm_logging_obj=litellm_logging_obj,
-                    passthrough_success_handler_obj=passthrough_success_handler_obj,
-                    url_route=url_route,
-                    request_body=request_body,
-                    endpoint_type=endpoint_type,
-                    start_time=start_time,
-                    all_chunks=all_chunks,
-                    end_time=end_time,
-                )
-                standard_logging_response_object = (
-                    anthropic_passthrough_logging_handler_result["result"]
-                )
-                kwargs = anthropic_passthrough_logging_handler_result["kwargs"]
-            elif endpoint_type == EndpointType.VERTEX_AI:
-                vertex_passthrough_logging_handler_result = VertexPassthroughLoggingHandler._handle_logging_vertex_collected_chunks(
+            # Always reached from an async context (anthropic_messages,
+            # google_genai, and proxy pass-through stream tasks). prefer_async_handlers
+            # keeps async-only loggers running even when call_type isn't pass_through
+            # and litellm_params lacks an async flag (e.g. aanthropic_messages).
+            await litellm_logging_obj.dispatch_success_handlers(
+                result=standard_logging_response_object,
+                start_time=start_time,
+                end_time=end_time,
+                cache_hit=False,
+                prefer_async_handlers=True,
+                **kwargs,
+            )
+        except Exception as e:
+            verbose_proxy_logger.error(
+                f"Error in _route_streaming_logging_to_handler: {str(e)}"
+            )
+
+    @staticmethod
+    def _build_passthrough_logging_result(
+        litellm_logging_obj: LiteLLMLoggingObj,
+        passthrough_success_handler_obj: PassThroughEndpointLogging,
+        url_route: str,
+        request_body: dict,
+        endpoint_type: EndpointType,
+        start_time: datetime,
+        raw_bytes: List[bytes],
+        end_time: datetime,
+        model: Optional[str],
+    ) -> Tuple[PassThroughEndpointLoggingResultValues, dict]:
+        """
+        Synchronous, CPU-bound reconstruction of the standard logging payload
+        from collected raw SSE bytes. Extracted from
+        _route_streaming_logging_to_handler so the per-endpoint dispatch can
+        be unit-tested in isolation. Still invoked synchronously on the event
+        loop; an off-loop dispatch is a future change, not part of this PR.
+        """
+        all_chunks = PassThroughStreamingHandler._convert_raw_bytes_to_str_lines(
+            raw_bytes
+        )
+        standard_logging_response_object: Optional[
+            PassThroughEndpointLoggingResultValues
+        ] = None
+        kwargs: dict = {}
+        if endpoint_type == EndpointType.ANTHROPIC:
+            anthropic_passthrough_logging_handler_result = AnthropicPassthroughLoggingHandler._handle_logging_anthropic_collected_chunks(
+                litellm_logging_obj=litellm_logging_obj,
+                passthrough_success_handler_obj=passthrough_success_handler_obj,
+                url_route=url_route,
+                request_body=request_body,
+                endpoint_type=endpoint_type,
+                start_time=start_time,
+                all_chunks=all_chunks,
+                end_time=end_time,
+            )
+            standard_logging_response_object = (
+                anthropic_passthrough_logging_handler_result["result"]
+            )
+            kwargs = anthropic_passthrough_logging_handler_result["kwargs"]
+        elif endpoint_type == EndpointType.VERTEX_AI:
+            vertex_passthrough_logging_handler_result = (
+                VertexPassthroughLoggingHandler._handle_logging_vertex_collected_chunks(
                     litellm_logging_obj=litellm_logging_obj,
                     passthrough_success_handler_obj=passthrough_success_handler_obj,
                     url_route=url_route,
@@ -149,12 +199,14 @@ class PassThroughStreamingHandler:
                     end_time=end_time,
                     model=model,
                 )
-                standard_logging_response_object = (
-                    vertex_passthrough_logging_handler_result["result"]
-                )
-                kwargs = vertex_passthrough_logging_handler_result["kwargs"]
-            elif endpoint_type == EndpointType.OPENAI:
-                openai_passthrough_logging_handler_result = OpenAIPassthroughLoggingHandler._handle_logging_openai_collected_chunks(
+            )
+            standard_logging_response_object = (
+                vertex_passthrough_logging_handler_result["result"]
+            )
+            kwargs = vertex_passthrough_logging_handler_result["kwargs"]
+        elif endpoint_type == EndpointType.OPENAI:
+            openai_passthrough_logging_handler_result = (
+                OpenAIPassthroughLoggingHandler._handle_logging_openai_collected_chunks(
                     litellm_logging_obj=litellm_logging_obj,
                     passthrough_success_handler_obj=passthrough_success_handler_obj,
                     url_route=url_route,
@@ -164,40 +216,17 @@ class PassThroughStreamingHandler:
                     all_chunks=all_chunks,
                     end_time=end_time,
                 )
-                standard_logging_response_object = (
-                    openai_passthrough_logging_handler_result["result"]
-                )
-                kwargs = openai_passthrough_logging_handler_result["kwargs"]
+            )
+            standard_logging_response_object = (
+                openai_passthrough_logging_handler_result["result"]
+            )
+            kwargs = openai_passthrough_logging_handler_result["kwargs"]
 
-            if standard_logging_response_object is None:
-                standard_logging_response_object = StandardPassThroughResponseObject(
-                    response=f"cannot parse chunks to standard response object. Chunks={all_chunks}"
-                )
-            await litellm_logging_obj.async_success_handler(
-                result=standard_logging_response_object,
-                start_time=start_time,
-                end_time=end_time,
-                cache_hit=False,
-                **kwargs,
-            )
-            if (
-                litellm_logging_obj._should_run_sync_callbacks_for_async_calls()
-                is False
-            ):
-                return
-
-            executor.submit(
-                litellm_logging_obj.success_handler,
-                result=standard_logging_response_object,
-                end_time=end_time,
-                cache_hit=False,
-                start_time=start_time,
-                **kwargs,
-            )
-        except Exception as e:
-            verbose_proxy_logger.error(
-                f"Error in _route_streaming_logging_to_handler: {str(e)}"
+        if standard_logging_response_object is None:
+            standard_logging_response_object = StandardPassThroughResponseObject(
+                response=f"cannot parse chunks to standard response object. Chunks={all_chunks}"
             )
+        return standard_logging_response_object, kwargs
 
     @staticmethod
     def _extract_model_for_cost_injection(

litellm/proxy/pass_through_endpoints/success_handler.py

@@ -11,7 +11,6 @@ from litellm.types.passthrough_endpoints.pass_through_endpoints import (
     PassthroughStandardLoggingPayload,
 )
 from litellm.types.utils import StandardPassThroughResponseObject
-from litellm.utils import executor as thread_pool_executor
 
 from .llm_provider_handlers.anthropic_passthrough_logging_handler import (
     AnthropicPassthroughLoggingHandler,
@@ -94,19 +93,15 @@ class PassThroughEndpointLogging:
         cache_hit: bool,
         **kwargs,
     ):
-        """Helper function to handle both sync and async logging operations"""
-        # Submit to thread pool for sync logging
-        thread_pool_executor.submit(
-            logging_obj.success_handler,
-            standard_logging_response_object,
-            start_time,
-            end_time,
-            cache_hit,
-            **kwargs,
-        )
-
-        # Handle async logging
-        await logging_obj.async_success_handler(
+        """Log pass-through success via the shared async dispatch path."""
+        # Always reached from pass_through_async_success_handler, which runs in
+        # an async context. call_type is "pass_through_endpoint" here, so the
+        # passthrough guard in dispatch_success_handlers already forces the
+        # async handler to run; pass prefer_async_handlers explicitly to match
+        # the streaming sibling (_route_streaming_logging_to_handler) and keep
+        # async-only loggers (e.g. the proxy spend logger) firing regardless of
+        # how the call-type classification evolves.
+        await logging_obj.dispatch_success_handlers(
             result=(
                 json.dumps(result)
                 if isinstance(result, dict)
@@ -115,6 +110,7 @@ class PassThroughEndpointLogging:
             start_time=start_time,
             end_time=end_time,
             cache_hit=False,
+            prefer_async_handlers=True,
             **kwargs,
         )
 

litellm/types/llms/vertex_ai.py

@@ -14,13 +14,19 @@ from litellm.types.llms.openai import EmbeddingInput
 GeminiEmbeddingInput = Union[EmbeddingInput, List[List[str]]]
 
 
-class FunctionResponse(TypedDict):
-    name: str
+class FunctionResponse(TypedDict, total=False):
+    # `id` correlates this response with the originating `functionCall` part.
+    # Supported on Google AI Studio Gemini 3.5+; Vertex AI rejects this field.
+    id: str
+    name: Required[str]
     response: Optional[dict]
 
 
-class FunctionCall(TypedDict):
-    name: str
+class FunctionCall(TypedDict, total=False):
+    # `id` correlates the corresponding `functionResponse` on Google AI Studio
+    # Gemini 3.5+. Vertex AI and older Gemini models omit/reject this field.
+    id: str
+    name: Required[str]
     args: Optional[dict]
 
 
@@ -45,8 +51,11 @@ class PartType(TypedDict, total=False):
     media_resolution: Literal["low", "medium", "high"]
 
 
-class HttpxFunctionCall(TypedDict):
-    name: str
+class HttpxFunctionCall(TypedDict, total=False):
+    # `id` correlates the corresponding `functionResponse` on Google AI Studio
+    # Gemini 3.5+. Vertex AI and older Gemini models omit/reject this field.
+    id: str
+    name: Required[str]
     args: dict
 
 

litellm/types/utils.py

@@ -3148,6 +3148,7 @@ all_litellm_params = (
         "allowed_openai_params",
         "litellm_session_id",
         "use_litellm_proxy",
+        "use_chat_completions_api",
         "prompt_label",
         "shared_session",
         "search_tool_name",

model_prices_and_context_window.json

@@ -15645,6 +15645,64 @@
         },
         "web_search_billing_unit": "per_query"
     },
+    "vertex_ai/gemini-3.5-flash": {
+        "cache_read_input_token_cost": 1.5e-07,
+        "input_cost_per_token": 1.5e-06,
+        "input_cost_per_audio_token": 1e-06,
+        "litellm_provider": "vertex_ai",
+        "max_audio_length_hours": 8.4,
+        "max_audio_per_prompt": 1,
+        "max_images_per_prompt": 3000,
+        "max_input_tokens": 1048576,
+        "max_output_tokens": 65535,
+        "max_pdf_size_mb": 30,
+        "max_tokens": 65535,
+        "max_video_length": 1,
+        "max_videos_per_prompt": 10,
+        "mode": "chat",
+        "output_cost_per_reasoning_token": 9e-06,
+        "output_cost_per_token": 9e-06,
+        "source": "https://cloud.google.com/vertex-ai/generative-ai/pricing",
+        "supported_endpoints": [
+            "/v1/chat/completions",
+            "/v1/completions",
+            "/v1/batch"
+        ],
+        "supported_modalities": [
+            "text",
+            "image",
+            "audio",
+            "video"
+        ],
+        "supported_output_modalities": [
+            "text"
+        ],
+        "supports_audio_input": true,
+        "supports_function_calling": true,
+        "supports_parallel_function_calling": true,
+        "supports_pdf_input": true,
+        "supports_prompt_caching": true,
+        "supports_reasoning": true,
+        "supports_response_schema": true,
+        "supports_system_messages": true,
+        "supports_tool_choice": true,
+        "supports_url_context": true,
+        "supports_video_input": true,
+        "supports_vision": true,
+        "supports_web_search": true,
+        "supports_native_streaming": true,
+        "input_cost_per_token_priority": 2.7e-06,
+        "input_cost_per_audio_token_priority": 1.8e-06,
+        "output_cost_per_token_priority": 1.62e-05,
+        "cache_read_input_token_cost_priority": 2.7e-07,
+        "supports_service_tier": true,
+        "search_context_cost_per_query": {
+            "search_context_size_low": 0.014,
+            "search_context_size_medium": 0.014,
+            "search_context_size_high": 0.014
+        },
+        "web_search_billing_unit": "per_query"
+    },
     "vertex_ai/gemini-3.1-pro-preview": {
         "cache_read_input_token_cost": 2e-07,
         "cache_read_input_token_cost_above_200k_tokens": 4e-07,
@@ -17022,6 +17080,67 @@
         },
         "web_search_billing_unit": "per_query"
     },
+    "gemini/gemini-3.5-flash": {
+        "cache_read_input_token_cost": 1.5e-07,
+        "input_cost_per_audio_token": 1e-06,
+        "input_cost_per_token": 1.5e-06,
+        "litellm_provider": "gemini",
+        "max_audio_length_hours": 8.4,
+        "max_audio_per_prompt": 1,
+        "max_images_per_prompt": 3000,
+        "max_input_tokens": 1048576,
+        "max_output_tokens": 65535,
+        "max_pdf_size_mb": 30,
+        "max_tokens": 65535,
+        "max_video_length": 1,
+        "max_videos_per_prompt": 10,
+        "mode": "chat",
+        "output_cost_per_reasoning_token": 9e-06,
+        "output_cost_per_token": 9e-06,
+        "rpm": 2000,
+        "source": "https://ai.google.dev/pricing/gemini-3",
+        "supported_endpoints": [
+            "/v1/chat/completions",
+            "/v1/completions",
+            "/v1/batch"
+        ],
+        "supported_modalities": [
+            "text",
+            "image",
+            "audio",
+            "video"
+        ],
+        "supported_output_modalities": [
+            "text"
+        ],
+        "supports_audio_output": false,
+        "supports_audio_input": true,
+        "supports_function_calling": true,
+        "supports_parallel_function_calling": true,
+        "supports_pdf_input": true,
+        "supports_prompt_caching": true,
+        "supports_reasoning": true,
+        "supports_response_schema": true,
+        "supports_system_messages": true,
+        "supports_tool_choice": true,
+        "supports_url_context": true,
+        "supports_video_input": true,
+        "supports_vision": true,
+        "supports_web_search": true,
+        "supports_native_streaming": true,
+        "tpm": 800000,
+        "input_cost_per_token_priority": 2.7e-06,
+        "input_cost_per_audio_token_priority": 1.8e-06,
+        "output_cost_per_token_priority": 1.62e-05,
+        "cache_read_input_token_cost_priority": 2.7e-07,
+        "supports_service_tier": true,
+        "search_context_cost_per_query": {
+            "search_context_size_low": 0.014,
+            "search_context_size_medium": 0.014,
+            "search_context_size_high": 0.014
+        },
+        "web_search_billing_unit": "per_query"
+    },
     "gemini/gemini-3.1-pro-preview": {
         "cache_read_input_token_cost": 2e-07,
         "cache_read_input_token_cost_above_200k_tokens": 4e-07,
@@ -17207,6 +17326,65 @@
         },
         "web_search_billing_unit": "per_query"
     },
+    "gemini-3.5-flash": {
+        "cache_read_input_token_cost": 1.5e-07,
+        "input_cost_per_audio_token": 1e-06,
+        "input_cost_per_token": 1.5e-06,
+        "litellm_provider": "vertex_ai-language-models",
+        "max_audio_length_hours": 8.4,
+        "max_audio_per_prompt": 1,
+        "max_images_per_prompt": 3000,
+        "max_input_tokens": 1048576,
+        "max_output_tokens": 65535,
+        "max_pdf_size_mb": 30,
+        "max_tokens": 65535,
+        "max_video_length": 1,
+        "max_videos_per_prompt": 10,
+        "mode": "chat",
+        "output_cost_per_reasoning_token": 9e-06,
+        "output_cost_per_token": 9e-06,
+        "source": "https://ai.google.dev/pricing/gemini-3",
+        "supported_endpoints": [
+            "/v1/chat/completions",
+            "/v1/completions",
+            "/v1/batch"
+        ],
+        "supported_modalities": [
+            "text",
+            "image",
+            "audio",
+            "video"
+        ],
+        "supported_output_modalities": [
+            "text"
+        ],
+        "supports_audio_output": false,
+        "supports_audio_input": true,
+        "supports_function_calling": true,
+        "supports_parallel_function_calling": true,
+        "supports_pdf_input": true,
+        "supports_prompt_caching": true,
+        "supports_reasoning": true,
+        "supports_response_schema": true,
+        "supports_system_messages": true,
+        "supports_tool_choice": true,
+        "supports_url_context": true,
+        "supports_video_input": true,
+        "supports_vision": true,
+        "supports_web_search": true,
+        "supports_native_streaming": true,
+        "input_cost_per_token_priority": 2.7e-06,
+        "input_cost_per_audio_token_priority": 1.8e-06,
+        "output_cost_per_token_priority": 1.62e-05,
+        "cache_read_input_token_cost_priority": 2.7e-07,
+        "supports_service_tier": true,
+        "search_context_cost_per_query": {
+            "search_context_size_low": 0.014,
+            "search_context_size_medium": 0.014,
+            "search_context_size_high": 0.014
+        },
+        "web_search_billing_unit": "per_query"
+    },
     "gemini/gemini-2.5-pro-preview-tts": {
         "cache_read_input_token_cost": 1.25e-07,
         "cache_read_input_token_cost_above_200k_tokens": 2.5e-07,

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "litellm"
-version = "1.86.2"
+version = "1.86.3"
 description = "Library to easily interface with LLM API providers"
 readme = "README.md"
 requires-python = ">=3.10, <3.14"
@@ -251,7 +251,7 @@ source-exclude = [
 profile = "black"
 
 [tool.commitizen]
-version = "1.86.2"
+version = "1.86.3"
 version_files = [
     "pyproject.toml:^version",
 ]

tests/litellm_utils_tests/test_proxy_budget_reset.py

@@ -22,6 +22,60 @@ from litellm.proxy.common_utils.reset_budget_job import ResetBudgetJob
 # In a real-world scenario, these would be instances of LiteLLM_VerificationToken, LiteLLM_UserTable, etc.
 
 
+def _attrify(d: dict):
+    """
+    Wrap a dict so that attribute access (`.token`, `.user_id`, `.team_id`,
+    etc.) works alongside the existing item-access the fake_reset_* helpers
+    rely on. The reset job's narrow-write helpers use `getattr(item, "token",
+    None)` (et al), which returns None for plain dicts — that would silently
+    skip the row.
+    """
+    class _AttrDict(dict):
+        def __getattr__(self, k):
+            try:
+                return self[k]
+            except KeyError:
+                raise AttributeError(k)
+
+        def __setattr__(self, k, v):
+            self[k] = v
+
+    return _AttrDict(d)
+
+
+def _wire_batcher_for_test(prisma_client):
+    """
+    Wire prisma_client.db.batch_() to return a mock batcher whose .commit() is
+    awaitable and whose per-table .update() calls get captured. The reset job
+    writes key/user/team resets via prisma.db.batch_().<table>.update — not via
+    prisma_client.update_data — so tests must let that batch path complete.
+
+    Returns the list that will accumulate {table, where, data} dicts from
+    each captured update call.
+    """
+    batch_calls = []
+
+    def make_batcher():
+        class _Table:
+            def __init__(self, table_name):
+                self._table_name = table_name
+
+            def update(self, where=None, data=None):
+                batch_calls.append(
+                    {"table": self._table_name, "where": where, "data": data}
+                )
+
+        batcher = MagicMock()
+        batcher.litellm_verificationtoken = _Table("key")
+        batcher.litellm_usertable = _Table("user")
+        batcher.litellm_teamtable = _Table("team")
+        batcher.commit = AsyncMock(return_value=None)
+        return batcher
+
+    prisma_client.db.batch_ = MagicMock(side_effect=make_batcher)
+    return batch_calls
+
+
 @pytest.mark.asyncio
 async def test_reset_budget_keys_partial_failure():
     """
@@ -45,6 +99,9 @@ async def test_reset_budget_keys_partial_failure():
         return_value=[key1, key2, key3, key4, key5, key6]
     )
     prisma_client.update_data = AsyncMock()
+    # Reset job writes key resets via prisma.db.batch_().<table>.update — not
+    # via update_data — so wire that path.
+    batch_calls = _wire_batcher_for_test(prisma_client)
 
     # Using a dummy logging object with async hooks mocked out.
     proxy_logging_obj = MagicMock()
@@ -56,6 +113,15 @@ async def test_reset_budget_keys_partial_failure():
 
     now = datetime.utcnow()
 
+    # token is needed because the new write path uses where={"token": ...}
+    # and _AttrDict makes getattr work alongside item access used by fake_reset_key.
+    for k in [key1, key2, key3, key4, key5, key6]:
+        k.setdefault("token", k["id"])
+    key1, key2, key3, key4, key5, key6 = (
+        _attrify(k) for k in [key1, key2, key3, key4, key5, key6]
+    )
+    prisma_client.get_data = AsyncMock(return_value=[key1, key2, key3, key4, key5, key6])
+
     async def fake_reset_key(key, current_time):
         if key["id"] == "key1":
             # Simulate a failure on key1 (for example, this might be due to an invariant check)
@@ -80,17 +146,17 @@ async def test_reset_budget_keys_partial_failure():
     # Assert that the helper was called for 6 keys
     assert mock_reset_key.call_count == 6
 
-    # Assert that update_data was called once with a list containing all 6 keys
-    prisma_client.update_data.assert_awaited_once()
-    update_call = prisma_client.update_data.call_args
-    assert update_call.kwargs.get("table_name") == "key"
-    updated_keys = update_call.kwargs.get("data_list", [])
-    assert len(updated_keys) == 5
-    assert updated_keys[0]["id"] == "key2"
-    assert updated_keys[1]["id"] == "key3"
-    assert updated_keys[2]["id"] == "key4"
-    assert updated_keys[3]["id"] == "key5"
-    assert updated_keys[4]["id"] == "key6"
+    # Assert that the new narrow write path got 5 batched updates (key1 failed).
+    # update_data must NOT have been called for keys.
+    prisma_client.update_data.assert_not_awaited()
+    key_writes = [c for c in batch_calls if c["table"] == "key"]
+    assert len(key_writes) == 5
+    written_ids = [c["where"]["token"] for c in key_writes]
+    assert written_ids == ["key2", "key3", "key4", "key5", "key6"]
+    # And every write must carry only {spend, budget_reset_at} — never the full row.
+    for c in key_writes:
+        assert set(c["data"].keys()) == {"spend", "budget_reset_at"}
+        assert c["data"]["spend"] == 0
 
     # Verify that the failure logging hook was scheduled (due to the failure for key1)
     failure_hook_calls = (
@@ -125,6 +191,7 @@ async def test_reset_budget_users_partial_failure():
         return_value=[user1, user2, user3, user4, user5, user6]
     )
     prisma_client.update_data = AsyncMock()
+    batch_calls = _wire_batcher_for_test(prisma_client)
 
     proxy_logging_obj = MagicMock()
     proxy_logging_obj.service_logging_obj = MagicMock()
@@ -133,6 +200,15 @@ async def test_reset_budget_users_partial_failure():
 
     job = ResetBudgetJob(proxy_logging_obj, prisma_client)
 
+    # user_id required for the new write path's where clause; _AttrDict so
+    # getattr(u, 'user_id') works alongside the dict access fake_reset_user uses.
+    for u in [user1, user2, user3, user4, user5, user6]:
+        u.setdefault("user_id", u["id"])
+    user1, user2, user3, user4, user5, user6 = (
+        _attrify(u) for u in [user1, user2, user3, user4, user5, user6]
+    )
+    prisma_client.get_data = AsyncMock(return_value=[user1, user2, user3, user4, user5, user6])
+
     async def fake_reset_user(user, current_time):
         if user["id"] == "user1":
             raise Exception("Simulated failure for user1")
@@ -150,16 +226,14 @@ async def test_reset_budget_users_partial_failure():
         await asyncio.sleep(0.1)
 
     assert mock_reset_user.call_count == 6
-    prisma_client.update_data.assert_awaited_once()
-    update_call = prisma_client.update_data.call_args
-    assert update_call.kwargs.get("table_name") == "user"
-    updated_users = update_call.kwargs.get("data_list", [])
-    assert len(updated_users) == 5
-    assert updated_users[0]["id"] == "user2"
-    assert updated_users[1]["id"] == "user3"
-    assert updated_users[2]["id"] == "user4"
-    assert updated_users[3]["id"] == "user5"
-    assert updated_users[4]["id"] == "user6"
+    prisma_client.update_data.assert_not_awaited()
+    user_writes = [c for c in batch_calls if c["table"] == "user"]
+    assert len(user_writes) == 5
+    written_ids = [c["where"]["user_id"] for c in user_writes]
+    assert written_ids == ["user2", "user3", "user4", "user5", "user6"]
+    for c in user_writes:
+        assert set(c["data"].keys()) == {"spend", "budget_reset_at"}
+        assert c["data"]["spend"] == 0
 
     failure_hook_calls = (
         proxy_logging_obj.service_logging_obj.async_service_failure_hook.call_args_list
@@ -308,6 +382,7 @@ async def test_reset_budget_teams_partial_failure():
     prisma_client = MagicMock()
     prisma_client.get_data = AsyncMock(return_value=[team1, team2])
     prisma_client.update_data = AsyncMock()
+    batch_calls = _wire_batcher_for_test(prisma_client)
 
     proxy_logging_obj = MagicMock()
     proxy_logging_obj.service_logging_obj = MagicMock()
@@ -316,6 +391,12 @@ async def test_reset_budget_teams_partial_failure():
 
     job = ResetBudgetJob(proxy_logging_obj, prisma_client)
 
+    # team_id required for the new write path's where clause; _AttrDict for getattr.
+    for t in [team1, team2]:
+        t.setdefault("team_id", t["id"])
+    team1, team2 = _attrify(team1), _attrify(team2)
+    prisma_client.get_data = AsyncMock(return_value=[team1, team2])
+
     async def fake_reset_team(team, current_time):
         if team["id"] == "team1":
             raise Exception("Simulated failure for team1")
@@ -333,12 +414,12 @@ async def test_reset_budget_teams_partial_failure():
         await asyncio.sleep(0.1)
 
     assert mock_reset_team.call_count == 2
-    prisma_client.update_data.assert_awaited_once()
-    update_call = prisma_client.update_data.call_args
-    assert update_call.kwargs.get("table_name") == "team"
-    updated_teams = update_call.kwargs.get("data_list", [])
-    assert len(updated_teams) == 1
-    assert updated_teams[0]["id"] == "team2"
+    prisma_client.update_data.assert_not_awaited()
+    team_writes = [c for c in batch_calls if c["table"] == "team"]
+    assert len(team_writes) == 1
+    assert team_writes[0]["where"] == {"team_id": "team2"}
+    assert set(team_writes[0]["data"].keys()) == {"spend", "budget_reset_at"}
+    assert team_writes[0]["data"]["spend"] == 0
 
     failure_hook_calls = (
         proxy_logging_obj.service_logging_obj.async_service_failure_hook.call_args_list
@@ -402,6 +483,18 @@ async def test_reset_budget_continues_other_categories_on_failure():
 
     prisma_client.get_data = AsyncMock(side_effect=fake_get_data)
     prisma_client.update_data = AsyncMock()
+    batch_calls = _wire_batcher_for_test(prisma_client)
+    # ID fields required by the new write path's where clauses; _AttrDict
+    # lets getattr() see them alongside the item-access fake_reset_* helpers use.
+    for k in [key1, key2]:
+        k.setdefault("token", k["id"])
+    for u in [user1, user2]:
+        u.setdefault("user_id", u["id"])
+    for t in [team1, team2]:
+        t.setdefault("team_id", t["id"])
+    key1, key2 = _attrify(key1), _attrify(key2)
+    user1, user2 = _attrify(user1), _attrify(user2)
+    team1, team2 = _attrify(team1), _attrify(team2)
     # Mock db.litellm_verificationtoken.update_many (used by reset_budget_for_keys_linked_to_budgets)
     prisma_client.db.litellm_verificationtoken.update_many = AsyncMock(
         return_value={"count": 0}
@@ -488,32 +581,29 @@ async def test_reset_budget_continues_other_categories_on_failure():
         "team_membership",
     }
 
-    # Verify that update_data was called three times (one per category, enduser update includes two)
-    assert prisma_client.update_data.await_count == 5
+    # After the fix, keys/users/teams write via prisma.db.batch_().<table>.update,
+    # so only budget + enduser still go through update_data.
     calls = prisma_client.update_data.await_args_list
-
-    # Check keys update: both keys succeed.
-    keys_call = calls[0]
-    assert keys_call.kwargs.get("table_name") == "key"
-    assert len(keys_call.kwargs.get("data_list", [])) == 2
-
-    # Check users update: only user2 succeeded.
-    users_call = calls[1]
-    assert users_call.kwargs.get("table_name") == "user"
-    users_updated = users_call.kwargs.get("data_list", [])
-    assert len(users_updated) == 1
-    assert users_updated[0]["id"] == "user2"
-
-    # Check teams update: both teams succeed.
-    teams_call = calls[2]
-    assert teams_call.kwargs.get("table_name") == "team"
-    assert len(teams_call.kwargs.get("data_list", [])) == 2
+    update_data_tables = [c.kwargs.get("table_name") for c in calls]
+    assert sorted(update_data_tables) == ["budget", "enduser"]
 
     # Check enduser update: enduser succeed.
-    enduser_call = calls[4]
-    assert enduser_call.kwargs.get("table_name") == "enduser"
+    enduser_call = next(c for c in calls if c.kwargs.get("table_name") == "enduser")
     assert len(enduser_call.kwargs.get("data_list", [])) == 1
 
+    # Check the new batch write path: 2 keys + 1 user (user1 failed) + 2 teams.
+    key_writes = [c for c in batch_calls if c["table"] == "key"]
+    user_writes = [c for c in batch_calls if c["table"] == "user"]
+    team_writes = [c for c in batch_calls if c["table"] == "team"]
+    assert len(key_writes) == 2
+    assert len(user_writes) == 1
+    assert user_writes[0]["where"] == {"user_id": "user2"}
+    assert len(team_writes) == 2
+    # Every batched write must carry only the two reset fields, never the full row.
+    for c in key_writes + user_writes + team_writes:
+        assert set(c["data"].keys()) == {"spend", "budget_reset_at"}
+        assert c["data"]["spend"] == 0
+
 
 # ---------------------------------------------------------------------------
 # Additional tests for service logger behavior (keys, users, teams, endusers)
@@ -527,12 +617,13 @@ async def test_service_logger_keys_success():
     logger success hook is called with the correct event metadata and no exception is logged.
     """
     keys = [
-        {"id": "key1", "spend": 10.0, "budget_duration": 60},
-        {"id": "key2", "spend": 15.0, "budget_duration": 60},
+        {"id": "key1", "spend": 10.0, "budget_duration": 60, "token": "key1"},
+        {"id": "key2", "spend": 15.0, "budget_duration": 60, "token": "key2"},
     ]
     prisma_client = MagicMock()
     prisma_client.get_data = AsyncMock(return_value=keys)
     prisma_client.update_data = AsyncMock()
+    _wire_batcher_for_test(prisma_client)
 
     proxy_logging_obj = MagicMock()
     proxy_logging_obj.service_logging_obj = MagicMock()
@@ -644,12 +735,13 @@ async def test_service_logger_users_success():
     the correct metadata and no exception is logged.
     """
     users = [
-        {"id": "user1", "spend": 20.0, "budget_duration": 120},
-        {"id": "user2", "spend": 25.0, "budget_duration": 120},
+        {"id": "user1", "spend": 20.0, "budget_duration": 120, "user_id": "user1"},
+        {"id": "user2", "spend": 25.0, "budget_duration": 120, "user_id": "user2"},
     ]
     prisma_client = MagicMock()
     prisma_client.get_data = AsyncMock(return_value=users)
     prisma_client.update_data = AsyncMock()
+    _wire_batcher_for_test(prisma_client)
 
     proxy_logging_obj = MagicMock()
     proxy_logging_obj.service_logging_obj = MagicMock()
@@ -756,12 +848,13 @@ async def test_service_logger_teams_success():
     the proper metadata and nothing is logged as an exception.
     """
     teams = [
-        {"id": "team1", "spend": 30.0, "budget_duration": 180},
-        {"id": "team2", "spend": 35.0, "budget_duration": 180},
+        {"id": "team1", "spend": 30.0, "budget_duration": 180, "team_id": "team1"},
+        {"id": "team2", "spend": 35.0, "budget_duration": 180, "team_id": "team2"},
     ]
     prisma_client = MagicMock()
     prisma_client.get_data = AsyncMock(return_value=teams)
     prisma_client.update_data = AsyncMock()
+    _wire_batcher_for_test(prisma_client)
 
     proxy_logging_obj = MagicMock()
     proxy_logging_obj.service_logging_obj = MagicMock()

tests/pass_through_unit_tests/test_unit_test_streaming.py

@@ -97,6 +97,123 @@ async def test_chunk_processor_yields_raw_bytes(endpoint_type, url_route):
     ), "Collected chunks do not match raw chunks"
 
 
+@pytest.mark.asyncio
+async def test_route_streaming_logging_runs_async_handler_for_sdk_passthrough():
+    """
+    SDK pass-through streaming (anthropic_messages, google generate_content) must run
+    the async success handler so async-only loggers record the assembled stream.
+
+    Regression for duplicate-trace dedupe: dispatch_success_handlers treated these as
+    sync SDK requests because call_type is not ``pass_through_endpoint`` and
+    litellm_params carries no ``acompletion`` flag, so only the sync success_handler
+    ran and CustomLogger.async_log_success_event never fired.
+    """
+    import time
+
+    from litellm.types.utils import CallTypes
+
+    logging_obj = LiteLLMLoggingObj(
+        model="claude-sonnet-4-5",
+        messages=[{"role": "user", "content": "hi"}],
+        stream=True,
+        call_type=CallTypes.anthropic_messages.value,
+        start_time=time.time(),
+        litellm_call_id="test-id",
+        function_id="fn",
+    )
+    logging_obj.model_call_details["litellm_params"] = {"anthropic_messages": True}
+
+    with (
+        patch.object(
+            PassThroughStreamingHandler,
+            "_build_passthrough_logging_result",
+            return_value=({"id": "slp"}, {}),
+        ),
+        patch.object(
+            logging_obj, "async_success_handler", new_callable=AsyncMock
+        ) as mock_async,
+        patch.object(
+            logging_obj, "success_handler", new_callable=MagicMock
+        ) as mock_sync,
+        patch.object(
+            logging_obj,
+            "_should_run_sync_callbacks_for_async_calls",
+            return_value=False,
+        ),
+    ):
+        await PassThroughStreamingHandler._route_streaming_logging_to_handler(
+            litellm_logging_obj=logging_obj,
+            passthrough_success_handler_obj=MagicMock(),
+            url_route="/v1/messages",
+            request_body={},
+            endpoint_type=EndpointType.ANTHROPIC,
+            start_time=datetime.now(),
+            raw_bytes=[],
+            end_time=datetime.now(),
+        )
+
+    mock_async.assert_awaited_once()
+    mock_sync.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_handle_logging_runs_async_handler_for_passthrough():
+    """
+    Non-streaming pass-through logging (_handle_logging) must always run the
+    async success handler so async-only loggers (e.g. the proxy spend logger)
+    record the request.
+
+    _handle_logging is only ever reached from pass_through_async_success_handler
+    (an async context), so it forces async dispatch via prefer_async_handlers.
+    This pins that contract independent of the call-type classification: even a
+    call_type that _is_sync_litellm_request would classify as sync (here
+    "completion" with no async marker in litellm_params) must still reach
+    async_success_handler. Without prefer_async_handlers=True the sync-only
+    branch would return early and async_log_success_event would never fire.
+    """
+    import time
+
+    from litellm.types.utils import CallTypes
+
+    logging_obj = LiteLLMLoggingObj(
+        model="claude-sonnet-4-5",
+        messages=[{"role": "user", "content": "hi"}],
+        stream=False,
+        call_type=CallTypes.completion.value,
+        start_time=time.time(),
+        litellm_call_id="test-id",
+        function_id="fn",
+    )
+    logging_obj.model_call_details["litellm_params"] = {}
+
+    handler = PassThroughEndpointLogging()
+
+    with (
+        patch.object(
+            logging_obj, "async_success_handler", new_callable=AsyncMock
+        ) as mock_async,
+        patch.object(
+            logging_obj, "success_handler", new_callable=MagicMock
+        ) as mock_sync,
+        patch.object(
+            logging_obj,
+            "_should_run_sync_callbacks_for_async_calls",
+            return_value=False,
+        ),
+    ):
+        await handler._handle_logging(
+            logging_obj=logging_obj,
+            standard_logging_response_object={"id": "slp"},
+            result="",
+            start_time=datetime.now(),
+            end_time=datetime.now(),
+            cache_hit=False,
+        )
+
+    mock_async.assert_awaited_once()
+    mock_sync.assert_not_called()
+
+
 def test_convert_raw_bytes_to_str_lines():
     """
     Test that the _convert_raw_bytes_to_str_lines method correctly converts raw bytes to a list of strings

tests/proxy_unit_tests/test_proxy_reject_logging.py

@@ -95,6 +95,21 @@ router = Router(
 )
 
 
+def _register_proxy_test_logger(callback_logger: testLogger) -> None:
+    """
+    Register the test logger on global callback lists.
+
+    ``function_setup`` dedupes by object identity; each parametrized case
+    constructs a new ``testLogger`` and must replace the global lists, not
+    only ``litellm.callbacks``.
+    """
+    litellm.callbacks = [callback_logger]
+    litellm.success_callback = [callback_logger]
+    litellm.failure_callback = [callback_logger]
+    litellm._async_success_callback = [callback_logger]
+    litellm._async_failure_callback = [callback_logger]
+
+
 @pytest.mark.parametrize(
     "route, body",
     [
@@ -115,7 +130,7 @@ router = Router(
             "/v1/embeddings",
             {
                 "input": "The food was delicious and the waiter...",
-                "model": "text-embedding-ada-002",
+                "model": "fake-model",
                 "encoding_format": "float",
             },
         ),
@@ -133,7 +148,7 @@ async def test_chat_completion_request_with_redaction(route, body):
 
     setattr(proxy_server, "llm_router", router)
     _test_logger = testLogger()
-    litellm.callbacks = [_test_logger]
+    _register_proxy_test_logger(_test_logger)
     litellm.set_verbose = True
 
     # Prepare the query string

tests/test_litellm/litellm_core_utils/test_litellm_logging.py

@@ -1,6 +1,7 @@
 import os
 import sys
-from unittest.mock import MagicMock, patch
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -786,6 +787,211 @@ def test_success_handler_runs_sync_callbacks_for_sync_requests(logging_obj, call
     dummy_logger.log_stream_event.assert_not_called()
 
 
+def test_is_sync_litellm_request():
+    assert LitellmLogging._is_sync_litellm_request({}) is True
+    assert LitellmLogging._is_sync_litellm_request({"acompletion": True}) is False
+
+
+@pytest.mark.asyncio
+async def test_dispatch_success_handlers_invokes_callbacks_once_for_final_stream(
+    logging_obj,
+):
+    """Second final-stream dispatch must not re-export (CSW + deferred guardrail paths)."""
+    import litellm
+    from litellm.integrations.custom_logger import CustomLogger
+
+    class MockCallback(CustomLogger):
+        pass
+
+    mock_callback = MockCallback()
+    original_async_callbacks = list(litellm._async_success_callback or [])
+    litellm._async_success_callback = [mock_callback]
+
+    result = ModelResponse(
+        id="resp-dedupe",
+        model="gpt-4o-mini",
+        choices=[
+            {
+                "message": {"role": "assistant", "content": "hi"},
+                "finish_reason": "stop",
+                "index": 0,
+            }
+        ],
+        usage={"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
+    )
+
+    try:
+        logging_obj.stream = True
+        logging_obj.model_call_details["litellm_params"] = {"acompletion": True}
+
+        with (
+            patch.object(
+                mock_callback, "async_log_success_event", new_callable=AsyncMock
+            ) as mock_async_log,
+            patch.object(mock_callback, "log_success_event") as mock_sync_log,
+            patch.object(
+                logging_obj,
+                "_success_handler_helper_fn",
+                return_value=(time.time(), time.time(), result),
+            ),
+            patch.object(
+                logging_obj,
+                "_get_assembled_streaming_response",
+                return_value=result,
+            ),
+            patch.object(
+                logging_obj,
+                "_should_run_sync_callbacks_for_async_calls",
+                return_value=True,
+            ),
+        ):
+            await logging_obj.dispatch_success_handlers(result=result)
+            await logging_obj.dispatch_success_handlers(result=result)
+
+        mock_async_log.assert_awaited_once()
+        mock_sync_log.assert_not_called()
+    finally:
+        litellm._async_success_callback = original_async_callbacks
+
+
+@pytest.mark.asyncio
+async def test_dispatch_success_handlers_sync_path_invokes_callback_once_for_final_stream(
+    logging_obj,
+):
+    """Sync dispatch path must also dedupe when dispatch is called twice."""
+    import litellm
+    from litellm.integrations.custom_logger import CustomLogger
+
+    class MockCallback(CustomLogger):
+        pass
+
+    mock_callback = MockCallback()
+    original_success_callbacks = list(litellm.success_callback or [])
+    litellm.success_callback = [mock_callback]
+
+    result = ModelResponse(
+        id="resp-sync-dedupe",
+        model="gpt-4o-mini",
+        choices=[
+            {
+                "message": {"role": "assistant", "content": "hi"},
+                "finish_reason": "stop",
+                "index": 0,
+            }
+        ],
+        usage={"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
+    )
+
+    try:
+        logging_obj.stream = True
+        logging_obj.model_call_details["litellm_params"] = {}
+
+        with (
+            patch.object(mock_callback, "log_success_event") as mock_sync_log,
+            patch.object(
+                mock_callback, "async_log_success_event", new_callable=AsyncMock
+            ) as mock_async_log,
+            patch.object(
+                logging_obj,
+                "_success_handler_helper_fn",
+                return_value=(time.time(), time.time(), result),
+            ),
+            patch.object(
+                logging_obj,
+                "_get_assembled_streaming_response",
+                return_value=result,
+            ),
+        ):
+            await logging_obj.dispatch_success_handlers(result=result)
+            await logging_obj.dispatch_success_handlers(result=result)
+
+        mock_sync_log.assert_called_once()
+        mock_async_log.assert_not_awaited()
+    finally:
+        litellm.success_callback = original_success_callbacks
+
+
+@pytest.mark.asyncio
+async def test_dispatch_prefer_async_handlers_runs_legacy_callbacks(
+    logging_obj,
+):
+    """``prefer_async_handlers`` must not skip executor.submit for string callbacks."""
+    result = ModelResponse(
+        id="resp-prefer-async",
+        model="gpt-4o-mini",
+        choices=[
+            {
+                "message": {"role": "assistant", "content": "hi"},
+                "finish_reason": "stop",
+                "index": 0,
+            }
+        ],
+    )
+
+    logging_obj.stream = True
+    logging_obj.model_call_details["litellm_params"] = {}
+
+    with (
+        patch.object(
+            logging_obj, "async_success_handler", new_callable=AsyncMock
+        ) as mock_async,
+        patch.object(
+            logging_obj, "success_handler", new_callable=MagicMock
+        ) as mock_sync,
+        patch.object(
+            logging_obj,
+            "_should_run_sync_callbacks_for_async_calls",
+            return_value=True,
+        ),
+        patch(
+            "litellm.litellm_core_utils.litellm_logging.executor.submit"
+        ) as mock_submit,
+    ):
+        await logging_obj.dispatch_success_handlers(
+            result=result,
+            prefer_async_handlers=True,
+        )
+
+    mock_async.assert_awaited_once()
+    mock_sync.assert_not_called()
+    mock_submit.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_dispatch_success_handlers_invokes_async_callback_for_pass_through(
+    logging_obj,
+):
+    """Pass-through must use async_success_handler (CustomLogger skips sync success_handler)."""
+    import litellm
+    from litellm.integrations.custom_logger import CustomLogger
+    from litellm.types.utils import CallTypes
+
+    class MockCallback(CustomLogger):
+        pass
+
+    mock_callback = MockCallback()
+    original_async_callbacks = list(litellm._async_success_callback or [])
+    litellm._async_success_callback = [mock_callback]
+
+    logging_obj.call_type = CallTypes.pass_through.value
+    logging_obj.stream = False
+    logging_obj.model_call_details["litellm_params"] = {}
+
+    try:
+        with (
+            patch.object(
+                mock_callback, "async_log_success_event", new_callable=AsyncMock
+            ) as mock_async_log,
+            patch.object(mock_callback, "log_success_event") as mock_sync_log,
+        ):
+            await logging_obj.dispatch_success_handlers(result={"id": "pt-1"})
+
+        mock_async_log.assert_awaited_once()
+        mock_sync_log.assert_not_called()
+    finally:
+        litellm._async_success_callback = original_async_callbacks
+
+
 def test_success_handler_skips_guardrail_logging_hook_when_disabled(logging_obj):
     """Ensure CustomGuardrail logging_hook is skipped when should_run_guardrail is False."""
     import datetime
@@ -1351,7 +1557,7 @@ async def test_e2e_generate_cold_storage_object_key_with_custom_logger_s3_path()
     Test that _generate_cold_storage_object_key uses s3_path from custom logger instance.
     """
     from datetime import datetime, timezone
-    from unittest.mock import MagicMock, patch
+    from unittest.mock import AsyncMock, MagicMock, patch
 
     from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
 
@@ -1404,7 +1610,7 @@ async def test_e2e_generate_cold_storage_object_key_with_logger_no_s3_path():
     Test that _generate_cold_storage_object_key falls back to empty s3_path when logger has no s3_path.
     """
     from datetime import datetime, timezone
-    from unittest.mock import MagicMock, patch
+    from unittest.mock import AsyncMock, MagicMock, patch
 
     from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
 

tests/test_litellm/litellm_core_utils/test_streaming_handler.py

@@ -569,8 +569,6 @@ async def test_streaming_with_usage_and_logging(sync_mode: bool):
                 == final_usage_block
             )
 
-        print(mock_log_success_event.call_args.kwargs.keys())
-
 
 def test_streaming_handler_with_stop_chunk(
     initialized_custom_stream_wrapper: CustomStreamWrapper,
@@ -2036,23 +2034,19 @@ async def test_azure_streaming_role_preserved_with_include_usage(sync_mode: bool
             chunks.append(chunk)
 
     # The prompt_filter chunk should be forwarded with choices=[]
-    assert len(chunks[0].choices) == 0, (
-        f"Expected prompt_filter chunk with choices=[], got {len(chunks[0].choices)} choices"
-    )
+    assert (
+        len(chunks[0].choices) == 0
+    ), f"Expected prompt_filter chunk with choices=[], got {len(chunks[0].choices)} choices"
 
     # At least one chunk must have role='assistant' in its delta
     has_role = any(
-        len(c.choices) > 0
-        and getattr(c.choices[0].delta, "role", None) == "assistant"
+        len(c.choices) > 0 and getattr(c.choices[0].delta, "role", None) == "assistant"
         for c in chunks
     )
     assert has_role, (
         "No chunk contained role='assistant' in delta (issue #24221). "
         "Chunk deltas: "
-        + str([
-            c.choices[0].delta if c.choices else "no choices"
-            for c in chunks
-        ])
+        + str([c.choices[0].delta if c.choices else "no choices" for c in chunks])
     )
 
 

tests/test_litellm/llms/openai/test_use_chat_completions_api_no_leak.py

@@ -0,0 +1,74 @@
+"""
+Regression test for issue #28146.
+
+`use_chat_completions_api` is a LiteLLM-internal control flag (it forces the
+/responses -> /chat/completions bridge). When set as a model-level param in the
+proxy config, it must never be forwarded to the upstream provider's request
+body. OpenAI/Anthropic reject unknown body params with HTTP 400.
+"""
+
+import os
+import sys
+from unittest.mock import MagicMock
+
+sys.path.insert(0, os.path.abspath("../../../.."))
+
+import litellm
+from litellm.types.utils import all_litellm_params
+from litellm.utils import get_non_default_completion_params
+
+
+def test_use_chat_completions_api_is_a_known_litellm_param():
+    assert "use_chat_completions_api" in all_litellm_params
+
+
+def test_use_chat_completions_api_not_forwarded_as_provider_param():
+    forwarded = get_non_default_completion_params(
+        {"use_chat_completions_api": True, "temperature": 0.5}
+    )
+    assert "use_chat_completions_api" not in forwarded
+
+
+def test_completion_does_not_leak_flag_into_provider_request_body():
+    mock_response = MagicMock()
+    mock_response.model_dump.return_value = {
+        "id": "chatcmpl-1",
+        "object": "chat.completion",
+        "created": 1234567890,
+        "model": "gpt-4o-mini",
+        "choices": [
+            {
+                "index": 0,
+                "message": {"role": "assistant", "content": "hi"},
+                "finish_reason": "stop",
+            }
+        ],
+        "usage": {
+            "prompt_tokens": 1,
+            "completion_tokens": 1,
+            "total_tokens": 2,
+        },
+    }
+
+    mock_raw_response = MagicMock()
+    mock_raw_response.headers = {}
+    mock_raw_response.parse.return_value = mock_response
+
+    mock_client = MagicMock()
+    mock_client.chat.completions.with_raw_response.create.return_value = (
+        mock_raw_response
+    )
+
+    litellm.completion(
+        model="openai/gpt-4o-mini",
+        messages=[{"role": "user", "content": "hi"}],
+        use_chat_completions_api=True,
+        api_key="sk-test",
+        client=mock_client,
+    )
+
+    create_kwargs = (
+        mock_client.chat.completions.with_raw_response.create.call_args.kwargs
+    )
+    assert "use_chat_completions_api" not in create_kwargs
+    assert "use_chat_completions_api" not in (create_kwargs.get("extra_body") or {})

tests/test_litellm/llms/vertex_ai/gemini/test_vertex_and_google_ai_studio_gemini.py

@@ -2097,6 +2097,125 @@ def test_is_gemini_3_or_newer():
     assert VertexGeminiConfig._is_gemini_3_or_newer("") == False
 
 
+def test_forward_gemini_function_call_id_vertex_vs_google_ai_studio():
+    """Vertex AI rejects `id` on function_call/function_response; Google AI Studio accepts it on Gemini 3.5+."""
+    from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
+        VertexGeminiConfig,
+    )
+
+    model = "gemini-3.5-flash"
+    assert (
+        VertexGeminiConfig._forward_gemini_function_call_id(model, "vertex_ai") is False
+    )
+    assert (
+        VertexGeminiConfig._forward_gemini_function_call_id(model, "vertex_ai_beta")
+        is False
+    )
+    assert VertexGeminiConfig._forward_gemini_function_call_id(model, "gemini") is True
+    assert VertexGeminiConfig._forward_gemini_function_call_id(model, None) is False
+    assert (
+        VertexGeminiConfig._forward_gemini_function_call_id(
+            "gemini-2.5-flash", "gemini"
+        )
+        is False
+    )
+
+
+def test_vertex_ai_gemini_35_tool_calls_omit_function_call_id():
+    """Regression: Vertex must not send OpenAI tool_call id inside Gemini function_call parts."""
+    from litellm.llms.vertex_ai.gemini.transformation import (
+        _gemini_convert_messages_with_history,
+    )
+
+    messages = [
+        {"role": "user", "content": "Explore this directory"},
+        {
+            "role": "assistant",
+            "content": "",
+            "tool_calls": [
+                {
+                    "id": "call_50e7e0fe0989464a89f188eda443",
+                    "type": "function",
+                    "function": {
+                        "name": "read",
+                        "arguments": '{"filePath": "/tmp"}',
+                    },
+                }
+            ],
+        },
+        {
+            "role": "tool",
+            "tool_call_id": "call_50e7e0fe0989464a89f188eda443",
+            "content": "ok",
+        },
+    ]
+
+    contents = _gemini_convert_messages_with_history(
+        messages=messages,
+        model="gemini-3.5-flash",
+        custom_llm_provider="vertex_ai",
+    )
+
+    for content in contents:
+        for part in content.get("parts", []):
+            fc = part.get("function_call")
+            if fc is not None:
+                assert "id" not in fc, f"Vertex payload must not include id: {fc}"
+            fr = part.get("function_response")
+            if fr is not None:
+                assert "id" not in fr, f"Vertex payload must not include id: {fr}"
+
+
+def test_google_ai_studio_gemini_35_tool_calls_include_function_call_id():
+    from litellm.llms.vertex_ai.gemini.transformation import (
+        _gemini_convert_messages_with_history,
+    )
+
+    tool_call_id = "call_50e7e0fe0989464a89f188eda443"
+    messages = [
+        {"role": "user", "content": "hi"},
+        {
+            "role": "assistant",
+            "content": "",
+            "tool_calls": [
+                {
+                    "id": tool_call_id,
+                    "type": "function",
+                    "function": {
+                        "name": "read",
+                        "arguments": '{"filePath": "/tmp"}',
+                    },
+                }
+            ],
+        },
+        {
+            "role": "tool",
+            "tool_call_id": tool_call_id,
+            "content": "ok",
+        },
+    ]
+
+    contents = _gemini_convert_messages_with_history(
+        messages=messages,
+        model="gemini-3.5-flash",
+        custom_llm_provider="gemini",
+    )
+
+    function_call_ids = []
+    function_response_ids = []
+    for content in contents:
+        for part in content.get("parts", []):
+            fc = part.get("function_call")
+            if fc is not None:
+                function_call_ids.append(fc.get("id"))
+            fr = part.get("function_response")
+            if fr is not None:
+                function_response_ids.append(fr.get("id"))
+
+    assert function_call_ids == [tool_call_id]
+    assert function_response_ids == [tool_call_id]
+
+
 def test_reasoning_effort_maps_to_thinking_level_gemini_3():
     """Test that reasoning_effort maps to thinking_level AND includeThoughts for Gemini 3+ models"""
     from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
@@ -2959,6 +3078,38 @@ def test_vertex_ai_gemini3_tool_combination_no_drop():
     assert len(tools) == 3
 
 
+def test_vertex_ai_mixed_tools_and_web_search_options_drops_search():
+    """
+    When function tools and web_search_options are sent separately (Codex-style),
+    search tools are dropped unless include_server_side_tool_invocations is set.
+    """
+    v = VertexGeminiConfig()
+    optional_params: dict = {}
+    non_default_params = {
+        "tools": [
+            {
+                "type": "function",
+                "function": {"name": "exec_command", "description": "Run a command"},
+            }
+        ],
+        "web_search_options": {},
+    }
+
+    result = v.map_openai_params(
+        non_default_params=non_default_params,
+        optional_params=optional_params,
+        model="gemini-3.5-flash",
+        drop_params=True,
+    )
+
+    assert not result.get("include_server_side_tool_invocations")
+    tool_keys = set()
+    for tool in result.get("tools", []):
+        tool_keys.update(tool.keys())
+    assert "function_declarations" in tool_keys
+    assert "googleSearch" not in tool_keys
+
+
 def test_vertex_ai_openai_web_search_tool_transformation():
     """
     Test that OpenAI-style web_search and web_search_preview tools are transformed to googleSearch.
@@ -3499,7 +3650,12 @@ def test_video_metadata_supported_for_all_gemini_models():
         }
     ]
 
-    for model in ["gemini-1.5-pro", "gemini-2.5-flash", "gemini-2.5-pro", "gemini-3-pro-preview"]:
+    for model in [
+        "gemini-1.5-pro",
+        "gemini-2.5-flash",
+        "gemini-2.5-pro",
+        "gemini-3-pro-preview",
+    ]:
         contents = _gemini_convert_messages_with_history(messages=messages, model=model)
 
         file_part = None
@@ -3509,19 +3665,25 @@ def test_video_metadata_supported_for_all_gemini_models():
                 break
 
         assert file_part is not None, f"{model}: file part should exist"
-        assert "video_metadata" in file_part, f"{model}: video_metadata should be present"
+        assert (
+            "video_metadata" in file_part
+        ), f"{model}: video_metadata should be present"
         assert file_part["video_metadata"]["fps"] == 5, f"{model}: fps should be 5"
 
     # Per-part media_resolution is Gemini 3+ only; 2.x uses generation_config global
     for model in ["gemini-3-pro-preview"]:
         contents = _gemini_convert_messages_with_history(messages=messages, model=model)
         file_part = next(p for p in contents[0]["parts"] if "file_data" in p)
-        assert "media_resolution" in file_part, f"{model}: media_resolution should be present"
+        assert (
+            "media_resolution" in file_part
+        ), f"{model}: media_resolution should be present"
 
     for model in ["gemini-1.5-pro", "gemini-2.5-flash", "gemini-2.5-pro"]:
         contents = _gemini_convert_messages_with_history(messages=messages, model=model)
         file_part = next(p for p in contents[0]["parts"] if "file_data" in p)
-        assert "media_resolution" not in file_part, f"{model}: per-part media_resolution should not be set"
+        assert (
+            "media_resolution" not in file_part
+        ), f"{model}: per-part media_resolution should not be set"
 
 
 def test_chunk_parser_handles_prompt_feedback_block():
@@ -4154,8 +4316,9 @@ def test_vertex_ai_usage_metadata_with_document_tokens_in_prompt():
 
     # DOCUMENT tokens should be included in text_tokens: 8 (TEXT) + 774 (DOCUMENT) = 782
     assert result.prompt_tokens_details is not None
-    assert result.prompt_tokens_details.text_tokens == 782, \
-        "DOCUMENT modality tokens should be added to text_tokens (8 TEXT + 774 DOCUMENT = 782)"
+    assert (
+        result.prompt_tokens_details.text_tokens == 782
+    ), "DOCUMENT modality tokens should be added to text_tokens (8 TEXT + 774 DOCUMENT = 782)"
 
     # Verify completion token details
     assert result.completion_tokens_details is not None
@@ -4190,8 +4353,9 @@ def test_vertex_ai_usage_metadata_with_document_tokens_cached():
 
     # DOCUMENT cached tokens map to cached_text_tokens, so:
     # text_tokens = (8 TEXT + 774 DOCUMENT) - 400 cached = 382
-    assert result.prompt_tokens_details.text_tokens == 382, \
-        "text_tokens should be (8 + 774) - 400 cached = 382"
+    assert (
+        result.prompt_tokens_details.text_tokens == 382
+    ), "text_tokens should be (8 + 774) - 400 cached = 382"
     assert result.prompt_tokens_details.cached_tokens == 400
 
 

tests/test_litellm/llms/vertex_ai/test_vertex.py

@@ -1490,39 +1490,31 @@ def test_vertex_parallel_tool_calls_true():
     assert "tools" in optional_params
 
 
-def test_vertex_parallel_tool_calls_false_multiple_tools_error():
+def test_vertex_parallel_tool_calls_false_multiple_tools_dropped():
     """
-    Test that parallel_tool_calls = False with multiple tools raises UnsupportedParamsError
-    when drop_params is False.
+    parallel_tool_calls=False with multiple tools is dropped for Gemini
+  (unsupported upstream). Request should succeed without the param.
     """
     tools = [
         {"type": "function", "function": {"name": "get_weather"}},
         {"type": "function", "function": {"name": "get_time"}},
     ]
-    with pytest.raises(litellm.utils.UnsupportedParamsError) as excinfo:
-        get_optional_params(
-            model="gemini-1.5-pro",
-            custom_llm_provider="vertex_ai",
-            tools=tools,
-            parallel_tool_calls=False,
-        )
-    assert (
-        "`parallel_tool_calls=False` is not supported by Gemini when multiple tools are"
-        in str(excinfo.value)
+    optional_params = get_optional_params(
+        model="gemini-1.5-pro",
+        custom_llm_provider="vertex_ai",
+        tools=tools,
+        parallel_tool_calls=False,
     )
+    assert "parallel_tool_calls" not in optional_params
+    assert "tools" in optional_params
 
-    # works when specified as "functions"
-    with pytest.raises(litellm.utils.UnsupportedParamsError) as excinfo:
-        get_optional_params(
-            model="gemini-1.5-pro",
-            custom_llm_provider="vertex_ai",
-            functions=tools,
-            parallel_tool_calls=False,
-        )
-    assert (
-        "`parallel_tool_calls=False` is not supported by Gemini when multiple tools are"
-        in str(excinfo.value)
+    optional_params = get_optional_params(
+        model="gemini-1.5-pro",
+        custom_llm_provider="vertex_ai",
+        functions=tools,
+        parallel_tool_calls=False,
     )
+    assert "parallel_tool_calls" not in optional_params
 
 
 def test_vertex_parallel_tool_calls_false_single_tool():

tests/test_litellm/proxy/_experimental/mcp_server/auth/test_user_api_key_auth_mcp.py

@@ -213,7 +213,7 @@ class TestMCPRequestHandler:
             # Test case 2: Authorization header present (fallback)
             (
                 [(b"authorization", b"Bearer test-auth-token")],
-                "Bearer test-auth-token",
+                "test-auth-token",
                 None,
                 {},
             ),
@@ -674,7 +674,9 @@ class TestMCPOAuth2AuthFlow:
             ) = await MCPRequestHandler.process_mcp_request(scope)
 
             # Should succeed with the LiteLLM key from Authorization header
-            assert auth_result.api_key == "Bearer sk-litellm-valid-key"
+            from litellm.proxy.utils import hash_token
+
+            assert auth_result.api_key == hash_token("sk-litellm-valid-key")
             mock_auth.assert_called_once()
 
     async def test_non_auth_http_exception_still_raises(self):

tests/test_litellm/proxy/common_utils/test_reset_budget_job.py

@@ -92,6 +92,37 @@ class MockLiteLLMEndUserTable:
         return self._find_many_results
 
 
+class MockBatcher:
+    """Captures per-row update calls and exposes them after commit().
+
+    Mirrors prisma's `db.batch_()` ergonomics enough that the reset job's
+    narrow-write helpers (`_write_key_reset_updates` et al) can run against
+    the mock and the test can assert on what would have been written.
+    """
+
+    def __init__(self):
+        self.calls: List[Dict[str, Any]] = []
+        self.committed: bool = False
+
+        class _Table:
+            def __init__(_self, table_name: str, outer: "MockBatcher"):
+                _self._table_name = table_name
+                _self._outer = outer
+
+            def update(_self, where, data):
+                _self._outer.calls.append(
+                    {"table": _self._table_name, "where": where, "data": data}
+                )
+
+        self.litellm_verificationtoken = _Table("key", self)
+        self.litellm_usertable = _Table("user", self)
+        self.litellm_teamtable = _Table("team", self)
+
+    async def commit(self):
+        self.committed = True
+        return self.calls
+
+
 class MockDB:
     def __init__(self):
         self.litellm_teammembership = MockLiteLLMTeamMembership()
@@ -99,6 +130,19 @@ class MockDB:
         self.litellm_endusertable = MockLiteLLMEndUserTable()
         self.litellm_organizationtable = MockLiteLLMOrganizationTable()
         self.litellm_tagtable = MockLiteLLMTagTable()
+        self.batch_calls: List[Dict[str, Any]] = []
+
+    def batch_(self):
+        batcher = MockBatcher()
+        # Aggregate calls across all batches so tests can assert on cumulative writes.
+        original_commit = batcher.commit
+
+        async def _record_and_commit():
+            self.batch_calls.extend(batcher.calls)
+            return await original_commit()
+
+        batcher.commit = _record_and_commit  # type: ignore[assignment]
+        return batcher
 
 
 class MockPrismaClient:
@@ -205,6 +249,7 @@ def test_reset_budget_for_key(reset_budget_job, mock_prisma_client):
             "budget_duration": "30d",
             "budget_reset_at": now,
             "id": "test-key-1",
+            "token": "tok-key-1",
         },
     )
 
@@ -213,11 +258,16 @@ def test_reset_budget_for_key(reset_budget_job, mock_prisma_client):
     # Run the test
     asyncio.run(reset_budget_job.reset_budget_for_litellm_keys())
 
-    # Verify results
-    assert len(mock_prisma_client.updated_data["key"]) == 1
-    updated_key = mock_prisma_client.updated_data["key"][0]
-    assert updated_key.spend == 0.0
-    assert updated_key.budget_reset_at > now
+    # The reset writes only {spend, budget_reset_at} per row via batch_().
+    # Full-row writes would re-detonate the Prisma DataError on rows carrying
+    # object_permission_id / budget_limits (see #27730).
+    key_writes = [c for c in mock_prisma_client.db.batch_calls if c["table"] == "key"]
+    assert len(key_writes) == 1
+    write = key_writes[0]
+    assert write["where"] == {"token": "tok-key-1"}
+    assert write["data"]["spend"] == 0
+    assert write["data"]["budget_reset_at"] > now
+    assert set(write["data"].keys()) == {"spend", "budget_reset_at"}
 
 
 def test_reset_budget_for_user(reset_budget_job, mock_prisma_client):
@@ -231,6 +281,7 @@ def test_reset_budget_for_user(reset_budget_job, mock_prisma_client):
             "budget_duration": "7d",
             "budget_reset_at": now,
             "id": "test-user-1",
+            "user_id": "uid-1",
         },
     )
 
@@ -239,11 +290,13 @@ def test_reset_budget_for_user(reset_budget_job, mock_prisma_client):
     # Run the test
     asyncio.run(reset_budget_job.reset_budget_for_litellm_users())
 
-    # Verify results
-    assert len(mock_prisma_client.updated_data["user"]) == 1
-    updated_user = mock_prisma_client.updated_data["user"][0]
-    assert updated_user.spend == 0.0
-    assert updated_user.budget_reset_at > now
+    user_writes = [c for c in mock_prisma_client.db.batch_calls if c["table"] == "user"]
+    assert len(user_writes) == 1
+    write = user_writes[0]
+    assert write["where"] == {"user_id": "uid-1"}
+    assert write["data"]["spend"] == 0
+    assert write["data"]["budget_reset_at"] > now
+    assert set(write["data"].keys()) == {"spend", "budget_reset_at"}
 
 
 def test_reset_budget_for_team(reset_budget_job, mock_prisma_client):
@@ -257,6 +310,7 @@ def test_reset_budget_for_team(reset_budget_job, mock_prisma_client):
             "budget_duration": "1mo",
             "budget_reset_at": now,
             "id": "test-team-1",
+            "team_id": "tid-1",
         },
     )
 
@@ -265,11 +319,13 @@ def test_reset_budget_for_team(reset_budget_job, mock_prisma_client):
     # Run the test
     asyncio.run(reset_budget_job.reset_budget_for_litellm_teams())
 
-    # Verify results
-    assert len(mock_prisma_client.updated_data["team"]) == 1
-    updated_team = mock_prisma_client.updated_data["team"][0]
-    assert updated_team.spend == 0.0
-    assert updated_team.budget_reset_at > now
+    team_writes = [c for c in mock_prisma_client.db.batch_calls if c["table"] == "team"]
+    assert len(team_writes) == 1
+    write = team_writes[0]
+    assert write["where"] == {"team_id": "tid-1"}
+    assert write["data"]["spend"] == 0
+    assert write["data"]["budget_reset_at"] > now
+    assert set(write["data"].keys()) == {"spend", "budget_reset_at"}
 
 
 def test_reset_budget_for_enduser(reset_budget_job, mock_prisma_client):
@@ -324,6 +380,7 @@ def test_reset_budget_all(reset_budget_job, mock_prisma_client):
             "budget_duration": "30d",
             "budget_reset_at": now,
             "id": "test-key-1",
+            "token": "tok-all-1",
         },
     )
 
@@ -335,6 +392,7 @@ def test_reset_budget_all(reset_budget_job, mock_prisma_client):
             "budget_duration": "7d",
             "budget_reset_at": now,
             "id": "test-user-1",
+            "user_id": "uid-all-1",
         },
     )
 
@@ -346,6 +404,7 @@ def test_reset_budget_all(reset_budget_job, mock_prisma_client):
             "budget_duration": "1mo",
             "budget_reset_at": now,
             "id": "test-team-1",
+            "team_id": "tid-all-1",
         },
     )
 
@@ -379,17 +438,22 @@ def test_reset_budget_all(reset_budget_job, mock_prisma_client):
     # Run the test
     asyncio.run(reset_budget_job.reset_budget())
 
-    # Verify results
-    assert len(mock_prisma_client.updated_data["key"]) == 1
-    assert len(mock_prisma_client.updated_data["user"]) == 1
-    assert len(mock_prisma_client.updated_data["team"]) == 1
+    # key/user/team rows are written via batch_().<table>.update — verify each
+    # one fired exactly once with the narrow {spend, budget_reset_at} payload.
+    for table_name, where in [
+        ("key", {"token": "tok-all-1"}),
+        ("user", {"user_id": "uid-all-1"}),
+        ("team", {"team_id": "tid-all-1"}),
+    ]:
+        writes = [c for c in mock_prisma_client.db.batch_calls if c["table"] == table_name]
+        assert len(writes) == 1, f"expected 1 {table_name} write, got {len(writes)}"
+        assert writes[0]["where"] == where
+        assert writes[0]["data"]["spend"] == 0
+        assert set(writes[0]["data"].keys()) == {"spend", "budget_reset_at"}
+
+    # Enduser + budget rows still go through update_data (not narrowed; different path).
     assert len(mock_prisma_client.updated_data["enduser"]) == 1
     assert len(mock_prisma_client.updated_data["budget"]) == 1
-
-    # Check that all spends were reset to 0
-    assert mock_prisma_client.updated_data["key"][0].spend == 0.0
-    assert mock_prisma_client.updated_data["user"][0].spend == 0.0
-    assert mock_prisma_client.updated_data["team"][0].spend == 0.0
     assert mock_prisma_client.updated_data["enduser"][0].spend == 0.0
 
 
@@ -1399,6 +1463,105 @@ def test_reset_budget_for_teams_invalidates_redis_counter(
     )
 
 
+def test_reset_does_not_zero_counter_when_db_write_fails(monkeypatch):
+    """
+    Regression for #27730 (the bypass-half).
+
+    If the DB write inside the reset job raises (e.g. Prisma DataError on a
+    row carrying object_permission_id or budget_limits), the Redis spend
+    counter MUST NOT be zeroed — that would let get_current_spend admit
+    requests past the cap while the DB row still holds the over-budget
+    spend.
+
+    Pre-fix: _reset_budget_common pre-zeroed the counter before the DB
+    write attempt, opening the bypass window.
+    Post-fix: counter invalidation lives in the caller, AFTER the DB write
+    commits. If the write raises, the post-write invalidation never runs.
+    """
+    counter_cache = _make_counter_invalidation_job(monkeypatch)
+
+    now = datetime.now(timezone.utc)
+    prisma_client = MagicMock()
+
+    matching_key = type(
+        "Key",
+        (),
+        {
+            "spend": 100.0,
+            "budget_duration": "30d",
+            "budget_reset_at": now - timedelta(seconds=1),
+            "token": "sk-failing",
+        },
+    )
+
+    # get_data returns one key needing reset; the batched DB write then explodes.
+    async def fake_get_data(table_name, query_type, **kwargs):
+        if table_name == "key":
+            return [matching_key]
+        return []
+
+    prisma_client.get_data = fake_get_data
+
+    batcher = MagicMock()
+    batcher.litellm_verificationtoken.update = MagicMock()
+
+    async def failing_commit():
+        raise RuntimeError("simulated Prisma DataError on update")
+
+    batcher.commit = failing_commit
+    prisma_client.db.batch_ = MagicMock(return_value=batcher)
+
+    job = ResetBudgetJob(
+        proxy_logging_obj=MockProxyLogging(), prisma_client=prisma_client
+    )
+
+    asyncio.run(job.reset_budget_for_litellm_keys())
+
+    # CRITICAL: counter invalidation must NOT have been called at all —
+    # the DB write raised before the post-write invalidation loop. Using
+    # assert_not_called() instead of iterating call_args_list, because the
+    # latter is vacuously true when the list is empty (would pass even if
+    # the bypass were re-introduced via a different code path).
+    counter_cache.in_memory_cache.set_cache.assert_not_called()
+
+
+def test_reset_budget_for_keys_writes_only_spend_and_reset_at(reset_budget_job, mock_prisma_client):
+    """
+    Regression for #27730 (the trigger-half).
+
+    The reset job must write only {spend, budget_reset_at} per row — never
+    the full key object. Sending the full object via the old update_data
+    batcher path made Prisma reject any row carrying object_permission_id
+    or budget_limits (both became non-NULL on UI-created keys after v1.84.0).
+    """
+    now = datetime.now(timezone.utc)
+    key_with_problematic_fields = type(
+        "LiteLLM_VerificationToken",
+        (),
+        {
+            "spend": 50.0,
+            "budget_duration": "30d",
+            "budget_reset_at": now,
+            "token": "sk-problematic",
+            "object_permission_id": "perm-abc",   # would be rejected on update
+            "budget_limits": [{"max_budget": 5}],   # would be rejected on update
+            "metadata": {"some": "thing"},
+        },
+    )
+    mock_prisma_client.data["key"] = [key_with_problematic_fields]
+
+    asyncio.run(reset_budget_job.reset_budget_for_litellm_keys())
+
+    key_writes = [c for c in mock_prisma_client.db.batch_calls if c["table"] == "key"]
+    assert len(key_writes) == 1
+    payload_keys = set(key_writes[0]["data"].keys())
+    assert payload_keys == {"spend", "budget_reset_at"}, (
+        f"reset payload must not include any field besides spend / budget_reset_at, "
+        f"got: {payload_keys}. Any extra field (object_permission_id, budget_limits, etc.) "
+        f"trips Prisma DataError and detonates the whole batch."
+    )
+
+
 def test_reset_budget_for_keys_linked_to_budgets_invalidates_redis_counter(monkeypatch):
     """Resetting keys via budget tier must clear each linked key's counter."""
     counter_cache = _make_counter_invalidation_job(monkeypatch)

tests/test_litellm/proxy/guardrails/test_deferred_guardrail_logging.py

@@ -18,7 +18,7 @@ import asyncio
 import os
 import sys
 from typing import Any
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -38,6 +38,24 @@ from litellm.types.guardrails import GuardrailEventHooks
 # ---------------------------------------------------------------------------
 
 
+def _attach_mock_success_dispatch(mock_logging_obj, async_success_fn):
+    """Match production entrypoint: ``_run_deferred_stream_guardrails`` uses dispatch."""
+
+    async def dispatch_success_handlers(
+        result=None, start_time=None, end_time=None, cache_hit=None, **kwargs
+    ):
+        await async_success_fn(
+            result,
+            start_time=start_time,
+            end_time=end_time,
+            cache_hit=cache_hit,
+            **kwargs,
+        )
+
+    mock_logging_obj.dispatch_success_handlers = dispatch_success_handlers
+    mock_logging_obj.async_success_handler = async_success_fn
+
+
 class PostCallGuardrail(CustomGuardrail):
     """A post-call guardrail."""
 
@@ -454,7 +472,7 @@ class TestDeferredStreamingClosure:
         async def track_async_success(*args, **kwargs):
             pass
 
-        mock_logging_obj.async_success_handler = track_async_success
+        _attach_mock_success_dispatch(mock_logging_obj, track_async_success)
 
         tracking_guardrail = TrackingGuardrail()
         tracking_logger = TrackingLogger()
@@ -511,7 +529,7 @@ class TestDeferredStreamingClosure:
             nonlocal logged_response
             logged_response = args[0] if args else None
 
-        mock_logging_obj.async_success_handler = track_async_success
+        _attach_mock_success_dispatch(mock_logging_obj, track_async_success)
 
         class ModifyingGuardrail(CustomGuardrail):
             def __init__(self):
@@ -573,7 +591,7 @@ class TestDeferredStreamingClosure:
             nonlocal logging_called
             logging_called = True
 
-        mock_logging_obj.async_success_handler = track_async_success
+        _attach_mock_success_dispatch(mock_logging_obj, track_async_success)
 
         guardrail = BlockingGuardrail()
 
@@ -621,7 +639,7 @@ class TestDeferredStreamingClosure:
         async def track_async_success(*args, **kwargs):
             pass
 
-        mock_logging_obj.async_success_handler = track_async_success
+        _attach_mock_success_dispatch(mock_logging_obj, track_async_success)
 
         guardrail = TransientErrorGuardrail()
 
@@ -656,7 +674,7 @@ class TestDeferredStreamingClosure:
             nonlocal logged_response
             logged_response = args[0] if args else None
 
-        mock_logging_obj.async_success_handler = track_async_success
+        _attach_mock_success_dispatch(mock_logging_obj, track_async_success)
 
         class TestGuardrail(CustomGuardrail):
             def __init__(self):
@@ -739,7 +757,7 @@ class TestDeferredStreamingClosure:
         async def track_async_success(*args, **kwargs):
             pass
 
-        mock_logging_obj.async_success_handler = track_async_success
+        _attach_mock_success_dispatch(mock_logging_obj, track_async_success)
 
         guardrail = ApplyGuardrailType()
 
@@ -792,7 +810,7 @@ class TestDeferredStreamingClosure:
         async def track_async_success(*args, **kwargs):
             pass
 
-        mock_logging_obj.async_success_handler = track_async_success
+        _attach_mock_success_dispatch(mock_logging_obj, track_async_success)
 
         guardrail = IteratorHookGuardrail()
 
@@ -847,7 +865,7 @@ class TestDeferredStreamingClosure:
         async def track_async_success(*args, **kwargs):
             pass
 
-        mock_logging_obj.async_success_handler = track_async_success
+        _attach_mock_success_dispatch(mock_logging_obj, track_async_success)
 
         guardrail = InspectingGuardrail()
 
@@ -914,7 +932,7 @@ class TestDeferredStreamingClosure:
         async def track_async_success(*args, **kwargs):
             pass
 
-        mock_logging_obj.async_success_handler = track_async_success
+        _attach_mock_success_dispatch(mock_logging_obj, track_async_success)
 
         guardrail_a = TaggedGuardrail("guardrail-a")
         guardrail_b = TaggedGuardrail("guardrail-b")
@@ -962,7 +980,7 @@ class TestDeferredStreamingClosure:
             nonlocal logging_called
             logging_called = True
 
-        mock_logging_obj.async_success_handler = track_async_success
+        _attach_mock_success_dispatch(mock_logging_obj, track_async_success)
 
         def exploding_merge(data, llm_router):
             raise RuntimeError("Simulated init failure")
@@ -986,6 +1004,67 @@ class TestDeferredStreamingClosure:
             logging_called is True
         ), "Logging must fire even when guardrail initialization raises"
 
+    @pytest.mark.asyncio
+    async def test_deferred_logging_forces_async_for_sync_classified_call_type(self):
+        """
+        Regression: proxy deferred streaming logging must reach the async success
+        handler (which runs the async-only DB/spend logger) even when the call
+        type is classified as a sync SDK request by _is_sync_litellm_request.
+
+        Without prefer_async_handlers=True, an async proxy stream whose
+        litellm_params lacks a recognized async marker would enter the sync
+        branch of dispatch_success_handlers and silently skip spend tracking.
+
+        Uses the real dispatch_success_handlers via the production
+        _run_deferred_stream_guardrails entrypoint.
+        """
+        import time
+
+        from litellm.litellm_core_utils.litellm_logging import (
+            Logging as LiteLLMLoggingObj,
+        )
+
+        logging_obj = LiteLLMLoggingObj(
+            model="gpt-4o-mini",
+            messages=[{"role": "user", "content": "hi"}],
+            stream=True,
+            call_type="completion",  # not pass_through_endpoint
+            start_time=time.time(),
+            litellm_call_id="test-id",
+            function_id="fn",
+        )
+        # litellm_params with no recognized async marker -> classified sync.
+        logging_obj.model_call_details["litellm_params"] = {}
+        assert LiteLLMLoggingObj._is_sync_litellm_request({}) is True
+
+        with (
+            patch.object(
+                logging_obj, "async_success_handler", new_callable=AsyncMock
+            ) as mock_async,
+            patch.object(
+                logging_obj, "success_handler", new_callable=MagicMock
+            ) as mock_sync,
+            patch.object(
+                logging_obj,
+                "_should_run_sync_callbacks_for_async_calls",
+                return_value=False,
+            ),
+            patch("litellm.callbacks", [PostCallGuardrail()]),
+        ):
+            await ProxyBaseLLMRequestProcessing._run_deferred_stream_guardrails(
+                captured_data={"model": "gpt-4o-mini", "metadata": {}},
+                captured_user_api_key_dict=UserAPIKeyAuth(api_key="test"),
+                captured_logging_obj=logging_obj,
+                assembled_response=MagicMock(),
+                cache_hit=False,
+            )
+
+            await asyncio.sleep(0)
+            await asyncio.sleep(0)
+
+        mock_async.assert_awaited_once()
+        mock_sync.assert_not_called()
+
 
 # ---------------------------------------------------------------------------
 # 7. _fire_deferred_stream_logging
@@ -1054,7 +1133,7 @@ class TestFireDeferredStreamLogging:
             nonlocal logged_response
             logged_response = args[0] if args else None
 
-        mock_logging_obj.async_success_handler = track_async_success
+        _attach_mock_success_dispatch(mock_logging_obj, track_async_success)
 
         class InfoWritingGuardrail(CustomGuardrail):
             def __init__(self):

tests/test_litellm/proxy/test_proxy_server.py

@@ -5648,6 +5648,7 @@ async def test_init_and_increment_spend_counter_reseeds_from_db_on_counter_miss(
     fake_redis = AsyncMock()
     fake_redis.async_increment = AsyncMock(side_effect=record_increment)
     fake_redis.async_get_cache = AsyncMock(return_value=None)  # counter missing
+    fake_redis.async_set_cache = AsyncMock(return_value=True)  # SET NX wins
     counter_cache.redis_cache = fake_redis
 
     # Prisma returns spend=42.0 (authoritative) while the stale cached
@@ -5684,16 +5685,131 @@ async def test_init_and_increment_spend_counter_reseeds_from_db_on_counter_miss(
         fake_prisma.db.litellm_teamtable.find_unique.assert_awaited_once_with(
             where={"team_id": "team-9"}
         )
-        # Two increments keyed on the counter: seed ($42) then request ($1.50).
+        # Seed uses SET NX with db_spend (42) — cross-pod safe, no INCR of 42.
+        # Only the per-request delta (1.5) goes through INCRBYFLOAT.
+        fake_redis.async_set_cache.assert_awaited_once_with(
+            key="spend:team:team-9", value=42.0, nx=True
+        )
         writes = [(c["key"], c["value"]) for c in recorded_increments]
-        assert ("spend:team:team-9", 42.0) in writes
-        assert ("spend:team:team-9", 1.5) in writes
+        assert writes == [("spend:team:team-9", 1.5)]
     finally:
         ps.user_api_key_cache = orig_user
         ps.spend_counter_cache = orig_counter
         ps.prisma_client = orig_prisma
 
 
+@pytest.mark.asyncio
+async def test_primary_spend_counter_redis_concurrent_seed_does_not_double_seed():
+    """Two pods both observing a missing Redis counter must not both
+    INCRBYFLOAT the full DB spend. SpendCounterReseed.coalesced uses SET NX
+    so the loser reads the winner's value; final Redis = db_spend, not
+    2 * db_spend.
+
+    The per-counter asyncio.Lock is per-process, so it does NOT coordinate
+    across pods. We simulate two pods by patching _get_lock to return a
+    fresh lock per call (each "pod" has its own lock registry in real life).
+    """
+    from litellm.caching.dual_cache import DualCache
+    from litellm.proxy.db.spend_counter_reseed import SpendCounterReseed
+
+    counter_key = "spend:team:team-concurrent-seed"
+    redis_store: dict = {}
+    db_read_count = 0
+    set_results: list = []
+    get_after_set_count = 0
+    set_completed_count = 0
+
+    async def redis_set_cache(key, value, nx=False, **_):
+        # Yield BEFORE the membership check so two concurrent callers
+        # interleave the way real atomic Redis SET NX does: the first
+        # to resume runs check + write atomically and wins; the second
+        # resumes after the key exists and loses. Yielding *after* the
+        # check would let both callers pass the empty-store check before
+        # either writes, so neither would ever lose.
+        await asyncio.sleep(0)
+        if nx and key in redis_store:
+            set_results.append(False)
+            return False
+        redis_store[key] = float(value)
+        set_results.append(True)
+        nonlocal set_completed_count
+        set_completed_count += 1
+        return True
+
+    async def redis_get_cache(key):
+        # Track reads that happen after at least one SET NX has completed
+        # — those are the loser-path fallback reads we want to verify.
+        if set_completed_count > 0:
+            nonlocal get_after_set_count
+            get_after_set_count += 1
+        return redis_store.get(key)
+
+    fake_redis = AsyncMock()
+    fake_redis.async_get_cache = AsyncMock(side_effect=redis_get_cache)
+    fake_redis.async_set_cache = AsyncMock(side_effect=redis_set_cache)
+
+    async def slow_find_unique(**_):
+        nonlocal db_read_count
+        db_read_count += 1
+        # Both pods read DB before either's SET NX lands.
+        await asyncio.sleep(0)
+        row = MagicMock()
+        row.spend = 506.0
+        return row
+
+    fake_prisma = MagicMock()
+    fake_prisma.db.litellm_teamtable.find_unique = AsyncMock(
+        side_effect=slow_find_unique
+    )
+
+    pod_a = DualCache()
+    pod_a.redis_cache = fake_redis
+    pod_b = DualCache()
+    pod_b.redis_cache = fake_redis
+
+    # Each "pod" has its own per-process lock registry. Patch _get_lock to
+    # always return a fresh lock so the two coalesced calls do not serialize
+    # via one in-process lock (which is what would happen across pods).
+    async def fresh_lock(_counter_key):
+        return asyncio.Lock()
+
+    with patch.object(SpendCounterReseed, "_get_lock", side_effect=fresh_lock):
+        results = await asyncio.gather(
+            SpendCounterReseed.coalesced(
+                prisma_client=fake_prisma,
+                spend_counter_cache=pod_a,
+                counter_key=counter_key,
+            ),
+            SpendCounterReseed.coalesced(
+                prisma_client=fake_prisma,
+                spend_counter_cache=pod_b,
+                counter_key=counter_key,
+            ),
+        )
+
+    assert all(r == 506.0 for r in results), results
+    assert redis_store[counter_key] == pytest.approx(506.0), redis_store
+    # Both pods read the DB and both attempted SET NX; exactly one wrote
+    # (winner) and one was rejected (loser).
+    assert db_read_count == 2
+    assert fake_redis.async_set_cache.await_count == 2
+    nx_writes = [
+        call
+        for call in fake_redis.async_set_cache.await_args_list
+        if call.kwargs.get("nx") is True
+    ]
+    assert len(nx_writes) == 2
+    assert sorted(set_results) == [False, True], (
+        f"expected exactly one SET NX winner and one loser, got {set_results}"
+    )
+    # Loser path executed: after the winner's SET NX returned True, the
+    # losing coalesced() call falls back to async_get_cache to read the
+    # winner's value rather than re-seeding.
+    assert get_after_set_count >= 1, (
+        "loser branch (else: read back winner's value) was never exercised"
+    )
+
+
 @pytest.mark.asyncio
 async def test_reseed_spend_from_db_user_and_org_prefixes():
     """User and org counters reseed from their own DB tables.
@@ -5817,9 +5933,16 @@ async def test_init_spend_counter_redis_clean_miss_skips_stale_in_memory():
         redis_store[key] = (redis_store.get(key) or 0.0) + value
         return redis_store[key]
 
+    async def redis_set_cache(key, value, nx=False, **_):
+        if nx and key in redis_store:
+            return False
+        redis_store[key] = float(value)
+        return True
+
     fake_redis = AsyncMock()
     fake_redis.async_get_cache = AsyncMock(return_value=None)
     fake_redis.async_increment = AsyncMock(side_effect=redis_increment)
+    fake_redis.async_set_cache = AsyncMock(side_effect=redis_set_cache)
     counter_cache.redis_cache = fake_redis
 
     db_row = MagicMock()
@@ -5847,6 +5970,7 @@ async def test_init_spend_counter_redis_clean_miss_skips_stale_in_memory():
         fake_prisma.db.litellm_teamtable.find_unique.assert_awaited_once_with(
             where={"team_id": "team-stale-local"}
         )
+        # Seed via SET NX (42) + delta via INCRBYFLOAT (1.5) = 43.5.
         assert redis_store[counter_key] == pytest.approx(43.5)
         assert counter_cache.in_memory_cache.get_cache(
             key=counter_key
@@ -6237,14 +6361,14 @@ async def test_get_current_spend_reseeds_from_db_when_counter_missing():
     from litellm.proxy.proxy_server import get_current_spend
 
     counter_cache = DualCache()
-    recorded_warms: list = []
+    recorded_seeds: list = []
 
-    async def record_increment(key, value, ttl=None, **kwargs):
-        recorded_warms.append({"key": key, "value": value})
-        return value
+    async def record_set_cache(key, value, nx=False, **kwargs):
+        recorded_seeds.append({"key": key, "value": value, "nx": nx})
+        return True
 
     fake_redis = AsyncMock()
-    fake_redis.async_increment = AsyncMock(side_effect=record_increment)
+    fake_redis.async_set_cache = AsyncMock(side_effect=record_set_cache)
     fake_redis.async_get_cache = AsyncMock(return_value=None)
     counter_cache.redis_cache = fake_redis
 
@@ -6269,9 +6393,9 @@ async def test_get_current_spend_reseeds_from_db_when_counter_missing():
             f"expected DB reseed to return 362.0, got {spend} "
             f"(fallback would have returned 30.0 and caused bypass)"
         )
-        # Counter warmed so subsequent reads are fast
-        assert ("spend:team_member:user-1:team-1", 362.0) in [
-            (w["key"], w["value"]) for w in recorded_warms
+        # Counter warmed via SET NX so subsequent reads are fast.
+        assert ("spend:team_member:user-1:team-1", 362.0, True) in [
+            (s["key"], s["value"], s["nx"]) for s in recorded_seeds
         ]
         assert counter_cache.in_memory_cache.get_cache(
             key="spend:team_member:user-1:team-1"
@@ -6348,8 +6472,15 @@ async def test_get_current_spend_coalesces_concurrent_reseeds():
         redis_store[key] = (redis_store.get(key) or 0.0) + value
         return redis_store[key]
 
+    async def redis_set_cache(key, value, nx=False, **_):
+        if nx and key in redis_store:
+            return False
+        redis_store[key] = float(value)
+        return True
+
     fake_redis.async_get_cache = AsyncMock(side_effect=redis_get)
     fake_redis.async_increment = AsyncMock(side_effect=redis_increment)
+    fake_redis.async_set_cache = AsyncMock(side_effect=redis_set_cache)
     counter_cache.redis_cache = fake_redis
 
     fake_prisma = MagicMock()
@@ -6456,9 +6587,16 @@ async def test_concurrent_read_and_write_paths_share_one_db_query():
         redis_store[key] = (redis_store.get(key) or 0.0) + value
         return redis_store[key]
 
+    async def redis_set_cache(key, value, nx=False, **_):
+        if nx and key in redis_store:
+            return False
+        redis_store[key] = float(value)
+        return True
+
     fake_redis = AsyncMock()
     fake_redis.async_get_cache = AsyncMock(side_effect=redis_get)
     fake_redis.async_increment = AsyncMock(side_effect=redis_increment)
+    fake_redis.async_set_cache = AsyncMock(side_effect=redis_set_cache)
     counter_cache.redis_cache = fake_redis
 
     fake_prisma = MagicMock()
@@ -6561,9 +6699,16 @@ async def test_reseed_warms_cache_even_on_zero_db_spend():
         redis_store[key] = (redis_store.get(key) or 0.0) + value
         return redis_store[key]
 
+    async def redis_set_cache(key, value, nx=False, **_):
+        if nx and key in redis_store:
+            return False
+        redis_store[key] = float(value)
+        return True
+
     fake_redis = AsyncMock()
     fake_redis.async_get_cache = AsyncMock(side_effect=redis_get)
     fake_redis.async_increment = AsyncMock(side_effect=redis_increment)
+    fake_redis.async_set_cache = AsyncMock(side_effect=redis_set_cache)
     counter_cache.redis_cache = fake_redis
 
     db_call_count = 0

tests/test_litellm/proxy/test_proxy_types.py

@@ -69,3 +69,21 @@ def test_internal_jobs_user_has_proxy_admin_role():
     assert system_user.user_id == "system"
     assert system_user.team_id == "system"
     assert system_user.team_alias == "system"
+
+
+def test_user_api_key_auth_hashes_authorization_header_form_of_key():
+    from litellm.proxy._types import UserAPIKeyAuth
+
+    raw_key = "sk-AbCdEfGhIjKlMnOpQrStUvWxYz0123456789"
+    baseline = UserAPIKeyAuth(api_key=raw_key)
+
+    for header_form in (
+        f"Bearer {raw_key}",
+        f"bearer {raw_key}",
+        f"BEARER {raw_key}",
+        f"BeArEr {raw_key}",
+    ):
+        from_header = UserAPIKeyAuth(api_key=header_form)
+        assert from_header.api_key == baseline.api_key
+        assert from_header.token == baseline.token
+        assert not from_header.api_key.lower().startswith("bearer")

uv.lock

@@ -9,7 +9,7 @@ resolution-markers = [
 ]
 
 [options]
-exclude-newer = "2026-05-24T07:24:03.668568Z"
+exclude-newer = "2026-05-30T23:40:12.713734Z"
 exclude-newer-span = "P3D"
 
 [manifest]
@@ -3189,7 +3189,7 @@ wheels = [
 
 [[package]]
 name = "litellm"
-version = "1.86.2"
+version = "1.86.3"
 source = { editable = "." }
 dependencies = [
     { name = "aiohttp" },