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/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)

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
 

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

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",

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 {}

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,
     )

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
 

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

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={},
             ),

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,
 ):

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,
 ):