chore(release): backport #31519, #31733 to stable/1.90.x and cut 1.90.2

litellm/litellm_core_utils/realtime_streaming.py

@@ -5,6 +5,7 @@
 
 import litellm
 from litellm._logging import verbose_logger
+from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
 from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig
 from litellm.types.llms.openai import (
     OpenAIRealtimeEvents,
@@ -327,8 +328,10 @@ async def log_messages(self):
                     self.tool_calls
                 )
             ## ASYNC LOGGING
-            # Create an event loop for the new thread
-            asyncio.create_task(self.logging_obj.async_success_handler(self.messages))
+            # Route through the bounded logging worker (per-coroutine timeout +
+            # concurrency cap) instead of a bare create_task, so a slow callback
+            # can't leave suspended tasks pinning each call's response in memory.
+            GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue(self.logging_obj.async_success_handler(self.messages))
             ## SYNC LOGGING
             executor.submit(self.logging_obj.success_handler(self.messages))
 
@@ -357,6 +360,7 @@ async def _send_to_backend(self, message: str) -> bool:
                 # send, causing subsequent client session.update messages to
                 # be treated as "subsequent" and dropped even though the
                 # backend never received the original setup.
+                msg = self._maybe_inject_guardrail_auto_response_disable(msg)
                 await self.backend_ws.send(msg)  # type: ignore[union-attr, attr-defined]
                 self._cache_session_configuration_request(msg)
                 sent = True
@@ -617,6 +621,39 @@ async def _maybe_send_guardrail_turn_detection_update(self) -> None:
         if sent:
             self._guardrail_turn_detection_update_sent = True
 
+    def _maybe_inject_guardrail_auto_response_disable(self, setup_message: str) -> str:
+        """Fold the transcription-guardrail auto-response disable into the setup.
+
+        Gemini/Vertex Live reject a second ``setup`` (1007), so the guardrail's
+        ``automaticActivityDetection.disabled=true`` cannot be delivered as a
+        follow-up session.update; it must live in the one-and-only setup, or a
+        ``realtime_input_transcription`` guardrail is bypassed (the model
+        auto-responds before the proxy can gate the turn). Applies only to the
+        bidi ``setup`` shape; OpenAI sessions accept follow-up updates and so are
+        left untouched (handled by ``_maybe_send_guardrail_turn_detection_update``).
+        """
+        if self._guardrail_turn_detection_update_sent:
+            return setup_message
+        if not self._has_audio_transcription_guardrails():
+            return setup_message
+        try:
+            obj = json.loads(setup_message)
+        except (json.JSONDecodeError, TypeError):
+            return setup_message
+        setup = obj.get("setup") if isinstance(obj, dict) else None
+        if not isinstance(setup, dict):
+            return setup_message
+        automatic = setup.setdefault("realtimeInputConfig", {}).setdefault(
+            "automaticActivityDetection", {}
+        )
+        automatic["disabled"] = True
+        self._guardrail_turn_detection_update_sent = True
+        verbose_logger.debug(
+            "Realtime: folded automaticActivityDetection.disabled=true into setup "
+            "for transcription-guardrail gating"
+        )
+        return json.dumps(obj)
+
     def _has_realtime_guardrails_for_event_hooks(
         self,
         event_hooks: List[Any],

litellm/llms/custom_httpx/llm_http_handler.py

@@ -5457,6 +5457,58 @@ def _append_query_params(
         new_query = parsed.query + ("&" if parsed.query else "") + urlencode(extras)
         return urlunparse(parsed._replace(query=new_query))
 
+    @staticmethod
+    async def _open_realtime_backend_ws(
+        websockets_module: Any,
+        url: str,
+        headers: dict,
+        ssl_context: Any,
+        *,
+        open_timeout: float = 8.0,
+        max_attempts: int = 3,
+    ) -> Any:
+        """Open the backend realtime websocket, retrying a hung open handshake.
+
+        The upstream Live handshake (e.g. Gemini Live) intermittently hangs on
+        open; waiting longer never recovers a hung attempt, but a fresh attempt
+        almost always connects in ~1s. So bound each attempt with ``open_timeout``
+        and retry, instead of surfacing one slow handshake to the caller as a
+        fatal 1011. A bounded attempt that timed out already spaced out the
+        retry, so no extra backoff is needed. Deterministic rejections (auth /
+        handshake status) are not retried.
+        """
+        # Handshake-status rejections are deterministic (auth / 4xx): retrying
+        # cannot help and the caller must see the upstream status, not a generic
+        # 1011. websockets <15 raises InvalidStatusCode, >=15 raises InvalidStatus.
+        deterministic_errors = tuple(
+            exc
+            for exc in (
+                getattr(websockets_module.exceptions, "InvalidStatus", None),
+                getattr(websockets_module.exceptions, "InvalidStatusCode", None),
+            )
+            if exc is not None
+        )
+        last_exc: Optional[BaseException] = None
+        for _ in range(max_attempts):
+            try:
+                return await websockets_module.connect(
+                    url,
+                    additional_headers=headers,
+                    max_size=REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES,
+                    ssl=ssl_context,
+                    open_timeout=open_timeout,
+                )
+            except deterministic_errors:
+                raise
+            except (
+                TimeoutError,
+                OSError,
+                websockets_module.exceptions.WebSocketException,
+            ) as e:
+                last_exc = e
+        assert last_exc is not None  # loop only exits via return or a captured exc
+        raise last_exc
+
     async def async_realtime(
         self,
         model: str,
@@ -5491,22 +5543,10 @@ async def async_realtime(
                 ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
                 ssl_context.check_hostname = False
                 ssl_context.verify_mode = ssl.CERT_NONE
-            async with websockets.connect(  # type: ignore
-                url,
-                additional_headers=headers,
-                max_size=REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES,
-                ssl=ssl_context,
-            ) as backend_ws:
-                # Auto-send session setup if the provider requires it
-                # (e.g. Gemini/Vertex AI Live needs a `setup` message before any realtime_input)
-                _session_config: Optional[str] = None
-                if provider_config.requires_session_configuration():
-                    _session_config = provider_config.session_configuration_request(
-                        model
-                    )
-                    if _session_config:
-                        await backend_ws.send(_session_config)
-
+            backend_ws = await self._open_realtime_backend_ws(
+                websockets, url, headers, ssl_context
+            )
+            async with backend_ws:
                 _request_data: Dict[str, Any] = {}
                 if litellm_metadata:
                     _request_data["litellm_metadata"] = litellm_metadata
@@ -5524,8 +5564,26 @@ async def async_realtime(
                         else None
                     ),
                 )
-                if _session_config:
-                    realtime_streaming.session_configuration_request = _session_config
+
+                # Auto-send session setup if the provider requires it (e.g.
+                # Gemini/Vertex AI Live needs a `setup` before any realtime_input).
+                # Build the streaming handler first so a transcription guardrail's
+                # auto-response disable can be folded into this one setup: Gemini
+                # rejects a second setup, so a follow-up disable would be dropped
+                # and the guardrail bypassed.
+                _session_config: Optional[str] = None
+                if provider_config.requires_session_configuration():
+                    _session_config = provider_config.session_configuration_request(
+                        model
+                    )
+                    if _session_config:
+                        _session_config = (
+                            realtime_streaming._maybe_inject_guardrail_auto_response_disable(
+                                _session_config
+                            )
+                        )
+                        await backend_ws.send(_session_config)
+                        realtime_streaming.session_configuration_request = _session_config
 
                 # For providers that defer setup until client session.update, optionally
                 # send synthetic session.created to unblock clients waiting on connect.

litellm/llms/gemini/realtime/transformation.py

@@ -433,17 +433,11 @@ def _handle_session_update(
         Handle session.update by sending setup to Gemini.
 
         On the FIRST session.update (when session_configuration_request is None),
-        the full setup with all configuration is sent.
-
-        Subsequent session.update messages are forwarded as a follow-up setup
-        with the new fields merged into the original setup. Gemini Live treats
-        a follow-up BidiGenerateContentSetup as a full session replacement
-        rather than a partial merge, so we carry forward the previous setup
-        (tools, generationConfig, inputAudioTranscription, systemInstruction,
-        ...) and overlay the new fields on top. This preserves the old
-        behavior where clients could refine the session via session.update
-        (e.g. add tools after the auto-setup on connect), and also keeps the
-        guardrail-driven turn_detection update working.
+        the full setup with all configuration is sent. Gemini Live accepts setup
+        as the first-and-only client message, so every later session.update is
+        dropped rather than forwarded as a second setup (which Gemini rejects
+        with a 1007, tearing the session down). To carry tools/instructions, send
+        them on the first session.update before any conversation content.
         """
         session_payload = json_message.get("session") or {}
         # Normalize GA-remapped fields (``output_modalities``,
@@ -472,82 +466,32 @@ def _handle_session_update(
                 )
             ]
 
-        if not new_overrides:
-            verbose_logger.debug(
-                "Gemini Realtime: Ignoring session.update (no mappable fields)"
-            )
-            return []
-
-        try:
-            original_setup = cast(
-                BidiGenerateContentSetup,
-                json.loads(session_configuration_request).get("setup", {}),
-            )
-        except (json.JSONDecodeError, AttributeError):
-            original_setup = {}
-
-        # Deep-merge ``generationConfig`` and ``realtimeInputConfig`` so a
-        # partial session.update (e.g. only ``temperature`` or only
-        # ``modalities``) does not silently drop unrelated sub-keys
-        # (``responseModalities``, ``maxOutputTokens``, ...) from the original
-        # setup.
-        follow_up_setup: BidiGenerateContentSetup = {
-            **original_setup,
-            **new_overrides,
-            "model": f"models/{model}",
-        }
-        original_generation_config = original_setup.get("generationConfig")
-        new_generation_config = new_overrides.get("generationConfig")
-        if isinstance(original_generation_config, dict) and isinstance(
-            new_generation_config, dict
-        ):
-            follow_up_setup["generationConfig"] = {
-                **original_generation_config,
-                **new_generation_config,
-            }
-        original_realtime_input_config = original_setup.get("realtimeInputConfig")
-        new_realtime_input_config = new_overrides.get("realtimeInputConfig")
-        if isinstance(original_realtime_input_config, dict) and isinstance(
-            new_realtime_input_config, dict
+        # Gemini Live accepts exactly one ``setup`` message: the first and only
+        # client message. A second ``setup`` closes the socket with
+        # ``1007 Request contains an invalid argument``, so a session.update
+        # after the initial setup must not be forwarded as a follow-up setup.
+        # Every GA client (pipecat included) sends several session.updates while
+        # configuring the session; forwarding a second one tears the session down
+        # before the first turn, which surfaces to callers as silence after the
+        # first response, reconnect/retry latency churn, and 1011 errors. Drop
+        # it. The Vertex subclass already drops subsequent setups for this exact
+        # reason; the constraint is identical on AI Studio.
+        client_turn_detection = self._extract_turn_detection(session_payload)
+        if (
+            isinstance(client_turn_detection, dict)
+            and client_turn_detection.get("create_response") is False
         ):
-            merged_realtime_input_config = {
-                **original_realtime_input_config,
-                **new_realtime_input_config,
-            }
-            # Deep-merge ``automaticActivityDetection`` so a partial VAD
-            # update (e.g. the guardrail-injected ``disabled: True`` from
-            # ``create_response: False``) does not silently drop unrelated
-            # knobs like ``silenceDurationMs`` / ``prefixPaddingMs`` from
-            # the original setup.
-            original_automatic_activity_detection = original_realtime_input_config.get(
-                "automaticActivityDetection"
-            )
-            new_automatic_activity_detection = new_realtime_input_config.get(
-                "automaticActivityDetection"
-            )
-            if isinstance(original_automatic_activity_detection, dict) and isinstance(
-                new_automatic_activity_detection, dict
-            ):
-                merged_realtime_input_config["automaticActivityDetection"] = {
-                    **original_automatic_activity_detection,
-                    **new_automatic_activity_detection,
-                }
-            follow_up_setup["realtimeInputConfig"] = cast(
-                BidiGenerateContentRealtimeInputConfig,
-                merged_realtime_input_config,
+            verbose_logger.warning(
+                "Gemini Realtime: Dropping subsequent session.update "
+                "(turn_detection.create_response=False) — Gemini Live rejects a "
+                "second setup message, so audio-transcription guardrails cannot "
+                "suppress the model's auto-response mid-session."
             )
-        verbose_logger.debug(
-            "Gemini Realtime: Forwarding session.update as follow-up setup"
-        )
-        return [
-            json.dumps(
-                {
-                    "setup": self._finalize_gemini_live_setup(
-                        model, cast(Dict[str, Any], follow_up_setup)
-                    )
-                }
+        else:
+            verbose_logger.debug(
+                "Gemini Realtime: Ignoring session.update (setup already sent)"
             )
-        ]
+        return []
 
     def _handle_conversation_item(self, json_message: dict) -> List[str]:
         """

pyproject.toml

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