diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index c56a70177bfa..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)) @@ -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], 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/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", ] 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..cad5f6d5b381 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,95 @@ 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 = [] + + +@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() 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) 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" },