From f34e5aa189556c8ff4dd57e54b68e068ff3c8a5d Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Wed, 21 Oct 2026 00:26:40 -0700 Subject: [PATCH] fix(gemini realtime): answer an error for session updates Gemini Live cannot apply An output modality the model cannot produce, or a session.update sent after the first content, now gets an invalid_request_error event instead of a silent swap or drop, unless drop_params is set. The synthetic session.created reports the model's default modality, setupComplete no longer adds an extra session.updated, and merging nested session updates is linear in their size --- .../litellm_core_utils/realtime_streaming.py | 30 +++- .../llms/base_llm/realtime/transformation.py | 29 +++- litellm/llms/bedrock/realtime/handler.py | 20 +-- .../llms/bedrock/realtime/transformation.py | 19 +-- .../llms/gemini/realtime/transformation.py | 53 ++++++- .../llms/vertex_ai/realtime/transformation.py | 5 +- litellm/utils.py | 2 +- .../test_realtime_streaming.py | 136 +++++++++++++++++- .../custom_httpx/test_llm_http_handler.py | 14 +- .../test_gemini_realtime_transformation.py | 36 +++++ .../test_vertex_ai_realtime_transformation.py | 19 +++ 11 files changed, 313 insertions(+), 50 deletions(-) diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index c56edd7778d..ed75031eb9e 100644 --- a/litellm/litellm_core_utils/realtime_streaming.py +++ b/litellm/litellm_core_utils/realtime_streaming.py @@ -148,7 +148,7 @@ def _merge_session_fields( ) return update[key] - return MappingProxyType({key: merged_value(key) for key in (*base, *update)}) + return MappingProxyType({key: merged_value(key) for key in dict.fromkeys((*base, *update))}) def _session_update_frame(session: Mapping[str, object]) -> str: @@ -639,11 +639,30 @@ class RealTimeStreaming: self._usage_meter.observe_client_event(sent_event) return True - def _holds_session_updates_until_first_content(self) -> bool: - if self.provider_config is None or self.session_configuration_request is not None: + def _builds_setup_lazily(self) -> bool: + if self.provider_config is None: return False return self.provider_config.builds_setup_from_session_update() and not self._uses_deferred_backend_setup() + def _holds_session_updates_until_first_content(self) -> bool: + return self.session_configuration_request is None and self._builds_setup_lazily() + + async def _rejected_setup_session_update(self, message: str) -> bool: + provider_config: Final = self.provider_config + if provider_config is None or not self._builds_setup_lazily(): + return False + client_event: Final = _decode_json_object(message) + session: Final = client_event.get("session") + rejection: Final = provider_config.setup_session_update_rejection( + self.model, + _SESSION_FIELDS.validate_python(session) if isinstance(session, Mapping) else _NO_SESSION_FIELDS, + setup_sent=self.session_configuration_request is not None, + ) + if rejection is None: + return False + await self.websocket.send_text(json.dumps(rejection.error_event(client_event.get("event_id")))) + return True + def _held_session_update_message(self) -> str: return _session_update_frame(self._held_session_update or _NO_SESSION_FIELDS) @@ -1222,6 +1241,9 @@ class RealTimeStreaming: break finally: self._flushing_pending_messages_until_setup = False + if self.session_created_sent_to_client and self._builds_setup_lazily(): + await self._maybe_send_guardrail_turn_detection_update() + continue if self.session_created_sent_to_client: # A synthetic session.created (with placeholder defaults) was # already forwarded to the client when we connected. The @@ -1779,6 +1801,8 @@ class RealTimeStreaming: # ``_maybe_send_guardrail_turn_detection_update`` would retry. if msg_type == "session.update" and self.provider_config is None: await self._send_session_gate_once() + if msg_type == "session.update" and await self._rejected_setup_session_update(message): + continue if self._holds_session_updates_until_first_content(): if msg_type == "session.update": await self._hold_session_update(message) diff --git a/litellm/llms/base_llm/realtime/transformation.py b/litellm/llms/base_llm/realtime/transformation.py index 51d34982959..47054b66187 100644 --- a/litellm/llms/base_llm/realtime/transformation.py +++ b/litellm/llms/base_llm/realtime/transformation.py @@ -1,11 +1,13 @@ +import dataclasses from abc import ABC, abstractmethod from collections.abc import Mapping, Sequence from types import TracebackType -from typing import TYPE_CHECKING, Any, Protocol +from typing import TYPE_CHECKING, Any, Literal, Protocol import httpx from typing_extensions import Self +from litellm._uuid import uuid from litellm.types.llms.openai import OpenAIRealtimeStreamSessionEvents from litellm.types.realtime import ( RealtimeInputAudioTranscriptionUsage, @@ -23,6 +25,26 @@ else: LiteLLMLoggingObj = Any +@dataclasses.dataclass(frozen=True, slots=True) +class RealtimeSessionUpdateRejection: + code: Literal["invalid_type", "invalid_value", "session_update_not_applied"] + message: str + param: str + + def error_event(self, client_event_id: object) -> dict[str, object]: + return { + "type": "error", + "event_id": f"event_{uuid.uuid4()}", + "error": { + "type": "invalid_request_error", + "code": self.code, + "message": self.message, + "param": self.param, + "event_id": client_event_id if isinstance(client_event_id, str) else None, + }, + } + + class RealtimeBackend(Protocol): async def __aenter__(self) -> Self: ... @@ -97,6 +119,11 @@ class BaseRealtimeConfig(ABC): def builds_setup_from_session_update(self) -> bool: return False + def setup_session_update_rejection( + self, model: str, session: Mapping[str, object], *, setup_sent: bool + ) -> RealtimeSessionUpdateRejection | None: + return None + def unbilled_usage_on_session_close(self, model: str) -> RealtimeInputAudioTranscriptionUsage | None: return None diff --git a/litellm/llms/bedrock/realtime/handler.py b/litellm/llms/bedrock/realtime/handler.py index e9b2f26e977..00622a1e287 100644 --- a/litellm/llms/bedrock/realtime/handler.py +++ b/litellm/llms/bedrock/realtime/handler.py @@ -17,7 +17,6 @@ from pydantic import JsonValue, TypeAdapter import litellm from litellm._logging import redact_string, verbose_proxy_logger -from litellm._uuid import uuid from litellm.constants import ( BEDROCK_REALTIME_CLOSE_TIMEOUT_SECONDS, BEDROCK_REALTIME_COMMITTED_FAILURE_SCOPE_KEY, @@ -31,13 +30,14 @@ from litellm.litellm_core_utils.aws_partition import get_aws_dns_suffix from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER from litellm.litellm_core_utils.realtime_streaming import DefaultLoggedRealTimeEventTypes +from litellm.llms.base_llm.realtime.transformation import RealtimeSessionUpdateRejection from litellm.types.llms.bedrock import AwsAuthParams from litellm.types.llms.openai import OpenAIRealtimeEvents from litellm.types.realtime import RealtimeResponseTransformInput from ..base_aws_llm import BaseAWSLLM, run_aws_signing from ..common_utils import BedrockError -from .transformation import BedrockRealtimeConfig, BedrockSessionUpdateRejection, requested_modalities +from .transformation import BedrockRealtimeConfig, requested_modalities _CLIENT_MESSAGE_ADAPTER: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue) @@ -256,23 +256,11 @@ async def _ack_session_update( async def _reject_session_update( client_ws: RealtimeClientWebSocket, parsed_client_message: Mapping[str, JsonValue], - rejection: BedrockSessionUpdateRejection, + rejection: RealtimeSessionUpdateRejection, ) -> bool: """Send an error event for a session.update Bedrock cannot apply; False means the client is gone.""" - client_event_id: Final = parsed_client_message.get("event_id") - error_event: Final = { - "type": "error", - "event_id": f"event_{uuid.uuid4()}", - "error": { - "type": "invalid_request_error", - "code": rejection.code, - "message": rejection.message, - "param": rejection.param, - "event_id": client_event_id if isinstance(client_event_id, str) else None, - }, - } try: - await client_ws.send_text(json.dumps(error_event)) + await client_ws.send_text(json.dumps(rejection.error_event(parsed_client_message.get("event_id")))) except Exception as e: # noqa: BLE001 # any send failure means the client is gone verbose_proxy_logger.debug("Client to Bedrock forwarding ended: %s", e, exc_info=True) return False diff --git a/litellm/llms/bedrock/realtime/transformation.py b/litellm/llms/bedrock/realtime/transformation.py index 9cb38822881..97eda5fdbc9 100644 --- a/litellm/llms/bedrock/realtime/transformation.py +++ b/litellm/llms/bedrock/realtime/transformation.py @@ -11,7 +11,7 @@ import math import uuid as uuid_lib from collections.abc import Mapping, Sequence from types import MappingProxyType -from typing import Final, Literal +from typing import Final import httpx from pydantic import ConfigDict, JsonValue, TypeAdapter @@ -19,7 +19,7 @@ from pydantic import ConfigDict, JsonValue, TypeAdapter from litellm._logging import verbose_logger from litellm._uuid import uuid from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj -from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig +from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig, RealtimeSessionUpdateRejection from litellm.llms.bedrock.common_utils import BedrockError from litellm.llms.bedrock.realtime.trigger_audio import ready_trigger_pcm from litellm.types.llms.base import LiteLLMBaseModel @@ -195,13 +195,6 @@ def _malformed_settings(session_config: Mapping[str, JsonValue]) -> tuple[tuple[ return tuple((name, expected) for name, expected, is_malformed in malformed if is_malformed) -@dataclasses.dataclass(frozen=True, slots=True) -class BedrockSessionUpdateRejection: - code: Literal["invalid_type", "invalid_value", "session_update_not_applied"] - message: str - param: str - - @dataclasses.dataclass(frozen=True, slots=True) class BedrockLockedSessionSettings: """The session settings Bedrock Nova Sonic locks in once the prompt starts.""" @@ -532,7 +525,7 @@ class BedrockRealtimeConfig(BaseRealtimeConfig): max_tokens=max_tokens, ) - def session_update_rejection(self, json_message: Mapping[str, JsonValue]) -> BedrockSessionUpdateRejection | None: + def session_update_rejection(self, json_message: Mapping[str, JsonValue]) -> RealtimeSessionUpdateRejection | None: """The error for a session.update this session cannot apply, or None when it applies.""" session_config: Final = json_message.get("session") if not isinstance(session_config, Mapping): @@ -540,7 +533,7 @@ class BedrockRealtimeConfig(BaseRealtimeConfig): malformed: Final = () if self.drop_params else _malformed_settings(session_config) if malformed: name, expected = malformed[0] - return BedrockSessionUpdateRejection( + return RealtimeSessionUpdateRejection( code="invalid_type", message=( f"Invalid type for 'session.{name}': Bedrock Nova Sonic needs {expected} there. " @@ -550,7 +543,7 @@ class BedrockRealtimeConfig(BaseRealtimeConfig): ) unplayable_voice: Final = None if self.drop_params else _unplayable_voice(session_config) if unplayable_voice is not None: - return BedrockSessionUpdateRejection( + return RealtimeSessionUpdateRejection( code="invalid_value", message=( f"Invalid value for 'session.voice': '{unplayable_voice}' is not a Bedrock Nova Sonic voice. " @@ -561,7 +554,7 @@ class BedrockRealtimeConfig(BaseRealtimeConfig): if not self.prompt_started: return None if self._apply_locked_settings(session_config, self._started_settings) != self._started_settings: - return BedrockSessionUpdateRejection( + return RealtimeSessionUpdateRejection( code="session_update_not_applied", message="Bedrock Nova Sonic cannot change tools, instructions, voice, output audio format, temperature or max output tokens after the session has started, so this session.update was not applied", param="session", diff --git a/litellm/llms/gemini/realtime/transformation.py b/litellm/llms/gemini/realtime/transformation.py index 1cde5f4956e..a8f11ba5484 100644 --- a/litellm/llms/gemini/realtime/transformation.py +++ b/litellm/llms/gemini/realtime/transformation.py @@ -14,7 +14,7 @@ import litellm from litellm import verbose_logger from litellm._uuid import uuid from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj -from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig +from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig, RealtimeSessionUpdateRejection from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import ( VertexGeminiConfig, ) @@ -154,8 +154,9 @@ def _system_instruction_text(system_instruction: HttpxContentType | None) -> str class GeminiRealtimeConfig(BaseRealtimeConfig): _TOOL_CALL_ID_TO_NAME_MAX = 256 # LRU cap for call_id→name mapping - def __init__(self): + def __init__(self, drop_params: bool = False): super().__init__() + self.drop_params = drop_params self._tool_call_id_to_name: OrderedDict[str, str] = OrderedDict() # Gemini Live sometimes emits usageMetadata in a standalone frame between # turns; buffer it here so the next response.done carries the token counts. @@ -467,6 +468,47 @@ class GeminiRealtimeConfig(BaseRealtimeConfig): return () return tuple(str(modality).upper() for modality in _JSON_ARRAY.validate_python(value)) + @staticmethod + def _unsupported_response_modalities(model: str, requested: tuple[str, ...]) -> tuple[str, ...]: + unknown: Final = tuple(modality for modality in requested if modality not in _GEMINI_LIVE_RESPONSE_MODALITIES) + if unknown: + return unknown + if GeminiRealtimeConfig._is_audio_only_live_model(model) and frozenset(requested) == frozenset(("TEXT",)): + return ("TEXT",) + if GeminiRealtimeConfig._is_text_only_live_model(model) and "AUDIO" in requested: + return ("AUDIO",) + return () + + def setup_session_update_rejection( + self, model: str, session: Mapping[str, object], *, setup_sent: bool + ) -> RealtimeSessionUpdateRejection | None: + if self.drop_params: + return None + if setup_sent: + return RealtimeSessionUpdateRejection( + code="session_update_not_applied", + message=( + "Gemini Live takes its session configuration once, with the first conversation content, " + "so this session.update was not applied. Send session.update before the first content, " + "or set drop_params to drop it instead" + ), + param="session", + ) + param: Final = "output_modalities" if "output_modalities" in session else "modalities" + unsupported: Final = GeminiRealtimeConfig._unsupported_response_modalities( + model, GeminiRealtimeConfig._requested_response_modalities(session.get(param)) + ) + if not unsupported: + return None + return RealtimeSessionUpdateRejection( + code="invalid_value", + message=( + f"Invalid value for 'session.{param}': {model} cannot produce " + f"{', '.join(modality.lower() for modality in unsupported)} output. Set drop_params to drop it instead" + ), + param=f"session.{param}", + ) + @staticmethod def _coerce_response_modalities(model: str, modalities: Sequence[object]) -> tuple[str, ...]: """Drop modalities Gemini Live cannot produce at all, swap the ones this @@ -708,10 +750,9 @@ class GeminiRealtimeConfig(BaseRealtimeConfig): logging_session_id: str, session_configuration_request: str | None = None, ) -> OpenAIRealtimeStreamSessionEvents: - if session_configuration_request: - session_configuration_request_dict: BidiGenerateContentSetup = _parse_setup(session_configuration_request) - else: - session_configuration_request_dict = {} + session_configuration_request_dict: Final = _parse_setup( + session_configuration_request or self.session_configuration_request(model) + ) _model: Final = session_configuration_request_dict.get("model") or model generation_config: Final = session_configuration_request_dict.get("generationConfig", {}) or {} diff --git a/litellm/llms/vertex_ai/realtime/transformation.py b/litellm/llms/vertex_ai/realtime/transformation.py index 55f1cf58cbe..421e324ede6 100644 --- a/litellm/llms/vertex_ai/realtime/transformation.py +++ b/litellm/llms/vertex_ai/realtime/transformation.py @@ -38,8 +38,8 @@ class VertexAIRealtimeConfig(GeminiRealtimeConfig): (they require async I/O) and injected at construction time. """ - def __init__(self, access_token: str, project: str, location: str) -> None: - super().__init__() + def __init__(self, access_token: str, project: str, location: str, drop_params: bool = False) -> None: + super().__init__(drop_params=drop_params) self._access_token = access_token self._project = project self._location = location @@ -261,4 +261,5 @@ def vertex_realtime_config( access_token=access_token, project=project, location=VertexBase.get_vertex_region(vertex_region=location, model=model), + drop_params=drop_params, ) diff --git a/litellm/utils.py b/litellm/utils.py index 379eddb3155..0c0f7019854 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -10487,7 +10487,7 @@ class ProviderConfigManager: if LlmProviders.GEMINI == provider: from litellm.llms.gemini.realtime.transformation import GeminiRealtimeConfig - return GeminiRealtimeConfig() + return GeminiRealtimeConfig(drop_params=drop_params) if LlmProviders.META == provider: from litellm.llms.meta.realtime.transformation import MetaRealtimeConfig diff --git a/tests/unit/litellm_core_utils/test_realtime_streaming.py b/tests/unit/litellm_core_utils/test_realtime_streaming.py index 6b915316d6c..1f78823dac7 100644 --- a/tests/unit/litellm_core_utils/test_realtime_streaming.py +++ b/tests/unit/litellm_core_utils/test_realtime_streaming.py @@ -1,6 +1,7 @@ import asyncio import base64 import json +import time from collections.abc import Coroutine, Mapping, Sequence from dataclasses import dataclass from typing import Final, Literal @@ -1264,6 +1265,7 @@ async def test_client_ack_caches_setup_to_prevent_duplicate_session_update_setup websocket.send_text = AsyncMock() provider_config = MagicMock() + provider_config.setup_session_update_rejection = MagicMock(return_value=None) def _transform(message: str, model: str, session_configuration_request=None): if session_configuration_request is None: @@ -2419,6 +2421,7 @@ async def test_subsequent_session_update_cannot_reenable_vad_when_guardrails_act logging_obj.success_handler = MagicMock() provider_config = MagicMock() + provider_config.setup_session_update_rejection = MagicMock(return_value=None) transformed_messages = [] def mock_transform(msg, model, session_config): @@ -2471,6 +2474,7 @@ async def test_follow_up_setup_updates_cached_session_configuration_request(): logging_obj.success_handler = MagicMock() provider_config = MagicMock() + provider_config.setup_session_update_rejection = MagicMock(return_value=None) follow_up_setup = json.dumps( { "setup": { @@ -2504,7 +2508,9 @@ async def test_follow_up_setup_updates_cached_session_configuration_request(): assert streaming.session_configuration_request == follow_up_setup -def _gemini_default_mode_streaming(monkeypatch, client_messages, model="gemini-live-2.5-flash-native-audio"): +def _gemini_default_mode_streaming( + monkeypatch, client_messages, model="gemini-live-2.5-flash-native-audio", drop_params=False +): monkeypatch.setattr(litellm, "gemini_live_defer_setup", False, raising=False) from litellm.llms.gemini.realtime.transformation import GeminiRealtimeConfig @@ -2517,12 +2523,16 @@ def _gemini_default_mode_streaming(monkeypatch, client_messages, model="gemini-l client_ws, backend_ws, MagicMock(), - provider_config=GeminiRealtimeConfig(), + provider_config=GeminiRealtimeConfig(drop_params=drop_params), model=model, ) return streaming, backend_ws +def _events_sent_to_client(streaming: RealTimeStreaming) -> list[dict]: + return [json.loads(call.args[0]) for call in streaming.websocket.send_text.await_args_list] + + _TEXT_ITEM_CREATE = json.dumps( { "type": "conversation.item.create", @@ -2610,6 +2620,26 @@ def test_merge_session_fields_replaces_nested_mappings_wholesale_past_the_depth_ assert level(merged, 9)["new"] == 1 + +def _merge_cost(levels: int, leaves: int) -> float: + def nested(depth: int) -> dict[str, object]: + return {"child": nested(depth - 1)} if depth else {f"k{index}": index for index in range(leaves)} + + session: Final = nested(levels) + + def once() -> float: + started: Final = time.perf_counter() + _merge_session_fields(session, session) + return time.perf_counter() - started + + return min(once() for _ in range(3)) + + +def test_merge_session_fields_cost_stays_flat_as_matching_updates_nest_deeper(): + growth: Final = _merge_cost(7, 2000) / _merge_cost(1, 2000) + + assert growth < 8.0, f"merging seven nested levels cost {growth:.1f}x one level" + @pytest.mark.asyncio async def test_default_mode_acknowledges_each_held_session_update_before_any_backend_frame(monkeypatch): """A client that waits for session.updated before sending content must not @@ -2633,6 +2663,108 @@ async def test_default_mode_acknowledges_each_held_session_update_before_any_bac backend_ws.send.assert_not_awaited() +_LATE_INSTRUCTIONS_UPDATE = json.dumps( + {"type": "session.update", "event_id": "evt_late", "session": {"instructions": "Be brief."}} +) + + +@pytest.mark.asyncio +async def test_default_mode_answers_a_session_update_after_the_first_content_with_an_error(monkeypatch): + streaming, backend_ws = _gemini_default_mode_streaming(monkeypatch, [_TEXT_ITEM_CREATE, _LATE_INSTRUCTIONS_UPDATE]) + + await streaming.client_ack_messages() + + errors: Final = [event["error"] for event in _events_sent_to_client(streaming) if event["type"] == "error"] + assert [(error["code"], error["event_id"]) for error in errors] == [("session_update_not_applied", "evt_late")] + frames: Final = [json.loads(call.args[0]) for call in backend_ws.send.await_args_list] + assert [next(iter(frame)) for frame in frames] == ["setup", "clientContent"] + + +@pytest.mark.asyncio +async def test_default_mode_drops_a_late_session_update_quietly_under_drop_params(monkeypatch): + streaming, backend_ws = _gemini_default_mode_streaming( + monkeypatch, [_TEXT_ITEM_CREATE, _LATE_INSTRUCTIONS_UPDATE], drop_params=True + ) + + await streaming.client_ack_messages() + + assert _events_sent_to_client(streaming) == [] + assert backend_ws.send.await_count == 2 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("model", "requested_output"), + [ + ("gemini-live-2.5-flash-native-audio", {"modalities": ["text"]}), + ("gemini-live-2.5-flash", {"output_modalities": ["image"]}), + ], +) +async def test_default_mode_rejects_an_output_modality_the_model_cannot_produce(monkeypatch, model, requested_output): + update: Final = json.dumps( + {"type": "session.update", "session": {**requested_output, "instructions": "Be brief."}} + ) + streaming, backend_ws = _gemini_default_mode_streaming(monkeypatch, [update, _TEXT_ITEM_CREATE], model=model) + + await streaming.client_ack_messages() + + events: Final = _events_sent_to_client(streaming) + assert [(event["type"], event["error"]["code"]) for event in events] == [("error", "invalid_value")] + setup: Final = json.loads(backend_ws.send.await_args_list[0].args[0])["setup"] + assert "systemInstruction" not in setup + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("modalities", "drop_params"), + [(["text", "audio"], False), (["text"], True)], + ids=["openai-default-pair", "text-dropped-under-drop-params"], +) +async def test_default_mode_applies_the_update_with_audio_output_on_an_audio_only_model( + monkeypatch, modalities, drop_params +): + update: Final = json.dumps( + {"type": "session.update", "session": {"modalities": modalities, "instructions": "Be brief."}} + ) + streaming, backend_ws = _gemini_default_mode_streaming( + monkeypatch, [update, _TEXT_ITEM_CREATE], drop_params=drop_params + ) + + await streaming.client_ack_messages() + + assert [event["type"] for event in _events_sent_to_client(streaming)] == ["session.updated"] + setup: Final = json.loads(backend_ws.send.await_args_list[0].args[0])["setup"] + assert setup["generationConfig"]["responseModalities"] == ["AUDIO"] + assert setup["systemInstruction"]["parts"][0]["text"] == "Be brief." + + +@pytest.mark.asyncio +async def test_default_mode_sends_no_extra_session_updated_when_gemini_confirms_the_setup(monkeypatch): + """Each held session.update was already acknowledged, so Gemini's setupComplete + must not reach the client as one more session.updated nobody asked for.""" + monkeypatch.setattr(litellm, "gemini_live_defer_setup", False, raising=False) + from litellm.llms.gemini.realtime.transformation import GeminiRealtimeConfig + + model: Final = "gemini-live-2.5-flash-native-audio" + config: Final = GeminiRealtimeConfig() + client_ws: Final = MagicMock() + client_ws.send_text = AsyncMock() + backend_ws: Final = MagicMock() + backend_ws.send = AsyncMock() + backend_ws.recv = AsyncMock(side_effect=[json.dumps({"setupComplete": {}}).encode(), ConnectionClosed(None, None)]) + logging_obj: Final = _mock_logging() + logging_obj.litellm_trace_id = "trace_lazy" + logging_obj.async_success_handler = AsyncMock() + logging_obj.success_handler = MagicMock() + streaming: Final = RealTimeStreaming(client_ws, backend_ws, logging_obj, provider_config=config, model=model) + streaming.session_created_sent_to_client = True + streaming.session_configuration_request = config.session_configuration_request(model) + + await streaming.backend_to_client_send_messages() + + assert [json.loads(call.args[0])["type"] for call in client_ws.send_text.await_args_list] == [] + + @pytest.mark.asyncio async def test_default_mode_folds_transcription_guardrail_vad_disable_into_the_merged_setup(monkeypatch): """A transcription guardrail gates every turn, so its auto-response disable diff --git a/tests/unit/llms/custom_httpx/test_llm_http_handler.py b/tests/unit/llms/custom_httpx/test_llm_http_handler.py index 42ec2d0bcec..0cdccb528fa 100644 --- a/tests/unit/llms/custom_httpx/test_llm_http_handler.py +++ b/tests/unit/llms/custom_httpx/test_llm_http_handler.py @@ -2766,7 +2766,7 @@ def _realtime_logging_obj(): @pytest.mark.asyncio @pytest.mark.parametrize( - "model, session, backend_modality, client_modality, backend_turn", + "model, session, backend_modality, client_modality, backend_turn, drop_params", [ ( "gemini-live-2.5-flash", @@ -2774,27 +2774,29 @@ def _realtime_logging_obj(): "TEXT", "text", [_GEMINI_LIVE_TEXT_TURN], + False, ), - ("gemini-live-2.5-flash", {"modalities": ["text"]}, "TEXT", "text", [_GEMINI_LIVE_TEXT_TURN]), + ("gemini-live-2.5-flash", {"modalities": ["text"]}, "TEXT", "text", [_GEMINI_LIVE_TEXT_TURN], False), ( "gemini-live-2.5-flash-native-audio", {"type": "realtime", "output_modalities": ["text"]}, "AUDIO", "audio", [], + True, ), ], ) async def test_async_realtime_vertex_first_session_update_builds_the_only_setup( - model, session, backend_modality, client_modality, backend_turn, monkeypatch + model, session, backend_modality, client_modality, backend_turn, drop_params: bool, monkeypatch ): """Regression for text-only Vertex Live sessions answered in audio: the connect-time default setup (AUDIO) used to go out before the client's session.update, and Gemini accepts only one setup per connection, so the requested modality was dropped and billed at the audio rate. The setup now goes out on the client's first message, built from session.update when that - comes first. An audio-only model still gets AUDIO, and the client learns the - downgrade from session.updated.""" + comes first. Under drop_params an audio-only model still gets AUDIO, and the + client learns the downgrade from session.updated.""" from litellm.llms.vertex_ai.realtime.transformation import VertexAIRealtimeConfig monkeypatch.setattr(litellm, "gemini_live_defer_setup", False, raising=False) @@ -2825,7 +2827,7 @@ async def test_async_realtime_vertex_first_session_update_builds_the_only_setup( websocket=client_ws, logging_obj=_realtime_logging_obj(), provider_config=VertexAIRealtimeConfig( - access_token="fake-token", project="fake-project", location="us-central1" + access_token="fake-token", project="fake-project", location="us-central1", drop_params=drop_params ), headers={}, ), diff --git a/tests/unit/llms/gemini/realtime/test_gemini_realtime_transformation.py b/tests/unit/llms/gemini/realtime/test_gemini_realtime_transformation.py index ccb824b080e..5cd60b7de4b 100644 --- a/tests/unit/llms/gemini/realtime/test_gemini_realtime_transformation.py +++ b/tests/unit/llms/gemini/realtime/test_gemini_realtime_transformation.py @@ -1983,6 +1983,42 @@ def test_gemini_transcribe_live_coerces_audio_modality_to_text(modalities, patch assert setup["generationConfig"]["responseModalities"] == ["TEXT"] +@pytest.mark.parametrize("model", ["gemini-3.5-transcribe-live", "gemini/gemini-3.5-transcribe-live"]) +def test_gemini_transcribe_live_synthetic_session_created_reports_text_output( + model, patch_gemini_transcribe_live_cost_map_entry +): + session = GeminiRealtimeConfig().transform_session_created_event(model, "sess_transcribe")["session"] + + assert session["modalities"] == ["text"] + + +@pytest.mark.parametrize("modalities", [["audio"], ["audio", "text"]]) +@pytest.mark.parametrize("drop_params", [False, True]) +def test_gemini_transcribe_live_audio_output_is_rejected_unless_drop_params( + modalities, drop_params, patch_gemini_transcribe_live_cost_map_entry +): + rejection = GeminiRealtimeConfig(drop_params=drop_params).setup_session_update_rejection( + "gemini-3.5-transcribe-live", {"output_modalities": modalities}, setup_sent=False + ) + + assert (rejection is None) is drop_params + assert rejection is None or rejection.param == "session.output_modalities" + + +@pytest.mark.parametrize("drop_params", [False, True]) +def test_gemini_provider_realtime_config_carries_drop_params(drop_params): + from litellm.types.utils import LlmProviders + from litellm.utils import ProviderConfigManager + + model = "gemini-live-2.5-flash-native-audio" + config = ProviderConfigManager.get_provider_realtime_config( + model=model, provider=LlmProviders.GEMINI, drop_params=drop_params + ) + + rejection = config.setup_session_update_rejection(model, {"modalities": ["text"]}, setup_sent=False) + assert (rejection is None) is drop_params + + def test_gemini_chat_model_with_text_output_modalities_keeps_audio_eager_setup( patch_gemini_transcribe_live_cost_map_entry, ): diff --git a/tests/unit/llms/vertex_ai/realtime/test_vertex_ai_realtime_transformation.py b/tests/unit/llms/vertex_ai/realtime/test_vertex_ai_realtime_transformation.py index 8c4e098417b..714a29c36a2 100644 --- a/tests/unit/llms/vertex_ai/realtime/test_vertex_ai_realtime_transformation.py +++ b/tests/unit/llms/vertex_ai/realtime/test_vertex_ai_realtime_transformation.py @@ -179,6 +179,25 @@ def test_vertex_audio_only_live_model_coerces_text_modality_to_audio( assert setup["generationConfig"]["responseModalities"] == ["AUDIO"] +@pytest.mark.parametrize("drop_params", [False, True]) +def test_vertex_realtime_config_carries_drop_params(drop_params, patch_native_audio_cost_map_entry): + from litellm.llms.vertex_ai.realtime.transformation import vertex_realtime_config + + cfg: Final = vertex_realtime_config( + _NATIVE_AUDIO_MODEL, + access_token="tok", + resolve_access_token=AsyncMock(return_value="tok"), + project="my-proj", + location="us-central1", + drop_params=drop_params, + ) + + rejection: Final = cfg.setup_session_update_rejection( + _NATIVE_AUDIO_MODEL, {"modalities": ["text"]}, setup_sent=False + ) + assert (rejection is None) is drop_params + + def test_vertex_session_update_normalizes_ga_remapped_fields( patch_native_audio_cost_map_entry, ):