From 13285671f23ac947e42d80c65b03a6b5b239ebdf Mon Sep 17 00:00:00 2001 From: mubashir1osmani Date: Tue, 30 Jun 2026 18:37:14 -0700 Subject: [PATCH 1/4] fix(realtime): stop second Gemini Live setup, retry hung handshake, close guardrail bypass (#31519) * fix(realtime): stop sending a second Gemini Live setup on follow-up session.update Gemini Live (BidiGenerateContent) accepts setup as the first-and-only client message; a second setup closes the socket with 1007 Request contains an invalid argument. The AI Studio Gemini path forwarded every client session.update after the first as a follow-up setup, and GA clients (pipecat) send several while configuring the session, so the second one tore the session down before the first turn. Callers saw silence after the first response, exponential per-turn latency from reconnect/retry churn, and intermittent 1011 errors. Drop subsequent session.updates instead of resending setup, matching what the Vertex subclass already does. Tools and instructions must ride on the first session.update before any conversation content. Adds regression tests covering the plain follow-up, a follow-up that adds tools (the case the previous identical-only dedup still forwarded), and the guardrail create_response=False warning path. * fix(realtime): retry the backend open handshake instead of failing with 1011 The upstream Live API open handshake (e.g. Gemini Live) intermittently hangs; waiting longer never recovers a hung attempt, but a fresh attempt almost always connects in ~1s. The proxy opened the backend websocket once with the default open_timeout and no retry, so a single slow handshake surfaced to the caller as a fatal 1011 internal error and dropped the call. Bound each open attempt with a short open_timeout and retry; a bounded attempt that already timed out spaces out the next try, so no backoff is needed. Deterministic handshake-status rejections (auth/4xx) are not retried, and the retry only ever wraps the open, never a live session. Adds tests for retry-then-succeed, raise-after-max-attempts, and no-retry-on-auth-failure. * fix(realtime): close guardrail bypass + surface handshake status; drop obsolete tests Three review fixes on the Gemini Live realtime path. Transcription-guardrail bypass: Gemini Live rejects a second setup (1007), so once the initial setup is sent the guardrail's automaticActivityDetection.disabled=true can no longer be delivered as a follow-up session.update. With that follow-up now dropped, the model's auto-response stayed enabled and a realtime_input_transcription guardrail was bypassed (the model answered before the proxy could gate the turn). Fold the disable into the one-and-only setup instead: the handler injects it into the auto-sent setup (gemini_live_defer_setup false) and _send_to_backend injects it into the deferred first setup. OpenAI sessions accept follow-up updates and are left untouched. Backend handshake status: the open-retry treated only InvalidStatusCode as deterministic; websockets>=15 raises InvalidStatus for a rejected client handshake, so a 401/403 fell into the broad WebSocketException branch and was retried before the caller closed the client with 1011 instead of the upstream status. Treat both as non-retryable. Obsolete tests: the four tests asserting a follow-up session.update is merged and re-sent as a second setup asserted behavior that crashes Gemini Live with 1007 (verified directly against the API). Removed; the drop is covered by new regression tests. * style(realtime): reformat changed files to ruff line-length 120 Post-merge with litellm_internal_staging, which unified ruff format width to 120 (#31518). The realtime change set was formatted at 88, so the changed lines tripped the whole-file ruff format check. Reformat with ruff 0.15.3 at the repo's 120 width; no logic changes. (cherry picked from commit ef5d05f137d719924a8b1116ca6c0ea17ea6e2c3) --- .../litellm_core_utils/realtime_streaming.py | 34 +++ litellm/llms/custom_httpx/llm_http_handler.py | 94 ++++-- .../llms/gemini/realtime/transformation.py | 112 ++------ .../test_realtime_streaming.py | 69 +++++ .../custom_httpx/test_llm_http_handler.py | 93 ++++++ .../test_gemini_realtime_transformation.py | 267 +++++++----------- 6 files changed, 403 insertions(+), 266 deletions(-) diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index c56a70177bfa..a35d113fc9d0 100644 --- a/litellm/litellm_core_utils/realtime_streaming.py +++ b/litellm/litellm_core_utils/realtime_streaming.py @@ -357,6 +357,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 +618,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], diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py index 70fb4cd95cf5..db05da19be52 100644 --- a/litellm/llms/custom_httpx/llm_http_handler.py +++ b/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. diff --git a/litellm/llms/gemini/realtime/transformation.py b/litellm/llms/gemini/realtime/transformation.py index 74f6cd4d8319..ea50259ad248 100644 --- a/litellm/llms/gemini/realtime/transformation.py +++ b/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]: """ diff --git a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py index 2c164f2169cb..18547f2f6a30 100644 --- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py +++ b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py @@ -2681,3 +2681,72 @@ async def test_deferred_setup_clear_drops_appends_when_buffered(): streaming._buffer_pending_message_until_setup(new_audio) assert streaming._pending_messages_until_setup == [new_audio] + + +def _transcription_guardrail(): + """A minimal real CustomGuardrail registered for the realtime transcript hook.""" + from litellm.integrations.custom_guardrail import CustomGuardrail + from litellm.types.guardrails import GuardrailEventHooks + + class _TranscriptionGuardrail(CustomGuardrail): + async def apply_guardrail( + self, inputs, request_data, input_type, logging_obj=None + ): + return inputs + + return _TranscriptionGuardrail( + guardrail_name="test_transcription_guard", + event_hook=GuardrailEventHooks.realtime_input_transcription, + default_on=True, + ) + + +def test_setup_folds_in_auto_response_disable_when_transcription_guardrail_active(): + """Gemini rejects a second setup, so a transcription guardrail's auto-response + disable must be folded into the one-and-only setup; otherwise the model + auto-responds and the guardrail is bypassed.""" + import litellm + + litellm.callbacks = [_transcription_guardrail()] + try: + streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock()) + setup = json.dumps( + { + "setup": { + "model": "models/gemini-3.1-flash-live-preview", + "generationConfig": {"responseModalities": ["AUDIO"]}, + "inputAudioTranscription": {}, + } + } + ) + out = json.loads(streaming._maybe_inject_guardrail_auto_response_disable(setup)) + aad = out["setup"]["realtimeInputConfig"]["automaticActivityDetection"] + assert aad["disabled"] is True + finally: + litellm.callbacks = [] + + +def test_setup_unchanged_without_transcription_guardrail(): + import litellm + + litellm.callbacks = [] + streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock()) + setup = json.dumps( + {"setup": {"model": "x", "generationConfig": {"responseModalities": ["AUDIO"]}}} + ) + out = streaming._maybe_inject_guardrail_auto_response_disable(setup) + assert json.loads(out) == json.loads(setup) + + +def test_non_bidi_setup_left_untouched_for_followup_capable_providers(): + """OpenAI realtime accepts a follow-up session.update, so a non-bidi message + (no top-level 'setup' key) must be left untouched even with a guardrail on.""" + import litellm + + litellm.callbacks = [_transcription_guardrail()] + try: + streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock()) + msg = json.dumps({"type": "session.update", "session": {"instructions": "hi"}}) + assert streaming._maybe_inject_guardrail_auto_response_disable(msg) == msg + finally: + litellm.callbacks = [] diff --git a/tests/test_litellm/llms/custom_httpx/test_llm_http_handler.py b/tests/test_litellm/llms/custom_httpx/test_llm_http_handler.py index f7d445d0788a..0d87232fe965 100644 --- a/tests/test_litellm/llms/custom_httpx/test_llm_http_handler.py +++ b/tests/test_litellm/llms/custom_httpx/test_llm_http_handler.py @@ -1051,3 +1051,96 @@ def test_async_compact_handler_sends_json_when_not_signed(): ) assert kwargs.get("json") == {"model": "openai.gpt-5.5", "input": "hi"} assert "data" not in kwargs + + +class _FakeWSExceptions: + class WebSocketException(Exception): + pass + + class InvalidStatusCode(WebSocketException): + def __init__(self) -> None: + super().__init__("HTTP 403") + + # websockets>=15 raises InvalidStatus (not InvalidStatusCode) for a rejected + # client handshake; both must be treated as deterministic. + class InvalidStatus(WebSocketException): + def __init__(self) -> None: + super().__init__("HTTP 401") + + +class _FakeWebsocketsModule: + """Stand-in for the ``websockets`` module so the realtime backend-open retry + can be exercised without a real network handshake (dependency injection, + no monkeypatching).""" + + def __init__(self, outcomes): + # outcomes: list where each item is either an Exception to raise or a + # sentinel object to return as the "connected" websocket. + self._outcomes = list(outcomes) + self.exceptions = _FakeWSExceptions + self.attempts = 0 + self.open_timeouts: list = [] + + async def connect(self, *args, **kwargs): + self.attempts += 1 + self.open_timeouts.append(kwargs.get("open_timeout")) + outcome = self._outcomes.pop(0) + if isinstance(outcome, Exception): + raise outcome + return outcome + + +@pytest.mark.asyncio +async def test_realtime_backend_open_retries_then_succeeds(): + """A hung/slow open handshake is retried; a later fresh attempt connects. + + Regression for intermittent ``1011 timed out during opening handshake``: + the proxy used to surface a single slow upstream handshake to the caller as + a fatal 1011 with no retry. + """ + sentinel = object() + fake = _FakeWebsocketsModule( + [TimeoutError("timed out during opening handshake"), sentinel] + ) + + result = await BaseLLMHTTPHandler._open_realtime_backend_ws( + fake, "wss://backend.example/live", {"Authorization": "Bearer x"}, None + ) + + assert result is sentinel + assert fake.attempts == 2 + # Each attempt must be bounded by a finite open_timeout (not the default/None). + assert all(t is not None and t > 0 for t in fake.open_timeouts) + + +@pytest.mark.asyncio +async def test_realtime_backend_open_raises_after_max_attempts(): + """When every attempt times out, the final error propagates (so the caller + still closes the client socket) rather than looping forever.""" + fake = _FakeWebsocketsModule([TimeoutError("hang")] * 2) + + with pytest.raises(TimeoutError): + await BaseLLMHTTPHandler._open_realtime_backend_ws( + fake, "wss://backend.example/live", {}, None, max_attempts=2 + ) + + assert fake.attempts == 2 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "rejection", + [_FakeWSExceptions.InvalidStatusCode, _FakeWSExceptions.InvalidStatus], +) +async def test_realtime_backend_open_does_not_retry_auth_failure(rejection): + """A deterministic handshake-status rejection (auth/4xx) must not be retried; + retrying cannot help and the upstream status must surface, not a 1011. Both + the websockets<15 (InvalidStatusCode) and >=15 (InvalidStatus) shapes apply.""" + fake = _FakeWebsocketsModule([rejection()]) + + with pytest.raises(_FakeWSExceptions.WebSocketException): + await BaseLLMHTTPHandler._open_realtime_backend_ws( + fake, "wss://backend.example/live", {}, None + ) + + assert fake.attempts == 1 diff --git a/tests/test_litellm/llms/gemini/realtime/test_gemini_realtime_transformation.py b/tests/test_litellm/llms/gemini/realtime/test_gemini_realtime_transformation.py index 85da855f8efa..4947d553317f 100644 --- a/tests/test_litellm/llms/gemini/realtime/test_gemini_realtime_transformation.py +++ b/tests/test_litellm/llms/gemini/realtime/test_gemini_realtime_transformation.py @@ -1145,60 +1145,6 @@ def test_gemini_function_call_output_includes_name(): assert "response" in function_response -def test_gemini_subsequent_session_update_forwards_tools_merged_with_original_setup(): - """A client session.update sent after the auto-setup must forward tools/ - instructions as a follow-up setup, merged with the original setup so we - don't drop the pre-existing config (model, generationConfig, etc.).""" - config = GeminiRealtimeConfig() - - original_setup = { - "setup": { - "model": "models/gemini-2.5-flash-native-audio", - "generationConfig": {"responseModalities": ["AUDIO"]}, - "inputAudioTranscription": {}, - "systemInstruction": {"role": "user", "parts": [{"text": "original"}]}, - } - } - - session_update = { - "type": "session.update", - "session": { - "tools": [ - { - "type": "function", - "function": { - "name": "get_weather", - "description": "Get weather.", - "parameters": { - "type": "object", - "properties": {"location": {"type": "string"}}, - "required": ["location"], - }, - }, - } - ], - "instructions": "Be concise.", - }, - } - - messages = config.transform_realtime_request( - json.dumps(session_update), - "gemini-2.5-flash-native-audio", - session_configuration_request=json.dumps(original_setup), - ) - - assert len(messages) == 1 - follow_up = json.loads(messages[0])["setup"] - assert "tools" in follow_up - assert follow_up["tools"][0]["function_declarations"][0]["name"] == "get_weather" - # systemInstruction overwritten by client's instructions - assert follow_up["systemInstruction"]["parts"][0]["text"] == "Be concise." - # Original generationConfig / model / inputAudioTranscription preserved - assert follow_up["generationConfig"]["responseModalities"] == ["AUDIO"] - assert follow_up["model"] == "models/gemini-2.5-flash-native-audio" - assert follow_up["inputAudioTranscription"] == {} - - def test_gemini_realtime_pipecat_ga_session_voice_and_tools(): """Pipecat OpenAIRealtimeSessionProperties: output_modalities, nested tools, and audio.output.voice (e.g. Kore) must map into Gemini setup.""" @@ -1340,116 +1286,6 @@ def test_gemini_input_audio_buffer_commit_maps_to_activity_end_when_manual_vad() assert json.loads(messages[0]) == {"realtimeInput": {"activityEnd": True}} -def test_gemini_subsequent_session_update_with_turn_detection_only_preserves_original_tools(): - """A subsequent session.update carrying only turn_detection (the - guardrail-injected disable) must keep the original tools/generationConfig.""" - config = GeminiRealtimeConfig() - - original_setup = { - "setup": { - "model": "models/gemini-2.5-flash-native-audio", - "generationConfig": {"responseModalities": ["AUDIO"]}, - "inputAudioTranscription": {}, - "tools": [ - { - "function_declarations": [ - {"name": "lookup", "description": "x", "parameters": {}} - ] - } - ], - } - } - - session_update = { - "type": "session.update", - "session": {"turn_detection": {"create_response": False}}, - } - - messages = config.transform_realtime_request( - json.dumps(session_update), - "gemini-2.5-flash-native-audio", - session_configuration_request=json.dumps(original_setup), - ) - - assert len(messages) == 1 - follow_up = json.loads(messages[0])["setup"] - assert follow_up["tools"] == original_setup["setup"]["tools"] - assert ( - follow_up["realtimeInputConfig"]["automaticActivityDetection"]["disabled"] - is True - ) - - -def test_gemini_follow_up_session_update_preserves_response_modalities_on_partial_generation_config(): - """A follow-up session.update that only sets `temperature` (or any other - generationConfig sub-field) must not wipe `responseModalities` from the - original setup.""" - config = GeminiRealtimeConfig() - - original_setup = { - "setup": { - "model": "models/gemini-2.5-flash-native-audio", - "generationConfig": { - "responseModalities": ["AUDIO"], - "maxOutputTokens": 2048, - }, - "inputAudioTranscription": {}, - } - } - - session_update = { - "type": "session.update", - "session": {"temperature": 0.7}, - } - - messages = config.transform_realtime_request( - json.dumps(session_update), - "gemini-2.5-flash-native-audio", - session_configuration_request=json.dumps(original_setup), - ) - - follow_up = json.loads(messages[0])["setup"] - assert follow_up["generationConfig"]["responseModalities"] == ["AUDIO"] - assert follow_up["generationConfig"]["maxOutputTokens"] == 2048 - assert follow_up["generationConfig"]["temperature"] == 0.7 - - -def test_gemini_subsequent_session_update_preserves_automatic_activity_detection_subfields(): - config = GeminiRealtimeConfig() - - original_setup = { - "setup": { - "model": "models/gemini-2.5-flash-native-audio", - "generationConfig": {"responseModalities": ["AUDIO"]}, - "realtimeInputConfig": { - "automaticActivityDetection": { - "disabled": False, - "silenceDurationMs": 500, - "prefixPaddingMs": 100, - } - }, - } - } - - session_update = { - "type": "session.update", - "session": {"turn_detection": {"create_response": False}}, - } - - messages = config.transform_realtime_request( - json.dumps(session_update), - "gemini-2.5-flash-native-audio", - session_configuration_request=json.dumps(original_setup), - ) - - automatic_activity_detection = json.loads(messages[0])["setup"][ - "realtimeInputConfig" - ]["automaticActivityDetection"] - assert automatic_activity_detection["disabled"] is True - assert automatic_activity_detection["silenceDurationMs"] == 500 - assert automatic_activity_detection["prefixPaddingMs"] == 100 - - def test_gemini_tool_call_id_to_name_evicts_oldest_when_capped(): """The call_id → name LRU must evict the oldest entry once the cap is reached so long sessions with many tool calls don't grow unboundedly, @@ -1732,3 +1568,106 @@ def test_gemini_in_frame_usage_metadata_clears_pending_buffer(): assert usage["output_tokens"] == 2 assert usage["total_tokens"] == 5 assert config._pending_usage_metadata is None + + +def test_gemini_subsequent_session_update_is_dropped_not_resent_as_setup(): + """Regression: Gemini Live accepts exactly one ``setup`` message; a second + one closes the socket with ``1007 Request contains an invalid argument``. + + Once the initial setup has been sent (``session_configuration_request`` is + set), a follow-up session.update must be dropped rather than forwarded as + another setup. Forwarding it tore the session down before the first turn, + which surfaced to callers as silence after the first response and 1011s. + """ + config = GeminiRealtimeConfig() + initial_setup = json.dumps( + { + "setup": { + "model": "models/gemini-2.5-flash", + "generationConfig": {"responseModalities": ["AUDIO"]}, + "inputAudioTranscription": {}, + } + } + ) + follow_up = { + "type": "session.update", + "session": {"instructions": "Updated instructions", "temperature": 0.4}, + } + + messages = config.transform_realtime_request( + json.dumps(follow_up), + "gemini-2.5-flash", + session_configuration_request=initial_setup, + ) + + assert messages == [], ( + "a session.update after the initial setup must be dropped, never " + "forwarded as a second setup (Gemini Live rejects it with 1007)" + ) + + +def test_gemini_subsequent_session_update_with_new_tools_is_dropped(): + """Regression: even a follow-up session.update that *differs* from the initial + setup (e.g. registers tools after connect) must be dropped. + + The previous dedup only skipped follow-ups identical to the initial setup; a + changed one was merged and re-sent as a second setup, still hitting the 1007. + Tools must instead ride on the first session.update. + """ + config = GeminiRealtimeConfig() + initial_setup = json.dumps( + { + "setup": { + "model": "models/gemini-2.5-flash", + "generationConfig": {"responseModalities": ["AUDIO"]}, + } + } + ) + follow_up = { + "type": "session.update", + "session": { + "tools": [ + { + "type": "function", + "function": { + "name": "terminate_call", + "description": "End the call.", + "parameters": {"type": "object", "properties": {}}, + }, + } + ] + }, + } + + messages = config.transform_realtime_request( + json.dumps(follow_up), + "gemini-2.5-flash", + session_configuration_request=initial_setup, + ) + + assert messages == [] + + +def test_gemini_subsequent_guardrail_session_update_dropped_with_warning(caplog): + """A dropped follow-up carrying ``turn_detection.create_response=False`` (the + transcription-guardrail signal) is still dropped, but warns so operators know + the guardrail cannot gate the model's auto-response mid-session on Gemini. + """ + import logging + + config = GeminiRealtimeConfig() + initial_setup = json.dumps({"setup": {"model": "models/gemini-2.5-flash"}}) + follow_up = { + "type": "session.update", + "session": {"turn_detection": {"type": "server_vad", "create_response": False}}, + } + + with caplog.at_level(logging.WARNING, logger="LiteLLM"): + messages = config.transform_realtime_request( + json.dumps(follow_up), + "gemini-2.5-flash", + session_configuration_request=initial_setup, + ) + + assert messages == [] + assert any("Dropping subsequent session.update" in record.message for record in caplog.records) From 9426f7b9ddf4a694839cb5a7514adbab30c9431e Mon Sep 17 00:00:00 2001 From: mubashir1osmani Date: Tue, 30 Jun 2026 12:54:47 -0700 Subject: [PATCH 2/4] fix(logging): route realtime success logging through the bounded worker (#31733) RealTimeStreaming.log_messages dispatched the success handler with a bare asyncio.create_task, bypassing GLOBAL_LOGGING_WORKER (which gives a per-coroutine timeout and a concurrency cap). On a long-lived realtime websocket a slow logging callback left one suspended task per logged turn, each pinning that turn's assembled response, accumulating without bound (~12-15k in-flight under load in a repro) until OOM. Route realtime success logging through the bounded worker so in-flight logging is capped and a hung callback is cancelled at the worker timeout. The chat and responses streaming success-logging paths are intentionally left unchanged: their success callbacks must complete within the call's event-loop run (the non-streaming path pairs the worker with a synchronous callback; the streaming path has no such companion), so deferring them through the worker would drop logs for one-shot SDK calls and breaks test_async_custom_handler_stream. Bounding those paths needs a load-shedding approach and is left to a follow-up. (cherry picked from commit d4c33b2b5922cdc780c7dac31c73a2a21f342540) --- .../litellm_core_utils/realtime_streaming.py | 7 ++++-- .../test_realtime_streaming.py | 23 +++++++++++++++++++ 2 files changed, 28 insertions(+), 2 deletions(-) diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index a35d113fc9d0..97a87ca1faf9 100644 --- a/litellm/litellm_core_utils/realtime_streaming.py +++ b/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)) diff --git a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py index 18547f2f6a30..cad5f6d5b381 100644 --- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py +++ b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py @@ -2750,3 +2750,26 @@ def test_non_bidi_setup_left_untouched_for_followup_capable_providers(): assert streaming._maybe_inject_guardrail_auto_response_disable(msg) == msg finally: litellm.callbacks = [] + + +@pytest.mark.asyncio +async def test_log_messages_routes_async_logging_through_bounded_worker(): + """Realtime success logging must go through GLOBAL_LOGGING_WORKER (bounded + queue + per-coroutine timeout), not a bare asyncio.create_task. A bare task + has no timeout/concurrency cap, so when a logging callback is slow every + realtime turn leaves a suspended task pinning its response in memory -> an + unbounded leak. Regression for that fix.""" + logging_obj = MagicMock() + streaming = RealTimeStreaming(MagicMock(), MagicMock(), logging_obj) + streaming.messages = [{"type": "session.created"}] + + with ( + patch("litellm.litellm_core_utils.realtime_streaming.GLOBAL_LOGGING_WORKER") as mock_worker, + patch("litellm.litellm_core_utils.realtime_streaming.asyncio.create_task") as mock_create_task, + patch("litellm.litellm_core_utils.realtime_streaming.executor.submit"), + ): + await streaming.log_messages() + + mock_worker.ensure_initialized_and_enqueue.assert_called_once() + # the bare create_task path must no longer be used for success logging + mock_create_task.assert_not_called() From 9c88929209146087406951c8e78508630a192002 Mon Sep 17 00:00:00 2001 From: Yuneng Jiang Date: Tue, 30 Jun 2026 19:00:58 -0700 Subject: [PATCH 3/4] =?UTF-8?q?bump:=20version=201.90.1=20=E2=86=92=201.90?= =?UTF-8?q?.2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pyproject.toml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 8c8cc6aa2902..063a32f01388 100644 --- a/pyproject.toml +++ b/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", ] From 8f96737c78175cad9deed43a6249f0d14a2c0837 Mon Sep 17 00:00:00 2001 From: Yuneng Jiang Date: Tue, 30 Jun 2026 19:01:24 -0700 Subject: [PATCH 4/4] chore: refresh uv.lock for 1.90.2 --- uv.lock | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/uv.lock b/uv.lock index 594b68b0bd29..a48c2ac373e8 100644 --- a/uv.lock +++ b/uv.lock @@ -9,7 +9,7 @@ resolution-markers = [ ] [options] -exclude-newer = "2026-06-27T01:16:05.524641Z" +exclude-newer = "2026-06-28T02:01:12.691586Z" exclude-newer-span = "P3D" [manifest] @@ -3245,7 +3245,7 @@ wheels = [ [[package]] name = "litellm" -version = "1.90.1" +version = "1.90.2" source = { editable = "." } dependencies = [ { name = "aiohttp" },