fix(realtime)!: send the Gemini Live setup at the first content so session.update settings apply (#39864)

* fix(realtime): send Gemini Live setup on the client's first message so session.update modalities apply

Gemini and Vertex AI Live accept exactly one setup per session, and the proxy sent its
default audio setup at connect, so a client's session.update (GA output_modalities or
beta modalities) was dropped and every answer came back as audio. The setup now goes
out on the client's first message: a session.update becomes that setup, anything else
gets the model default. Bedrock Nova Sonic keeps its connect-time setup through a
provider hook. Text setups label deltas response.output_text.delta, and downgrading
TEXT to AUDIO on an audio-only model logs a warning

* fix(realtime): keep one Gemini Live response modality and skip the "inf" token limit in the setup

* fix(realtime): merge every session.update sent before the first content frame into the Gemini Live setup

Gemini and Vertex Live accept one setup per connection, so the bridge used to
turn only the first session.update into that setup and drop the rest without
telling the client. Now every session.update that arrives before the first
content frame is held, deep-merged (later values win per field), acknowledged
with a session.updated built from the setup it would produce, and the merged
session becomes the one setup sent with the first conversation item, audio
frame, or tool result. The session events also surface the systemInstruction
text as instructions, which the old string check never matched because the
setup carries it as a Content object.

* fix(realtime): cap the session.update merge depth so the recursive detector accepts it

_merge_session_fields recurses once per nested mapping level, so CI's
recursive_detector flagged it. Bound the descent at eight levels (the OpenAI
session object nests four deep at most), replace anything deeper wholesale
with the later session.update's value, and list the helper with the other
depth-capped functions the detector allows.

* fix(gemini-live): accept a string output modality and drop values Live cannot produce

A bare string output_modalities is read as a one-element list and modality values Gemini Live cannot produce are dropped with a warning instead of reaching the backend, so a malformed session.update no longer costs the session. Transcription guardrails now fold their VAD disable into the first held session.update only, so the merged setup carries it exactly once.

* 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

* fix(gemini realtime): send the setup before guardrail blocks and idle deadlines, mark push-to-talk turns

A pre_call guardrail blocking the first typed message now goes out after the
setup, so Gemini and Vertex no longer close the session with 1007.

Push-to-talk sessions (turn_detection null or create_response false) wrap each
audio turn in empty activityStart and activityEnd messages, which Gemini needs
when automatic activity detection is off.

A session with no content after 60 seconds sends its setup, since Vertex
refuses a setup 300 seconds after connect. Held session.update frames past
10 MB answer session_update_too_large, and deferred mode now errors on a
second session.update unless drop_params is set.

* fix(realtime): ack unchanged Gemini Live session.update resends and honor null turn_detection

An update that leaves the applied setup unchanged now gets session.updated
instead of session_update_not_applied, so clients that resend their session
with the first response keep working. Deferred setup mode merges held updates
like the default mode, turn_detection null turns Gemini's activity detection
off, drop_params drops a voice outside the documented Live voices, and the
lazy setup deadline moves to 240 seconds

* fix(realtime): keep Vertex activity detection off for null turn_detection and set up deferred sessions before text content

* fix(realtime): refuse unscreenable Gemini audio, send held setup after buffered audio, reject unknown Vertex voices

A realtime_input_transcription guardrail cannot screen Gemini Live audio before the answer starts, so audio input under one is refused with an audio_input_not_screenable error instead of reaching the model. In deferred setup mode a session.update that follows buffered audio is sent as setup at once. Only conversation content triggers the setup, so control events no longer lock in a default one. Vertex rejects an undocumented voice on the session.update, or drops it with drop_params, while Google AI Studio forwards any name.

* fix(realtime): hold first content so a following session.update joins the Gemini setup

Convert flat realtime tool schemas through Gemini's schema builder, and collapse a text and audio pair to what the model can produce on text-only models too

* fix(realtime): check AI Studio Live voices like Vertex

* fix(realtime): forward any voice to AI Studio Live models that accept it

* fix(realtime): map OpenAI TTS voices before the Live voice check, keep outage spend rows, share the playground session config

The Live voice gate now judges the Gemini voice an OpenAI TTS name maps to, so fable and onyx reach Vertex as Umbriel and Orus again. An upstream close before the held setup was sent is logged as a session, not a refusal. The playground sends one server_vad session config for typed and voice turns, and the e2e, integration, and dashboard tests follow lazy setup.

* fix(realtime): take the lazy setup clock as a dependency so its tests run on a stepped clock

The deadline and grace tests waited on real 50 ms timers, and the merge cost
check compared wall-clock runs. The session now reads an injected clock,
time.monotonic by default, the two tests step it, and the merge cost check
moves to the CodSpeed benchmarks. The Nova Sonic style handler test passes
the backend through wrap_backend, which main added after the test was written

litellm/constants.py

@@ -420,6 +420,8 @@ REALTIME_CREDENTIAL_RESOLUTION_TIMEOUT_SECONDS: Final = get_env_float(
     "REALTIME_CREDENTIAL_RESOLUTION_TIMEOUT_SECONDS", 20.0
 )
 BEDROCK_REALTIME_CLOSE_TIMEOUT_SECONDS: Final = get_env_float("BEDROCK_REALTIME_CLOSE_TIMEOUT_SECONDS", 5.0)
+REALTIME_LAZY_SETUP_DEADLINE_SECONDS: Final = 240.0
+REALTIME_LAZY_SETUP_CONTENT_GRACE_SECONDS: Final = 1.0
 
 # RFC 6455 caps the close frame payload at 125 bytes, 2 of which carry the status code
 WEBSOCKET_CLOSE_REASON_MAX_BYTES: Final = 123

litellm/litellm_core_utils/realtime_streaming.py

@@ -1,5 +1,6 @@
 import asyncio
 import json
+import time
 import traceback
 from collections.abc import Callable, Coroutine, Mapping, MutableMapping, Sequence
 from dataclasses import dataclass
@@ -15,13 +16,19 @@ from litellm._logging import redact_internal_details_from_client_message, verbos
 from litellm._uuid import uuid
 from litellm.constants import (
     BEDROCK_REALTIME_PENDING_OPENING_MESSAGE_SCOPE_KEY,
+    REALTIME_LAZY_SETUP_CONTENT_GRACE_SECONDS,
+    REALTIME_LAZY_SETUP_DEADLINE_SECONDS,
     REALTIME_METERED_USAGE_KEY,
     REALTIME_SESSION_FAILURE_LOGGED_KEY,
     REALTIME_SESSION_SUCCESS_LOGGED_KEY,
 )
 from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
 from litellm.llms.base_llm.realtime.transcription_protocol import RealtimeTranscriptionProtocolError
-from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig, RealtimeBackend
+from litellm.llms.base_llm.realtime.transformation import (
+    BaseRealtimeConfig,
+    RealtimeBackend,
+    RealtimeSessionUpdateRejection,
+)
 from litellm.types.llms.openai import (
     OpenAIRealtimeEvents,
     OpenAIRealtimeOutputItemDone,
@@ -125,6 +132,65 @@ class _LoggingWorker(Protocol):
 _DECODED_JSON: Final = TypeAdapter(object)
 
 
+_SESSION_FIELDS: Final = TypeAdapter(dict[str, object])
+_NO_SESSION_FIELDS: Final[Mapping[str, object]] = MappingProxyType({})
+_SESSION_MERGE_MAX_DEPTH: Final = 8
+
+
+class _SessionUpdateFrame(TypedDict):
+    type: ReadOnly[str]
+    session: ReadOnly[Mapping[str, object]]
+
+
+def _merge_session_fields(
+    base: Mapping[str, object], update: Mapping[str, object], depth: int = 0
+) -> Mapping[str, object]:
+    def merged_value(key: str) -> object:
+        if key not in update:
+            return base[key]
+        value: Final = update[key]
+        base_value: Final = base.get(key)
+        if depth < _SESSION_MERGE_MAX_DEPTH and isinstance(value, Mapping) and isinstance(base_value, Mapping):
+            return _merge_session_fields(
+                _SESSION_FIELDS.validate_python(base_value), _SESSION_FIELDS.validate_python(value), depth + 1
+            )
+        return update[key]
+
+    return MappingProxyType({key: merged_value(key) for key in dict.fromkeys((*base, *update))})
+
+
+def _session_update_frame(session: Mapping[str, object]) -> str:
+    frame: Final[_SessionUpdateFrame] = {"type": "session.update", "session": session}
+    return json.dumps(frame, default=_SESSION_FIELDS.validate_python)
+
+
+_HELD_SESSION_UPDATE_TOO_LARGE: Final = RealtimeSessionUpdateRejection(
+    code="session_update_too_large",
+    message=(
+        "The session.update frames sent before the first conversation content add up to more than 10 MB, "
+        "so this session.update was not applied"
+    ),
+    param="session",
+)
+
+
+def _unscreened_audio_error_event(client_event_id: object) -> Mapping[str, object]:
+    return {
+        "type": "error",
+        "event_id": f"event_{uuid.uuid4()}",
+        "error": {
+            "type": "invalid_request_error",
+            "code": "audio_input_not_screenable",
+            "message": (
+                "A realtime_input_transcription guardrail is on, and Gemini Live only transcribes a turn after it "
+                "starts answering it, so audio input was not sent. Send text with conversation.item.create instead"
+            ),
+            "param": None,
+            "event_id": client_event_id if isinstance(client_event_id, str) else None,
+        },
+    }
+
+
 def _decode_json_object(payload: str) -> Mapping[str, object]:
     """Decode a realtime frame into its top-level field mapping."""
     return json.loads(payload)
@@ -243,8 +309,17 @@ class RealTimeStreaming:
         event_normalizer: RealtimeEventNormalizer | None = None,
         usage_meter: RealtimeUsageMeter | None = None,
         logging_worker: _LoggingWorker = GLOBAL_LOGGING_WORKER,
+        lazy_setup_deadline_seconds: float = REALTIME_LAZY_SETUP_DEADLINE_SECONDS,
+        lazy_setup_content_grace_seconds: float = REALTIME_LAZY_SETUP_CONTENT_GRACE_SECONDS,
+        clock: Callable[[], float] = time.monotonic,
     ):
         self.websocket: _ClientWebSocket = websocket
+        self._clock: Final = clock
+        self._lazy_setup_deadline: Final = clock() + lazy_setup_deadline_seconds
+        self._lazy_setup_content_grace_seconds: Final = lazy_setup_content_grace_seconds
+        self._held_client_content: tuple[str, ...] = ()
+        self._held_client_content_bytes: int = 0
+        self._held_client_content_deadline: float | None = None
         self._replay_scope: Final = client_websocket_scope(websocket)
         self.backend_ws = backend_ws
         self.logging_obj = logging_obj
@@ -300,6 +375,10 @@ class RealTimeStreaming:
         # Gemini Live rejects a follow-up BidiGenerateContentSetup once any
         # content (realtimeInput / clientContent / toolResponse) has been sent.
         self._content_sent_after_setup: bool = False
+        self._held_session_update: Mapping[str, object] | None = None
+        self._applied_session_fields: Mapping[str, object] | None = None
+        self._held_session_update_bytes: int = 0
+        self._unscreened_audio_refused: bool = False
         # Whether this is a transcription-only session (session.type == "transcription",
         # e.g. gpt-realtime-whisper). Such sessions must not be sent response.create and
         # their input_audio_transcription.completed usage drives duration-based cost.
@@ -337,6 +416,17 @@ class RealTimeStreaming:
         ]
     )
     _CLIENT_AUDIO_BUFFER_COMMIT_TYPES = frozenset(["input_audio_buffer.commit", "input_audio_buffer.end"])
+    _CLIENT_CONTENT_TYPES = frozenset(
+        [
+            "conversation.item.create",
+            "response.create",
+            "input_audio_buffer.append",
+            "input_audio_buffer.commit",
+            "input_audio_buffer.end",
+        ]
+    )
+    _HELD_CONTENT_TYPES: Final = frozenset(["conversation.item.create", "input_audio_buffer.append"])
+    _DEFERRED_SETUP_HELD_CONTENT_TYPES: Final = frozenset(["conversation.item.create"])
     _AUDIO_FORMAT_MAP: dict[str, dict[str, str | int]] = {
         "pcm16": {"type": "audio/pcm", "rate": 24000},
         "g711_ulaw": {"type": "audio/G711-ulaw", "rate": 8000},
@@ -377,7 +467,7 @@ class RealTimeStreaming:
             message_obj: dict[str, Any] = cast(dict[str, Any], message)
         else:
             message_obj = cast(dict[str, Any], json.loads(message))
-        self._collect_tool_calls_from_response_done(cast(dict, message_obj))
+        self._collect_tool_calls_from_response_done(message_obj)
         if not self._should_store_message(message_obj):
             return
         try:
@@ -502,7 +592,7 @@ class RealTimeStreaming:
         self.store_message(flush_event)
         self._capture_transcription_usage(flush_event)
 
-    def _collect_tool_calls_from_response_done(self, event_obj: dict | OpenAIRealtimeEvents) -> None:
+    def _collect_tool_calls_from_response_done(self, event_obj: Mapping[str, object] | OpenAIRealtimeEvents) -> None:
         """Extract function_call items from response.done events for spend logging."""
         try:
             if event_obj.get("type") != "response.done":
@@ -607,6 +697,167 @@ class RealTimeStreaming:
                 self._usage_meter.observe_client_event(sent_event)
         return True
 
+    def _builds_setup_lazily(self) -> bool:
+        if self.provider_config is None:
+            return False
+        return self.provider_config.builds_setup_from_session_update()
+
+    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 provider_config.builds_setup_from_session_update():
+            return False
+        client_event: Final = _decode_json_object(message)
+        session: Final = client_event.get("session")
+        update: Final = _SESSION_FIELDS.validate_python(session) if isinstance(session, Mapping) else _NO_SESSION_FIELDS
+        if self._leaves_applied_setup_unchanged(update):
+            await self._acknowledge_session_setup(self._applied_session_fields or _NO_SESSION_FIELDS)
+            return True
+        rejection: Final = provider_config.setup_session_update_rejection(
+            self.model, update, 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)
+
+    async def _hold_session_update(self, message: str) -> None:
+        client_event: Final = _decode_json_object(message)
+        if self._held_session_update_bytes + len(message.encode("utf-8")) > RealTimeStreaming._MAX_BUFFERED_BYTES:
+            await self.websocket.send_text(
+                json.dumps(_HELD_SESSION_UPDATE_TOO_LARGE.error_event(client_event.get("event_id")))
+            )
+            return
+        session: Final = client_event.get("session")
+        update: Final = _SESSION_FIELDS.validate_python(session) if isinstance(session, Mapping) else _NO_SESSION_FIELDS
+        held: Final = _merge_session_fields(self._held_session_update or _NO_SESSION_FIELDS, update)
+        self._held_session_update = held
+        self._held_session_update_bytes = len(_session_update_frame(held).encode("utf-8"))
+        await self._acknowledge_session_setup(held)
+
+    def _setup_preview(self, session: Mapping[str, object]) -> str | None:
+        if self.provider_config is None:
+            return None
+        preview: Final = self.provider_config.transform_realtime_request(
+            _session_update_frame(session), self.model, None
+        )
+        return next(
+            (frame for frame in preview if isinstance(frame, str) and "setup" in _decode_json_object(frame)), None
+        )
+
+    def _leaves_applied_setup_unchanged(self, update: Mapping[str, object]) -> bool:
+        applied: Final = self._applied_session_fields
+        if self.session_configuration_request is None or applied is None:
+            return False
+        applied_setup: Final = self._setup_preview(applied)
+        merged_setup: Final = self._setup_preview(_merge_session_fields(applied, update))
+        return (
+            applied_setup is not None
+            and merged_setup is not None
+            and _decode_json_object(applied_setup) == _decode_json_object(merged_setup)
+        )
+
+    async def _acknowledge_session_setup(self, session: Mapping[str, object]) -> None:
+        if self.provider_config is None:
+            return
+        setup: Final = self._setup_preview(session)
+        if setup is None:
+            return
+        acknowledged: Final = self.provider_config.transform_session_created_event(
+            self.model, str(self.logging_obj.litellm_trace_id), setup
+        )
+        if acknowledged is None:
+            return
+        event: Final[Mapping[str, object]] = {**acknowledged, "type": "session.updated"}
+        event_str: Final = json.dumps(event)
+        self.store_message(event_str)
+        await self._send_event_to_client(event, event_str)
+
+    async def _send_held_session_update_as_setup(self) -> None:
+        held: Final = self._held_session_update
+        if held is None:
+            return
+        message: Final = self._held_session_update_message()
+        self._held_session_update = None
+        self._held_session_update_bytes = 0
+        if await self._send_to_backend(message):
+            self._applied_session_fields = held
+            await self._send_held_client_content()
+
+    def _holds_client_content(self, msg_type: str | None, message: str) -> bool:
+        held_types: Final = (
+            RealTimeStreaming._DEFERRED_SETUP_HELD_CONTENT_TYPES
+            if self._uses_deferred_backend_setup()
+            else RealTimeStreaming._HELD_CONTENT_TYPES
+        )
+        return (
+            msg_type in held_types
+            and self._held_session_update is not None
+            and self._holds_session_updates_until_first_content()
+            and not self._pending_messages_until_setup
+            and len(self._held_client_content) < RealTimeStreaming._MAX_BUFFERED_MESSAGES
+            and self._held_client_content_bytes + len(message.encode("utf-8")) <= RealTimeStreaming._MAX_BUFFERED_BYTES
+        )
+
+    def _hold_client_content(self, message: str) -> None:
+        if self._held_client_content_deadline is None:
+            self._held_client_content_deadline = self._clock() + self._lazy_setup_content_grace_seconds
+        self._held_client_content = (*self._held_client_content, message)
+        self._held_client_content_bytes += len(message.encode("utf-8"))
+
+    async def _send_held_client_content(self) -> None:
+        held: Final = self._held_client_content
+        self._held_client_content = ()
+        self._held_client_content_bytes = 0
+        self._held_client_content_deadline = None
+        for message in held:
+            await self._send_to_backend(message)
+
+    def _setup_deadline(self) -> float:
+        content_deadline: Final = self._held_client_content_deadline
+        return (
+            self._lazy_setup_deadline if content_deadline is None else min(self._lazy_setup_deadline, content_deadline)
+        )
+
+    async def _send_setup_before_content(self, msg_type: str | None) -> None:
+        if self._holds_session_updates_until_first_content():
+            await self._send_held_session_update_as_setup()
+        await self._send_default_setup_before_first_client_message(msg_type)
+
+    async def _receive_client_message_before_setup_deadline(self) -> str | None:
+        if not self._holds_session_updates_until_first_content():
+            return await self._receive_client_message()
+        receive: Final = asyncio.ensure_future(self._receive_client_message())
+        try:
+            done, _ = await asyncio.wait((receive,), timeout=self._setup_deadline() - self._clock())
+            if not done:
+                await self._send_setup_before_content(None)
+            return await receive
+        finally:
+            receive.cancel()
+
+    async def _send_default_setup_before_first_client_message(self, msg_type: str | None) -> None:
+        if self.session_configuration_request is not None or self.provider_config is None:
+            return
+        if msg_type == "session.update" or (
+            self._uses_deferred_backend_setup()
+            and (msg_type in RealTimeStreaming._CLIENT_AUDIO_BUFFER_TYPES or self._pending_messages_until_setup)
+        ):
+            return
+        default_setup: Final = self.provider_config.session_configuration_request(self.model)
+        if not default_setup:
+            return
+        setup: Final = self.maybe_inject_guardrail_auto_response_disable(default_setup)
+        await self.backend_ws.send(setup)
+        self._cache_session_configuration_request(setup)
+        self._applied_session_fields = _NO_SESSION_FIELDS
+        await self._send_held_client_content()
+
     def _enforce_transcription_session_model(self, message: str) -> str:
         """Force client transcription session updates to the authorized model.
 
@@ -851,6 +1102,20 @@ class RealTimeStreaming:
         self._guardrail_cancel_event_ids = self._guardrail_cancel_event_ids | {event_id}
         await self._send_to_backend(json.dumps({"type": "response.cancel", "event_id": event_id}))
 
+    async def _send_blocked_turn_to_backend(self, pre_block_backend_message: str | None, item_id: str | None) -> None:
+        if pre_block_backend_message is None and self._is_transcription_session:
+            return
+        await self._send_setup_before_content("conversation.item.create")
+        # Deliver any caller-supplied backend message FIRST so that
+        # protocol contracts requiring a specific ordering (e.g.
+        # Gemini Live's mandatory ``toolResponse`` after a
+        # ``toolCall``) are honored before the guardrail's own
+        # clientContent / cancel messages are sent.
+        if pre_block_backend_message is not None:
+            await self._send_to_backend(pre_block_backend_message)
+        if not self._is_transcription_session:
+            await self._withdraw_blocked_turn(item_id)
+
     async def _withdraw_blocked_turn(self, item_id: str | None) -> None:
         await self._send_guardrail_cancel()
         if item_id is not None and self._deletes_conversation_items:
@@ -958,6 +1223,21 @@ class RealTimeStreaming:
         """Transcription-only sessions have no assistant turn to gate and reject realtime session updates."""
         return not self._is_transcription_session and self._has_audio_transcription_guardrails()
 
+    def _refuses_unscreened_audio(self, msg_type: str | None) -> bool:
+        return (
+            msg_type in RealTimeStreaming._CLIENT_AUDIO_BUFFER_TYPES
+            and self._builds_setup_lazily()
+            and self._should_disable_vad_auto_response()
+        )
+
+    async def _refuse_unscreened_audio(self, msg_type: str | None, message: str) -> None:
+        if self._unscreened_audio_refused and msg_type not in RealTimeStreaming._CLIENT_AUDIO_BUFFER_COMMIT_TYPES:
+            return
+        self._unscreened_audio_refused = True
+        await self.websocket.send_text(
+            json.dumps(_unscreened_audio_error_event(_decode_json_object(message).get("event_id")))
+        )
+
     async def run_realtime_guardrails(
         self,
         transcript: str,
@@ -1030,15 +1310,7 @@ class RealTimeStreaming:
                 # Use realtime_violation_message if configured; fall back to guardrail error text.
                 error_msg = getattr(callback, "realtime_violation_message", None) or safe_msg
 
-                # Deliver any caller-supplied backend message FIRST so that
-                # protocol contracts requiring a specific ordering (e.g.
-                # Gemini Live's mandatory ``toolResponse`` after a
-                # ``toolCall``) are honored before the guardrail's own
-                # clientContent / cancel messages are sent.
-                if pre_block_backend_message is not None:
-                    await self._send_to_backend(pre_block_backend_message)
-                if not self._is_transcription_session:
-                    await self._withdraw_blocked_turn(item_id)
+                await self._send_blocked_turn_to_backend(pre_block_backend_message, item_id)
                 # Send the policy violation hint (shows as small gray status text in UI).
                 await self.websocket.send_text(
                     json.dumps(
@@ -1136,6 +1408,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
@@ -1295,7 +1570,11 @@ class RealTimeStreaming:
             return BackendClose(code=1011, reason="proxy failed while relaying the upstream websocket")
 
     def _backend_refused_session(self, close: BackendClose) -> bool:
-        return close.code != 1000 and not self._backend_sent_frames
+        return (
+            close.code != 1000
+            and not self._backend_sent_frames
+            and not self._holds_session_updates_until_first_content()
+        )
 
     async def _log_session_end(self, transport_refusal: Exception | None) -> None:
         provider_refusal: Final = None if self.provider_config is None else self.provider_config.session_refusal()
@@ -1488,7 +1767,7 @@ class RealTimeStreaming:
         client_event: _ClientEventFrame
         try:
             while True:
-                message = await self._receive_client_message()
+                message = await self._receive_client_message_before_setup_deadline()
                 if message is None:
                     return ClientLoopExit.CLIENT_DISCONNECTED
 
@@ -1600,6 +1879,7 @@ class RealTimeStreaming:
                     if (
                         msg_type == "session.update"
                         and self.session_configuration_request is None
+                        and self._held_session_update is None
                         and not self._guardrail_turn_detection_update_sent
                         and self._should_disable_vad_auto_response()
                         and (
@@ -1663,12 +1943,25 @@ class RealTimeStreaming:
                 except (json.JSONDecodeError, AttributeError):
                     pass
 
+                if self._refuses_unscreened_audio(msg_type):
+                    await self._refuse_unscreened_audio(msg_type, message)
+                    continue
+
                 ## LOGGING
                 # Log after any in-place modifications (GA remap, guardrail
                 # turn_detection injection) so audit logs reflect what we
                 # actually forward to the backend.
                 self.store_input(message=message)
 
+                if self._holds_client_content(msg_type, message):
+                    self._hold_client_content(message)
+                    continue
+                if self._held_client_content and msg_type != "session.update":
+                    await self._send_setup_before_content(msg_type)
+
+                if msg_type in RealTimeStreaming._CLIENT_CONTENT_TYPES and self._uses_deferred_backend_setup():
+                    await self._send_setup_before_content(msg_type)
+
                 if self._should_buffer_client_message_until_setup(message):
                     self._buffer_pending_message_until_setup(message)
                     continue
@@ -1697,6 +1990,15 @@ 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 msg_type == "session.update" and self._holds_session_updates_until_first_content():
+                    await self._hold_session_update(message)
+                    if self._pending_messages_until_setup:
+                        await self._send_held_session_update_as_setup()
+                    continue
+                if msg_type in RealTimeStreaming._CLIENT_CONTENT_TYPES:
+                    await self._send_setup_before_content(msg_type)
                 try:
                     sent = await self._send_to_backend(message)
                 except RealtimeTranscriptionProtocolError as e:

litellm/llms/base_llm/realtime/transformation.py

@@ -1,10 +1,12 @@
+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 litellm._uuid import uuid
 from litellm.types.llms.openai import OpenAIRealtimeStreamSessionEvents
 from litellm.types.realtime import (
     RealtimeInputAudioTranscriptionUsage,
@@ -22,6 +24,26 @@ else:
     LiteLLMLoggingObj = Any
 
 
+@dataclasses.dataclass(frozen=True, slots=True)
+class RealtimeSessionUpdateRejection:
+    code: Literal["invalid_type", "invalid_value", "session_update_not_applied", "session_update_too_large"]
+    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) -> object: ...
 
@@ -93,6 +115,14 @@ class BaseRealtimeConfig(ABC):
     def session_configuration_request(self, model: str) -> str | None:  # message sent to setup the realtime session
         return None
 
+    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
@@ -196,13 +196,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."""
@@ -533,7 +526,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):
@@ -541,7 +534,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. "
@@ -551,7 +544,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. "
@@ -562,7 +555,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/custom_httpx/llm_http_handler.py

@@ -6973,14 +6973,19 @@ class BaseLLMHTTPHandler:
                     ),
                 )
 
-                # Auto-send session setup if the provider requires it (e.g.
-                # Gemini/Vertex AI Live needs a `setup` before any realtime_input).
+                # Auto-send session setup at connect unless the provider builds it from the
+                # client's first session.update (Gemini/Vertex AI Live accept exactly one
+                # setup, so RealTimeStreaming sends theirs on the first client message).
                 # 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: str | None = None
-                if provider_config.requires_session_configuration():
+                sends_setup_at_connect: Final = (
+                    provider_config.requires_session_configuration()
+                    and not provider_config.builds_setup_from_session_update()
+                )
+                if sends_setup_at_connect:
                     _session_config = provider_config.session_configuration_request(model)
                     if _session_config:
                         _session_config = realtime_streaming.maybe_inject_guardrail_auto_response_disable(

litellm/llms/gemini/realtime/transformation.py

@@ -7,6 +7,7 @@ from collections import OrderedDict
 from collections.abc import Mapping, Sequence
 from typing import Final, cast
 
+from pydantic import TypeAdapter
 from typing_extensions import ReadOnly, Required, TypedDict
 
 import litellm
@@ -15,8 +16,9 @@ from litellm._uuid import uuid
 from litellm.litellm_core_utils.core_helpers import as_str_mapping
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
 from litellm.litellm_core_utils.url_utils import join_api_base
-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 (
+    OPENAI_TO_GEMINI_TTS_VOICES,
     VertexGeminiConfig,
 )
 from litellm.responses.litellm_completion_transformation.transformation import (
@@ -76,27 +78,82 @@ MAP_GEMINI_FIELD_TO_OPENAI_EVENT: Final[dict[str, OpenAIRealtimeEventTypes | Res
 
 # Keys the main transform loop handles; siblings like ``usageMetadata`` are skipped.
 _KNOWN_GEMINI_TOP_LEVEL_KEYS: Final[set] = {map_key.split(".", 1)[0] for map_key in MAP_GEMINI_FIELD_TO_OPENAI_EVENT}
+_GEMINI_LIVE_RESPONSE_MODALITIES: Final = frozenset({"TEXT", "AUDIO"})
 
 
 OPENAI_STOCK_REALTIME_VOICES: Final[frozenset[str]] = frozenset(
     {"alloy", "ash", "ballad", "cedar", "coral", "echo", "marin", "sage", "shimmer", "verse"}
 )
 
+# Per https://docs.cloud.google.com/vertex-ai/generative-ai/docs/live-api/configure-language-voice (2026-10-28)
+GEMINI_LIVE_DOCUMENTED_VOICES: Final[frozenset[str]] = frozenset(
+    {
+        "achernar",
+        "achird",
+        "algenib",
+        "algieba",
+        "alnilam",
+        "aoede",
+        "autonoe",
+        "callirrhoe",
+        "charon",
+        "despina",
+        "enceladus",
+        "erinome",
+        "fenrir",
+        "gacrux",
+        "iapetus",
+        "kore",
+        "laomedeia",
+        "leda",
+        "orus",
+        "puck",
+        "pulcherrima",
+        "rasalgethi",
+        "sadachbia",
+        "sadaltager",
+        "schedar",
+        "sulafat",
+        "umbriel",
+        "vindemiatrix",
+        "zephyr",
+        "zubenelgenubi",
+    }
+)
 
-def _gemini_live_speech_config(voice: object) -> Mapping[str, object] | None:
+
+def _is_undocumented_live_voice(voice: object) -> bool:
+    return (
+        isinstance(voice, str)
+        and voice.lower() not in OPENAI_STOCK_REALTIME_VOICES
+        and OPENAI_TO_GEMINI_TTS_VOICES.get(voice.lower(), voice).lower() not in GEMINI_LIVE_DOCUMENTED_VOICES
+    )
+
+
+def _gemini_live_speech_config(voice: object, drop_undocumented: bool) -> Mapping[str, object] | None:
     """Build the Gemini Live speechConfig for a client-requested voice.
 
-    OpenAI stock voice names have no Gemini equivalent and Gemini Live closes
-    the session on an unknown voice, so they are dropped with a warning and
-    the model keeps its default voice. Every other name is forwarded verbatim.
+    OpenAI stock voice names have no Gemini equivalent, so they are always
+    dropped, and so is any name outside Google's documented Live voices when
+    ``drop_undocumented`` is set, keeping the model's default voice. Every
+    other name is forwarded verbatim.
     """
-    if isinstance(voice, str) and voice.lower() in OPENAI_STOCK_REALTIME_VOICES:
+    if not isinstance(voice, str):
+        return VertexGeminiConfig().map_audio_params({"voice": voice})
+    if voice.lower() in OPENAI_STOCK_REALTIME_VOICES:
         verbose_logger.warning(
             "Gemini Realtime: voice %s is an OpenAI voice with no Gemini equivalent; "
             "dropping it so the session keeps the model's default voice.",
             voice,
         )
         return None
+    if drop_undocumented and _is_undocumented_live_voice(voice):
+        verbose_logger.warning(
+            "Gemini Realtime: voice %s is not a documented Gemini Live voice; "
+            "drop_params is set, so it is dropped and the session keeps the model's default voice.",
+            voice,
+        )
+        return None
     return VertexGeminiConfig().map_audio_params({"voice": voice})
 
 
@@ -142,16 +199,34 @@ def _base64_decoded_byte_count(data: str) -> int:
     return max(len(data) * 3 // 4 - padding, 0)
 
 
+_JSON_ARRAY: Final = TypeAdapter(tuple[object, ...])
+
+
+def _chat_completion_tool(tool: object) -> dict[str, object]:
+    fields: Final = as_str_mapping(tool) or {}
+    if fields.get("type") != "function" or "function" in fields:
+        return {**fields}
+    return {"type": "function", "function": {key: value for key, value in fields.items() if key != "type"}}
+
+
+def _system_instruction_text(system_instruction: HttpxContentType | None) -> str | None:
+    parts: Final = system_instruction.get("parts", []) if system_instruction else []
+    texts: Final = tuple(text for part in parts if isinstance(text := part.get("text"), str))
+    return "\n".join(texts) if texts else None
+
+
 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.
         self._pending_usage_metadata: dict | None = None
         self._unbilled_input_audio_bytes: int = 0
+        self._manual_activity_open: bool = False
 
     def is_setup_message(self, msg_obj: dict) -> bool:
         return "setup" in msg_obj
@@ -167,6 +242,10 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
         """Google AI Studio closes the socket (1007) on padding or silence tuning next to ``disabled: true``."""
         return True
 
+    def _rejects_undocumented_voices(self, model: str) -> bool:
+        """Google AI Studio closes the socket (1007) on an unknown voice only for its native-audio Live models."""
+        return bool(GeminiRealtimeConfig._model_cost_entry(model).get("gemini_native_audio"))
+
     @staticmethod
     def _usage_detail_alias(details: object, defaults: dict[str, int]) -> dict[str, object]:
         detail_mapping: Final = as_str_mapping(details)
@@ -272,17 +351,26 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
         except (json.JSONDecodeError, TypeError, AttributeError):
             return False
 
+    def _start_manual_activity(self, session_configuration_request: str | None) -> tuple[str, ...]:
+        if self._manual_activity_open or not self._manual_turn_detection_enabled(session_configuration_request):
+            return ()
+        self._manual_activity_open = True
+        activity_start: Final[BidiGenerateContentRealtimeInput] = {"activityStart": {}}
+        return (json.dumps({"realtimeInput": activity_start}),)
+
+    def _end_manual_activity(self) -> tuple[str, ...]:
+        if not self._manual_activity_open:
+            return ()
+        self._manual_activity_open = False
+        activity_end: Final[BidiGenerateContentRealtimeInput] = {"activityEnd": {}}
+        return (json.dumps({"realtimeInput": activity_end}),)
+
     def _handle_input_audio_buffer_commit_or_end(self, session_configuration_request: str | None) -> list[str]:
         """Map OpenAI buffer commit/end to Gemini Live turn-boundary signals."""
         if self._manual_turn_detection_enabled(session_configuration_request):
-            realtime_input_dict: BidiGenerateContentRealtimeInput = {
-                "activityEnd": True,
-            }
-            verbose_logger.debug("Gemini Realtime: Sending activityEnd realtimeInput to backend")
-        else:
-            realtime_input_dict = {"audioStreamEnd": True}
-            verbose_logger.debug("Gemini Realtime: Sending audioStreamEnd realtimeInput to backend")
-        return [json.dumps({"realtimeInput": realtime_input_dict})]
+            return list(self._end_manual_activity())
+        audio_stream_end: Final[BidiGenerateContentRealtimeInput] = {"audioStreamEnd": True}
+        return [json.dumps({"realtimeInput": audio_stream_end})]
 
     def map_automatic_turn_detection(self, value: OpenAIRealtimeTurnDetection) -> AutomaticActivityDetection:
         """Map OpenAI ``server_vad`` to Gemini ``automaticActivityDetection``.
@@ -325,7 +413,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
             "voice",
         ]
 
-    def map_openai_params(self, optional_params: dict, non_default_params: dict) -> dict:
+    def map_openai_params(self, optional_params: dict, non_default_params: dict, *, model: str) -> dict:
         if "generationConfig" not in optional_params:
             optional_params["generationConfig"] = {}
         for key, value in non_default_params.items():
@@ -333,12 +421,12 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
                 optional_params["systemInstruction"] = HttpxContentType(role="user", parts=[{"text": value}])
             elif key == "temperature":
                 optional_params["generationConfig"]["temperature"] = value
-            elif key == "max_response_output_tokens":
+            elif key == "max_response_output_tokens" and isinstance(value, int):
                 optional_params["generationConfig"]["maxOutputTokens"] = value
             elif key == "modalities":
-                optional_params["generationConfig"]["responseModalities"] = [
-                    modality.upper() for modality in cast(list[str], value)
-                ]
+                optional_params["generationConfig"]["responseModalities"] = list(
+                    GeminiRealtimeConfig._requested_response_modalities(cast(object, value))
+                )
             elif key == "tools":
                 from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
                     VertexGeminiConfig,
@@ -347,34 +435,43 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
                 vertex_gemini_config = VertexGeminiConfig()
                 # Tools should be at the top level of setup, not inside generationConfig
                 optional_params["tools"] = vertex_gemini_config.map_function(
-                    value=value, optional_params=optional_params
+                    value=[_chat_completion_tool(tool) for tool in _JSON_ARRAY.validate_python(value)],
+                    optional_params=optional_params,
                 )
             elif key == "input_audio_transcription" and value is not None:
                 optional_params["inputAudioTranscription"] = {}
-            elif key == "turn_detection" and value is not None:
-                value_typed = cast(OpenAIRealtimeTurnDetection, value)
-                if (
-                    isinstance(value_typed, dict)
-                    and value_typed.get("type") == "semantic_vad"
-                    and "create_response" not in value_typed
-                ):
-                    # Pipecat/OpenAI GA semantic VAD — skip; Gemini uses its own VAD.
-                    # Only skip when there is no create_response override so that
-                    # a guardrail-injected create_response:false is not dropped.
-                    continue
-                transformed_audio_activity_config = self.map_automatic_turn_detection(value_typed)
-                if transformed_audio_activity_config:
-                    optional_params["realtimeInputConfig"] = BidiGenerateContentRealtimeInputConfig(
-                        automaticActivityDetection=transformed_audio_activity_config
-                    )
+            elif key == "turn_detection":
+                realtime_input_config = self._realtime_input_config(cast(object, value))
+                if realtime_input_config is not None:
+                    optional_params["realtimeInputConfig"] = realtime_input_config
             elif key == "voice":
-                speech_config = _gemini_live_speech_config(value)
+                speech_config = _gemini_live_speech_config(
+                    value, self.drop_params and self._rejects_undocumented_voices(model)
+                )
                 if speech_config:
                     optional_params["generationConfig"]["speechConfig"] = speech_config
         if len(optional_params["generationConfig"]) == 0:
             optional_params.pop("generationConfig")
         return optional_params
 
+    def _realtime_input_config(self, turn_detection: object) -> BidiGenerateContentRealtimeInputConfig | None:
+        if turn_detection is None:
+            return BidiGenerateContentRealtimeInputConfig(automaticActivityDetection={"disabled": True})
+        value_typed: Final = cast(OpenAIRealtimeTurnDetection, turn_detection)
+        if (
+            isinstance(value_typed, dict)
+            and value_typed.get("type") == "semantic_vad"
+            and "create_response" not in value_typed
+        ):
+            # Pipecat/OpenAI GA semantic VAD — skip; Gemini uses its own VAD.
+            # Only skip when there is no create_response override so that
+            # a guardrail-injected create_response:false is not dropped.
+            return None
+        transformed_audio_activity_config: Final = self.map_automatic_turn_detection(value_typed)
+        if not transformed_audio_activity_config:
+            return None
+        return BidiGenerateContentRealtimeInputConfig(automaticActivityDetection=transformed_audio_activity_config)
+
     @staticmethod
     def _extract_turn_detection(session: dict) -> dict | None:
         """Extract turn_detection from a session.update payload.
@@ -423,6 +520,8 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
             if isinstance(input_cfg, dict):
                 if "input_audio_transcription" not in normalized and "transcription" in input_cfg:
                     normalized["input_audio_transcription"] = input_cfg["transcription"]
+                if "turn_detection" not in normalized and "turn_detection" in input_cfg:
+                    normalized["turn_detection"] = input_cfg["turn_detection"]
             output_cfg: Final = audio.get("output")
             if isinstance(output_cfg, dict) and output_cfg.get("voice"):
                 normalized["voice"] = output_cfg["voice"]
@@ -434,7 +533,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
         return normalized
 
     @staticmethod
-    def _model_cost_entry(model: str) -> dict:
+    def _model_cost_entry(model: str) -> Mapping[str, object]:
         entry = litellm.model_cost.get(model)
         if entry is None:
             stripped: Final = model.split("/", 1)[-1]
@@ -455,18 +554,122 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
         return "TEXT" if GeminiRealtimeConfig._is_text_only_live_model(model) else "AUDIO"
 
     @staticmethod
-    def _coerce_response_modalities(model: str, modalities: Sequence[object]) -> tuple[str, ...]:
-        """Swap responseModalities a Live model cannot produce: TEXT to AUDIO for
-        audio-only models, AUDIO to TEXT for text-only ones (e.g. transcribe-live)."""
-        normalized: Final = tuple(
-            modality.upper() if isinstance(modality, str) else str(modality).upper() for modality in modalities
+    def _requested_response_modalities(value: object) -> tuple[str, ...]:
+        if isinstance(value, str):
+            return (value.upper(),)
+        if not isinstance(value, (list, tuple)):
+            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 frozenset(requested) == frozenset(("AUDIO",)):
+            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 accepts one session configuration per connection, and this session already sent "
+                    "its configuration, so this session.update was not applied. Put every setting in session.update "
+                    "before the first conversation content, or set drop_params to drop it instead"
+                ),
+                param="session",
+            )
+        return self._modality_rejection(model, session) or self._voice_rejection(model, session)
+
+    @staticmethod
+    def _modality_rejection(model: str, session: Mapping[str, object]) -> RealtimeSessionUpdateRejection | None:
+        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}",
+        )
+
+    def _voice_rejection(self, model: str, session: Mapping[str, object]) -> RealtimeSessionUpdateRejection | None:
+        audio_output: Final = as_str_mapping((as_str_mapping(session.get("audio")) or {}).get("output")) or {}
+        output_voice: Final = audio_output.get("voice")
+        voice: Final = output_voice or session.get("voice")
+        if not (self._rejects_undocumented_voices(model) and _is_undocumented_live_voice(voice)):
+            return None
+        param: Final = "session.audio.output.voice" if output_voice else "session.voice"
+        return RealtimeSessionUpdateRejection(
+            code="invalid_value",
+            message=(
+                f"Invalid value for '{param}': {voice} is not a Gemini Live voice {model} offers. "
+                "Set drop_params to drop it instead"
+            ),
+            param=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
+        model cannot (TEXT to AUDIO for audio-only models, AUDIO to TEXT for
+        text-only ones such as transcribe-live), then keep a single modality
+        because Gemini Live rejects a setup naming two."""
+        return GeminiRealtimeConfig._single_response_modality(
+            GeminiRealtimeConfig._swap_unsupported_response_modalities(
+                model, GeminiRealtimeConfig._known_response_modalities(model, modalities)
+            )
+        )
+
+    @staticmethod
+    def _known_response_modalities(model: str, modalities: Sequence[object]) -> tuple[str, ...]:
+        normalized: Final = tuple(str(modality).upper() for modality in modalities)
+        known: Final = tuple(modality for modality in normalized if modality in _GEMINI_LIVE_RESPONSE_MODALITIES)
+        if len(known) < len(normalized):
+            verbose_logger.warning(
+                "Gemini Live: %s only produces TEXT or AUDIO; ignoring the requested %s",
+                model,
+                ", ".join(modality for modality in normalized if modality not in _GEMINI_LIVE_RESPONSE_MODALITIES),
+            )
+        return known or (GeminiRealtimeConfig._default_response_modality(model),)
+
+    @staticmethod
+    def _swap_unsupported_response_modalities(model: str, normalized: tuple[str, ...]) -> tuple[str, ...]:
         if GeminiRealtimeConfig._is_audio_only_live_model(model) and "TEXT" in normalized:
+            verbose_logger.warning(
+                "Gemini Live: %s only produces audio; downgrading the requested TEXT response modality to AUDIO",
+                model,
+            )
             return tuple(modality for modality in normalized if modality != "TEXT") or ("AUDIO",)
         if GeminiRealtimeConfig._is_text_only_live_model(model) and "AUDIO" in normalized:
             return tuple(modality for modality in normalized if modality != "AUDIO") or ("TEXT",)
         return normalized
 
+    @staticmethod
+    def _single_response_modality(modalities: tuple[str, ...]) -> tuple[str, ...]:
+        if len(modalities) <= 1:
+            return modalities
+        chosen: Final = "AUDIO" if "AUDIO" in modalities else modalities[0]
+        verbose_logger.warning(
+            "Gemini Live: at most one response modality is allowed per session; using %s for the requested %s",
+            chosen,
+            ", ".join(modalities),
+        )
+        return (chosen,)
+
     @staticmethod
     def _finalize_gemini_live_setup(model: str, setup: dict[str, object]) -> dict[str, object]:
         generation_config: Final = setup.get("generationConfig")
@@ -504,7 +707,9 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
         # would be silently dropped because ``map_openai_params`` only
         # recognises the flat OpenAI-beta key names.
         session_payload = self._normalize_session_payload_for_mapping(session_payload)
-        new_overrides: Final = self.map_openai_params(optional_params={}, non_default_params=session_payload)
+        new_overrides: Final = self.map_openai_params(
+            optional_params={}, non_default_params=session_payload, model=model
+        )
 
         if session_configuration_request is None:
             generation_config: Final = new_overrides.setdefault("generationConfig", {})
@@ -618,14 +823,13 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
                 message_str = str(message)
             raise ValueError(f"Invalid JSON message: {message_str}")
 
-        messages: Final[list[str]] = []
         msg_type: Final = json_message.get("type")
 
         if msg_type == "session.update":
             return self._handle_session_update(json_message, model, session_configuration_request)
 
         if msg_type == "response.create":
-            return []  # Gemini responds automatically; nothing to forward
+            return list(self._end_manual_activity())
 
         if msg_type == "conversation.item.create":
             return self._handle_conversation_item(json_message)
@@ -643,8 +847,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
 
             gemini_msg: Final = json.dumps({"realtimeInput": realtime_input_dict})
             verbose_logger.debug("Gemini Realtime: Sending audio realtimeInput to backend")
-            messages.append(gemini_msg)
-            return messages
+            return [*self._start_manual_activity(session_configuration_request), gemini_msg]
 
         if msg_type in ("input_audio_buffer.commit", "input_audio_buffer.end"):
             return self._handle_input_audio_buffer_commit_or_end(session_configuration_request)
@@ -660,22 +863,21 @@ 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 {}
         gemini_modalities: Final = generation_config.get("responseModalities", ["AUDIO"])
         _modalities: Final = [modality.lower() for modality in cast(list[str], gemini_modalities)]
-        _system_instruction: Final = session_configuration_request_dict.get("systemInstruction")
+        instructions: Final = _system_instruction_text(session_configuration_request_dict.get("systemInstruction"))
         session: Final = OpenAIRealtimeStreamSession(
             id=logging_session_id,
             modalities=_modalities,
         )
-        if _system_instruction is not None and isinstance(_system_instruction, str):
-            session["instructions"] = _system_instruction
+        if instructions is not None:
+            session["instructions"] = instructions
         if _model is not None and isinstance(_model, str):
             # Strip Vertex/AI Studio path prefixes to expose the bare model name.
             if "/models/" in _model:
@@ -1662,6 +1864,9 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
         # Deferred setup opt-in: litellm.gemini_live_defer_setup = True
         return not litellm.gemini_live_defer_setup
 
+    def builds_setup_from_session_update(self) -> bool:
+        return True
+
     def session_configuration_request(self, model: str) -> str:
         """
 

litellm/llms/vertex_ai/realtime/transformation.py

@@ -39,8 +39,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
@@ -51,6 +51,9 @@ class VertexAIRealtimeConfig(GeminiRealtimeConfig):
     def _rejects_vad_tuning_with_detection_off(self) -> bool:
         return False
 
+    def _rejects_undocumented_voices(self, model: str) -> bool:
+        return True
+
     # ------------------------------------------------------------------
     # URL
     # ------------------------------------------------------------------
@@ -163,7 +166,7 @@ class VertexAIRealtimeConfig(GeminiRealtimeConfig):
         # settings would be silently dropped because ``map_openai_params`` only
         # recognises the flat OpenAI-beta key names.
         session_params = self._normalize_session_payload_for_mapping(session_params)
-        setup_config: Final = self.map_openai_params(optional_params={}, non_default_params=session_params)
+        setup_config: Final = self.map_openai_params(optional_params={}, non_default_params=session_params, model=model)
 
         # Use full Vertex AI model path
         setup_config["model"] = self._vertex_model_path(model)
@@ -177,15 +180,16 @@ class VertexAIRealtimeConfig(GeminiRealtimeConfig):
         # ``silence_duration_ms``). ``map_automatic_turn_detection`` sets
         # ``disabled=True`` whenever ``create_response`` is absent or
         # ``False``. Force ``disabled=False`` only when the client did
-        # not explicitly request ``create_response: False`` — that path
+        # not explicitly request ``create_response: False`` or
+        # ``turn_detection: null`` — the first of those paths
         # is how transcription guardrails suppress automatic responses,
         # and overriding it here would silently bypass the guardrail.
         # Vertex Live has no "VAD on, no auto-response" mode, so callers
         # that need that behaviour must accept that VAD is off.
         client_turn_detection: Final = session_params.get("turn_detection")
         client_disabled_auto_response: Final = (
-            isinstance(client_turn_detection, dict) and client_turn_detection.get("create_response") is False
-        )
+            "turn_detection" in session_params and client_turn_detection is None
+        ) or (isinstance(client_turn_detection, dict) and client_turn_detection.get("create_response") is False)
         realtime_input_config: Final = setup_config.setdefault("realtimeInputConfig", {})
         automatic_detection: Final = realtime_input_config.setdefault("automaticActivityDetection", {})
         if not client_disabled_auto_response:
@@ -261,4 +265,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/types/llms/gemini.py

@@ -81,6 +81,14 @@ class BidiGenerateContentServerMessage(TypedDict, total=False):
     """Output only. The setup complete message."""
 
 
+class BidiGenerateContentActivityStart(TypedDict):
+    pass
+
+
+class BidiGenerateContentActivityEnd(TypedDict):
+    pass
+
+
 class BidiGenerateContentRealtimeInput(TypedDict, total=False):
     text: str
     """The text to be sent to the model."""
@@ -94,11 +102,11 @@ class BidiGenerateContentRealtimeInput(TypedDict, total=False):
     audioStreamEnd: bool
     """Output only. If true, indicates that the audio stream has ended."""
 
-    activityStart: bool
-    """Output only. If true, indicates that the activity has started."""
+    activityStart: BidiGenerateContentActivityStart
+    """Marks the start of user activity when automatic activity detection is disabled."""
 
-    activityEnd: bool
-    """Output only. If true, indicates that the activity has ended."""
+    activityEnd: BidiGenerateContentActivityEnd
+    """Marks the end of user activity when automatic activity detection is disabled."""
 
 
 StartOfSpeechSensitivityEnum = Literal[

litellm/utils.py

@@ -10911,7 +10911,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/benchmarks/test_realtime_session_merge_benchmarks.py

@@ -0,0 +1,21 @@
+from typing import Final
+
+import pytest
+
+from litellm.litellm_core_utils.realtime_streaming import _merge_session_fields
+
+NESTED_LEVELS: Final = 8
+LEAVES: Final = 2000
+
+
+def _nested_session(depth: int) -> dict[str, object]:
+    return {"child": _nested_session(depth - 1)} if depth else {f"k{index}": index for index in range(LEAVES)}
+
+
+@pytest.mark.benchmark
+def test_merging_a_session_update_that_repeats_every_nested_level() -> None:
+    session: Final = _nested_session(NESTED_LEVELS)
+
+    merged: Final = _merge_session_fields(session, session)
+
+    assert merged == session

tests/code_coverage_tests/recursive_detector.py

@@ -85,6 +85,7 @@ IGNORE_FUNCTIONS = [
     "_sort_processed_sets",  # bounded by the nesting depth of the log-record extra it walks (a finite JSON tree, no cycles possible).
     "scrub_json_strings",  # max depth set (MAX_SCRUB_DEPTH); fails closed by returning "[Filtered]" for anything nested past the cap.
     "_without_requester_ip_address",  # walks only the fresh tree _sanitize_request_body_for_spend_logs_payload returns (mapping cycles already cut), so it never goes deeper than that sanitizer did.
+    "_merge_session_fields",  # max depth set (_SESSION_MERGE_MAX_DEPTH); deeper mappings are replaced wholesale by the later session.update.
 ]
 
 

tests/e2e/llm_translation/realtime/realtime_client.py

@@ -47,6 +47,7 @@ class RealtimeProvider:
     id: str
     alias: str
     litellm_params: LiteLLMParamsBody
+    modalities: tuple[str, ...] = ("text",)
 
 
 PROVIDERS = (
@@ -75,6 +76,7 @@ PROVIDERS = (
             model="gemini/gemini-3.1-flash-live-preview",
             api_key="os.environ/GEMINI_API_KEY",
         ),
+        modalities=("audio",),
     ),
     RealtimeProvider(
         "vertex_ai",
@@ -84,6 +86,7 @@ PROVIDERS = (
             vertex_location="us-central1",
             vertex_credentials="os.environ/VERTEXAI_CREDENTIALS",
         ),
+        modalities=("audio",),
     ),
     RealtimeProvider(
         "bedrock",

tests/e2e/llm_translation/realtime/test_realtime_e2e.py

@@ -124,7 +124,8 @@ def test_text_conversation(
         session.send(
             SessionUpdate(
                 session=SessionConfig(
-                    instructions="You are a terse assistant. Reply in one short sentence."
+                    modalities=list(provider.modalities),
+                    instructions="You are a terse assistant. Reply in one short sentence.",
                 )
             )
         )
@@ -159,6 +160,7 @@ def test_tool_call_round_trip(
         session.send(
             SessionUpdate(
                 session=SessionConfig(
+                    modalities=list(provider.modalities),
                     tools=[WEATHER_TOOL],
                     tool_choice="auto",
                     instructions=(

tests/integration/providers/test_realtime_transcription_guardrail_session_update.py

@@ -355,7 +355,13 @@ def _voice_scenario_created_after_the_first_frame(transcript: str) -> RealtimeRe
 
 
 def _live_scenario() -> RealtimeResponse:
-    return RealtimeResponse(content_type="application/x-realtime", events=())
+    return RealtimeResponse(
+        content_type="application/x-realtime",
+        events=(
+            {"serverContent": {"modelTurn": {"parts": [{"text": "scripted $REQUEST_ID"}]}}},
+            {"serverContent": {"turnComplete": True}},
+        ),
+    )
 
 
 def _realtime_update(turn_detection: JsonValue) -> dict[str, JsonValue]:
@@ -1348,10 +1354,14 @@ def test_gemini_ai_studio_voice_session_with_vad_tuning_connects_under_the_guard
             _ws_base(_owned_url(guardrail_proxy)),
             f"model={model}",
             key,
-            (_step(_update_frame(update), until="session.updated"),),
+            (
+                _step(_update_frame(update), until="session.updated"),
+                _step(_user_text_item(CLEAN_TRANSCRIPT), RESPONSE_CREATE, until=RESPONSE_DONE),
+            ),
         )
         observed: Final = _observed(upstream_url, handle.scenario_id)
-        assert session.types == ("session.created", "session.updated"), session
+        assert session.types[:2] == ("session.created", "session.updated"), session
+        assert session.types[-1] == RESPONSE_DONE, session
         assert _setup_detection(observed) == {"disabled": True}, _sent(observed)
 
 
@@ -1365,10 +1375,14 @@ def test_vertex_live_voice_session_keeps_vad_tuning_under_the_guardrail(guardrai
             _ws_base(_owned_url(guardrail_proxy)),
             f"model={model}",
             key,
-            (_step(_update_frame(_realtime_update(TUNED_SERVER_VAD)), until="session.updated"),),
+            (
+                _step(_update_frame(_realtime_update(TUNED_SERVER_VAD)), until="session.updated"),
+                _step(_user_text_item(CLEAN_TRANSCRIPT), RESPONSE_CREATE, until=RESPONSE_DONE),
+            ),
         )
         observed: Final = _observed(upstream_url, handle.scenario_id)
-        assert session.types == ("session.created", "session.updated"), session
+        assert session.types[:2] == ("session.created", "session.updated"), session
+        assert session.types[-1] == RESPONSE_DONE, session
         assert _setup_detection(observed) == GEMINI_TUNED_DETECTION, _sent(observed)
 
 

tests/unit/litellm_core_utils/test_realtime_streaming.py

@@ -1,7 +1,7 @@
 import asyncio
 import base64
 import json
-from collections.abc import Coroutine, Mapping, Sequence
+from collections.abc import Callable, Coroutine, Mapping, Sequence
 from dataclasses import dataclass
 from typing import Final, Literal
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -14,6 +14,7 @@ from websockets.frames import Close
 import litellm
 from litellm.constants import (
     BEDROCK_REALTIME_PENDING_OPENING_MESSAGE_SCOPE_KEY,
+    REALTIME_LAZY_SETUP_DEADLINE_SECONDS,
     REALTIME_METERED_USAGE_KEY,
     REALTIME_SESSION_FAILURE_LOGGED_KEY,
     REALTIME_SESSION_SUCCESS_LOGGED_KEY,
@@ -21,6 +22,7 @@ from litellm.constants import (
 from litellm.integrations.custom_guardrail import CustomGuardrail
 from litellm.litellm_core_utils.realtime_streaming import (
     RealTimeStreaming,
+    _merge_session_fields,
     client_sent_openai_beta_realtime_header,
 )
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
@@ -1255,11 +1257,15 @@ async def test_client_ack_caches_setup_to_prevent_duplicate_session_update_setup
         side_effect=[
             json.dumps({"type": "session.update", "session": {"tools": []}}),
             json.dumps({"type": "session.update", "session": {"tools": []}}),
+            _TEXT_ITEM_CREATE,
+            _RESPONSE_CREATE,
             Exception("client done"),
         ]
     )
+    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:
@@ -2285,6 +2291,7 @@ async def test_guardrail_turn_detection_injected_into_first_session_update_defer
     logging_obj.success_handler = MagicMock()
 
     provider_config = MagicMock()
+    provider_config.builds_setup_from_session_update = MagicMock(return_value=False)
     transformed_messages = []
 
     def mock_transform(msg, model, session_config):
@@ -2351,6 +2358,7 @@ async def test_guardrail_turn_detection_injection_tolerates_non_dict_value(
     logging_obj.success_handler = MagicMock()
 
     provider_config = MagicMock()
+    provider_config.builds_setup_from_session_update = MagicMock(return_value=False)
     transformed_messages = []
 
     def mock_transform(msg, model, session_config):
@@ -2413,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):
@@ -2465,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": {
@@ -2498,6 +2508,636 @@ 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: pytest.MonkeyPatch,
+    client_messages: Sequence[str],
+    model: str = "gemini-live-2.5-flash-native-audio",
+    drop_params: bool = False,
+    defer_setup: bool = False,
+) -> tuple[RealTimeStreaming, MagicMock]:
+    monkeypatch.setattr(litellm, "gemini_live_defer_setup", defer_setup, raising=False)
+    from litellm.llms.gemini.realtime.transformation import GeminiRealtimeConfig
+
+    client_ws = MagicMock()
+    client_ws.receive_text = AsyncMock(side_effect=[*client_messages, ConnectionClosed(None, None)])
+    client_ws.send_text = AsyncMock()
+    backend_ws = MagicMock()
+    backend_ws.send = AsyncMock()
+    streaming = RealTimeStreaming(
+        client_ws,
+        backend_ws,
+        MagicMock(),
+        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",
+        "item": {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "hi"}]},
+    }
+)
+
+_RESPONSE_CREATE = json.dumps({"type": "response.create"})
+
+
+@pytest.mark.asyncio
+async def test_default_mode_sends_default_setup_before_first_non_session_client_message(monkeypatch):
+    """Gemini Live setup is no longer pre-sent at connect, so a client that never
+    sends session.update still gets the provider default setup ahead of its
+    first content frame."""
+    streaming, backend_ws = _gemini_default_mode_streaming(monkeypatch, [_TEXT_ITEM_CREATE])
+
+    await streaming.client_ack_messages()
+
+    frames = [json.loads(call.args[0]) for call in backend_ws.send.await_args_list]
+    assert [next(iter(frame)) for frame in frames] == ["setup", "clientContent"]
+    assert frames[0]["setup"]["generationConfig"]["responseModalities"] == ["AUDIO"]
+    assert streaming.session_configuration_request == backend_ws.send.await_args_list[0].args[0]
+
+
+@pytest.mark.asyncio
+async def test_default_mode_session_update_first_builds_the_only_setup(monkeypatch):
+    session_update = json.dumps({"type": "session.update", "session": {"modalities": ["text"]}})
+    streaming, backend_ws = _gemini_default_mode_streaming(
+        monkeypatch, [session_update, _TEXT_ITEM_CREATE, _RESPONSE_CREATE], model="gemini-live-2.5-flash"
+    )
+
+    await streaming.client_ack_messages()
+
+    frames = [json.loads(call.args[0]) for call in backend_ws.send.await_args_list]
+    assert [next(iter(frame)) for frame in frames] == ["setup", "clientContent"]
+    assert frames[0]["setup"]["generationConfig"]["responseModalities"] == ["TEXT"]
+
+
+@pytest.mark.asyncio
+async def test_default_mode_merges_every_session_update_sent_before_the_first_content_frame(monkeypatch):
+    """Clients configure a session over several session.update messages. Gemini
+    Live takes one setup, so every update sent before the first content frame is
+    merged into that setup instead of only the first one applying."""
+    instructions_update = json.dumps({"type": "session.update", "session": {"instructions": "Be brief."}})
+    modalities_update = json.dumps({"type": "session.update", "session": {"modalities": ["text"]}})
+    streaming, backend_ws = _gemini_default_mode_streaming(
+        monkeypatch, [instructions_update, modalities_update, _TEXT_ITEM_CREATE, _RESPONSE_CREATE], model="gemini-live-2.5-flash"
+    )
+
+    await streaming.client_ack_messages()
+
+    frames = [json.loads(call.args[0]) for call in backend_ws.send.await_args_list]
+    assert [next(iter(frame)) for frame in frames] == ["setup", "clientContent"]
+    assert frames[0]["setup"]["generationConfig"]["responseModalities"] == ["TEXT"]
+    assert frames[0]["setup"]["systemInstruction"]["parts"][0]["text"] == "Be brief."
+
+
+@pytest.mark.asyncio
+async def test_default_mode_merges_nested_audio_settings_across_session_updates(monkeypatch):
+    vad_update = json.dumps(
+        {"type": "session.update", "session": {"audio": {"input": {"turn_detection": {"silence_duration_ms": 1500}}}}}
+    )
+    voice_update = json.dumps({"type": "session.update", "session": {"audio": {"output": {"voice": "Kore"}}}})
+    streaming, backend_ws = _gemini_default_mode_streaming(
+        monkeypatch, [vad_update, voice_update, _TEXT_ITEM_CREATE, _RESPONSE_CREATE], model="gemini-live-2.5-flash"
+    )
+
+    await streaming.client_ack_messages()
+
+    setup = json.loads(backend_ws.send.await_args_list[0].args[0])["setup"]
+    assert setup["realtimeInputConfig"]["automaticActivityDetection"]["silenceDurationMs"] == 1500
+    assert setup["generationConfig"]["speechConfig"]["voiceConfig"]["prebuiltVoiceConfig"]["voiceName"] == "Kore"
+
+
+def test_merge_session_fields_replaces_nested_mappings_wholesale_past_the_depth_cap():
+    def nested(levels: int, key: str) -> dict:
+        return {key: levels, **({"child": nested(levels - 1, key)} if levels else {})}
+
+    def level(node, depth: int):
+        return node if depth == 0 else level(node["child"], depth - 1)
+
+    merged = _merge_session_fields(nested(10, "keep"), nested(10, "new"))
+
+    assert all({"keep", "new"} <= set(level(merged, depth)) for depth in range(9))
+    assert "keep" not in level(merged, 9)
+    assert level(merged, 9)["new"] == 1
+
+
+@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
+    hang while its updates are held, so each one is acknowledged right away
+    with the modalities the merged setup will carry."""
+    instructions_update = json.dumps({"type": "session.update", "session": {"instructions": "Be brief."}})
+    modalities_update = json.dumps({"type": "session.update", "session": {"modalities": ["text"]}})
+    streaming, backend_ws = _gemini_default_mode_streaming(
+        monkeypatch, [instructions_update, modalities_update], model="gemini-live-2.5-flash"
+    )
+
+    await streaming.client_ack_messages()
+
+    acks = [json.loads(call.args[0]) for call in streaming.websocket.send_text.await_args_list]
+    assert [ack["type"] for ack in acks] == ["session.updated", "session.updated"]
+    assert [ack["session"].get("output_modalities") or ack["session"].get("modalities") for ack in acks] == [
+        ["audio"],
+        ["text"],
+    ]
+    assert acks[1]["session"]["instructions"] == "Be brief."
+    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, _RESPONSE_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
+    has to land in the one setup built from the held session.updates, injected
+    once rather than into every update that gets merged."""
+    monkeypatch.setattr(litellm, "callbacks", [_transcription_guardrail()])
+    debug = MagicMock()
+    monkeypatch.setattr("litellm.litellm_core_utils.realtime_streaming.verbose_logger.debug", debug)
+    instructions_update = json.dumps({"type": "session.update", "session": {"instructions": "Be brief."}})
+    vad_update = json.dumps(
+        {"type": "session.update", "session": {"turn_detection": {"type": "server_vad", "silence_duration_ms": 900}}}
+    )
+    streaming, backend_ws = _gemini_default_mode_streaming(
+        monkeypatch, [instructions_update, vad_update, _TEXT_ITEM_CREATE, _RESPONSE_CREATE], model="gemini-live-2.5-flash"
+    )
+
+    await streaming.client_ack_messages()
+
+    frames = [json.loads(call.args[0]) for call in backend_ws.send.await_args_list]
+    assert [next(iter(frame)) for frame in frames] == ["setup", "clientContent"]
+    assert frames[0]["setup"]["realtimeInputConfig"]["automaticActivityDetection"] == {"disabled": True}
+    assert frames[0]["setup"]["systemInstruction"]["parts"][0]["text"] == "Be brief."
+    assert streaming._guardrail_turn_detection_update_sent is True
+    injected = [call for call in debug.call_args_list if str(call.args[0]).startswith("Injected turn_detection")]
+    assert len(injected) == 1
+
+
+def _blocking_text_guardrail() -> CustomGuardrail:
+    class _BlockingTextGuardrail(CustomGuardrail):
+        async def apply_guardrail(self, inputs, request_data, input_type, logging_obj=None):
+            if any("zebrafruit" in text for text in inputs.get("texts", [])):
+                raise ValueError("Content blocked")
+            return inputs
+
+    return _BlockingTextGuardrail(
+        guardrail_name="test_text_guard", event_hook=GuardrailEventHooks.pre_call, default_on=True
+    )
+
+
+@pytest.mark.asyncio
+async def test_default_mode_sends_the_setup_before_a_guardrail_blocks_the_first_message(monkeypatch):
+    """Gemini Live closes the socket when content arrives before the setup, so a
+    guardrail that blocks the first typed message must not reach the backend first."""
+    monkeypatch.setattr(litellm, "callbacks", [_blocking_text_guardrail()])
+    instructions_update = json.dumps({"type": "session.update", "session": {"instructions": "Be brief."}})
+    blocked_item = json.dumps(
+        {
+            "type": "conversation.item.create",
+            "item": {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "zebrafruit"}]},
+        }
+    )
+    streaming, backend_ws = _gemini_default_mode_streaming(
+        monkeypatch, [instructions_update, blocked_item], model="gemini-live-2.5-flash"
+    )
+
+    await streaming.client_ack_messages()
+
+    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"]
+    assert frames[0]["setup"]["systemInstruction"]["parts"][0]["text"] == "Be brief."
+    assert "zebrafruit" not in json.dumps(frames[1])
+    errors: Final = [event["error"]["type"] for event in _events_sent_to_client(streaming) if event["type"] == "error"]
+    assert errors == ["guardrail_violation"]
+
+
+class _SteppedClock:
+    def __init__(self, epoch: float) -> None:
+        self.now = epoch
+
+    def __call__(self) -> float:
+        return self.now
+
+
+def _client_ws_receiving(*steps: Callable[[], Coroutine[object, object, str]]) -> MagicMock:
+    remaining: Final = iter(steps)
+
+    async def receive_text() -> str:
+        return await next(remaining)()
+
+    client_ws: Final = MagicMock()
+    client_ws.receive_text = receive_text
+    client_ws.send_text = AsyncMock()
+    return client_ws
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("epoch", [0.0, 1e9])
+async def test_default_mode_sends_the_held_setup_when_no_content_arrives_before_the_deadline(monkeypatch, epoch):
+    """Vertex AI refuses a setup sent minutes after connecting, so an idle session
+    sends what it holds at the deadline and answers any later session.update as late."""
+    monkeypatch.setattr(litellm, "gemini_live_defer_setup", False, raising=False)
+    from litellm.llms.gemini.realtime.transformation import GeminiRealtimeConfig
+
+    clock: Final = _SteppedClock(epoch)
+    setup_sent: Final = asyncio.Event()
+
+    async def instructions_update_then_the_deadline_passes() -> str:
+        clock.now += REALTIME_LAZY_SETUP_DEADLINE_SECONDS
+        return json.dumps({"type": "session.update", "session": {"instructions": "Be brief."}})
+
+    async def late_update_once_the_setup_is_out() -> str:
+        await setup_sent.wait()
+        return json.dumps(
+            {"type": "session.update", "event_id": "evt_late", "session": {"instructions": "Answer in French."}}
+        )
+
+    async def disconnect() -> str:
+        raise ConnectionClosed(None, None)
+
+    backend_ws: Final = MagicMock()
+    backend_ws.send = AsyncMock(side_effect=lambda frame: setup_sent.set())
+    streaming: Final = RealTimeStreaming(
+        _client_ws_receiving(instructions_update_then_the_deadline_passes, late_update_once_the_setup_is_out, disconnect),
+        backend_ws,
+        MagicMock(),
+        provider_config=GeminiRealtimeConfig(),
+        model="gemini-live-2.5-flash",
+        clock=clock,
+    )
+
+    await asyncio.wait_for(streaming.client_ack_messages(), timeout=5)
+
+    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"]
+    assert frames[0]["setup"]["systemInstruction"]["parts"][0]["text"] == "Be brief."
+    events: Final = _events_sent_to_client(streaming)
+    assert [(event["type"], event.get("error", {}).get("code")) for event in events] == [
+        ("session.updated", None),
+        ("error", "session_update_not_applied"),
+    ]
+
+
+@pytest.mark.asyncio
+async def test_default_mode_refuses_held_session_updates_past_the_size_cap(monkeypatch):
+    def instructions_update(text: str) -> str:
+        return json.dumps({"type": "session.update", "event_id": f"evt_{text[0]}", "session": {"instructions": text}})
+
+    half_the_cap: Final = RealTimeStreaming._MAX_BUFFERED_BYTES // 2
+    streaming, backend_ws = _gemini_default_mode_streaming(
+        monkeypatch,
+        [instructions_update("a" * half_the_cap), instructions_update("b" * half_the_cap), _TEXT_ITEM_CREATE, _RESPONSE_CREATE],
+        model="gemini-live-2.5-flash",
+    )
+
+    await streaming.client_ack_messages()
+
+    events: Final = _events_sent_to_client(streaming)
+    assert [(event["type"], event.get("error", {}).get("event_id")) for event in events] == [
+        ("session.updated", None),
+        ("error", "evt_b"),
+    ]
+    assert events[1]["error"]["code"] == "session_update_too_large"
+    setup: Final = json.loads(backend_ws.send.await_args_list[0].args[0])["setup"]
+    assert setup["systemInstruction"]["parts"][0]["text"].startswith("a")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("defer_setup", [False, True])
+async def test_a_session_update_between_the_first_items_and_response_create_joins_the_setup(monkeypatch, defer_setup):
+    modalities_update: Final = json.dumps({"type": "session.update", "session": {"modalities": ["text"]}})
+    instructions_update: Final = json.dumps({"type": "session.update", "session": {"instructions": "Be brief."}})
+    streaming, backend_ws = _gemini_default_mode_streaming(
+        monkeypatch,
+        [modalities_update, _TEXT_ITEM_CREATE, instructions_update, _RESPONSE_CREATE],
+        model="gemini-live-2.5-flash",
+        defer_setup=defer_setup,
+    )
+
+    await streaming.client_ack_messages()
+
+    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"]
+    assert frames[0]["setup"]["generationConfig"]["responseModalities"] == ["TEXT"]
+    assert frames[0]["setup"]["systemInstruction"]["parts"][0]["text"] == "Be brief."
+    assert [event["type"] for event in _events_sent_to_client(streaming)] == ["session.updated", "session.updated"]
+
+
+@pytest.mark.asyncio
+async def test_default_mode_applies_a_session_update_that_arrives_while_audio_streams_in(monkeypatch):
+    modalities_update: Final = json.dumps({"type": "session.update", "session": {"modalities": ["text"]}})
+    append: Final = json.dumps({"type": "input_audio_buffer.append", "audio": "AAAA"})
+    instructions_update: Final = json.dumps({"type": "session.update", "session": {"instructions": "Be brief."}})
+    commit: Final = json.dumps({"type": "input_audio_buffer.commit"})
+    streaming, backend_ws = _gemini_default_mode_streaming(
+        monkeypatch, [modalities_update, append, instructions_update, commit], model="gemini-live-2.5-flash"
+    )
+
+    await streaming.client_ack_messages()
+
+    frames: Final = [json.loads(call.args[0]) for call in backend_ws.send.await_args_list]
+    assert [next(iter(frame)) for frame in frames][:2] == ["setup", "realtimeInput"]
+    assert frames[0]["setup"]["systemInstruction"]["parts"][0]["text"] == "Be brief."
+    assert frames[1]["realtimeInput"]["audio"]["data"] == "AAAA"
+    assert "error" not in [event["type"] for event in _events_sent_to_client(streaming)]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("epoch", [0.0, 1e9])
+async def test_held_items_go_out_with_the_setup_when_no_response_create_follows(monkeypatch, epoch):
+    monkeypatch.setattr(litellm, "gemini_live_defer_setup", False, raising=False)
+    from litellm.llms.gemini.realtime.transformation import GeminiRealtimeConfig
+
+    item_sent: Final = asyncio.Event()
+
+    async def instructions_update() -> str:
+        return json.dumps({"type": "session.update", "session": {"instructions": "Be brief."}})
+
+    async def text_item() -> str:
+        return _TEXT_ITEM_CREATE
+
+    async def disconnect_once_the_item_is_out() -> str:
+        await item_sent.wait()
+        raise ConnectionClosed(None, None)
+
+    backend_ws: Final = MagicMock()
+    backend_ws.send = AsyncMock(
+        side_effect=lambda frame: item_sent.set() if "clientContent" in json.loads(frame) else None
+    )
+    streaming: Final = RealTimeStreaming(
+        _client_ws_receiving(instructions_update, text_item, disconnect_once_the_item_is_out),
+        backend_ws,
+        MagicMock(),
+        provider_config=GeminiRealtimeConfig(),
+        model="gemini-live-2.5-flash",
+        lazy_setup_content_grace_seconds=0,
+        clock=_SteppedClock(epoch),
+    )
+
+    await asyncio.wait_for(streaming.client_ack_messages(), timeout=5)
+
+    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"]
+    assert frames[0]["setup"]["systemInstruction"]["parts"][0]["text"] == "Be brief."
+
+
+@pytest.mark.asyncio
+async def test_default_mode_acknowledges_an_unchanged_session_update_after_content(monkeypatch):
+    """Pipecat resends its unchanged session.update right after its first
+    conversation items and treats any error event as fatal."""
+    session_update: Final = json.dumps(
+        {"type": "session.update", "session": {"modalities": ["text"], "instructions": "Be brief."}}
+    )
+    streaming, backend_ws = _gemini_default_mode_streaming(
+        monkeypatch, [session_update, _TEXT_ITEM_CREATE, _RESPONSE_CREATE, session_update], model="gemini-live-2.5-flash"
+    )
+
+    await streaming.client_ack_messages()
+
+    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"]
+    events: Final = _events_sent_to_client(streaming)
+    assert [event["type"] for event in events] == ["session.updated", "session.updated"]
+    assert events[0]["session"] == events[1]["session"]
+
+
+@pytest.mark.asyncio
+async def test_deferred_setup_merges_session_updates_until_the_first_content(monkeypatch):
+    """Deferred setup merges every session.update sent before the first content,
+    the same as the default mode, and answers a changed one after it as late."""
+    monkeypatch.setattr(litellm, "gemini_live_defer_setup", True, raising=False)
+    from litellm.llms.gemini.realtime.transformation import GeminiRealtimeConfig
+
+    client_ws: Final = MagicMock()
+    client_ws.receive_text = AsyncMock(
+        side_effect=[
+            json.dumps({"type": "session.update", "session": {"modalities": ["text"]}}),
+            json.dumps({"type": "session.update", "session": {"instructions": "Be brief."}}),
+            _TEXT_ITEM_CREATE,
+            _RESPONSE_CREATE,
+            json.dumps(
+                {"type": "session.update", "event_id": "evt_late", "session": {"instructions": "Answer in French."}}
+            ),
+            ConnectionClosed(None, None),
+        ]
+    )
+    client_ws.send_text = AsyncMock()
+    backend_ws: Final = MagicMock()
+    backend_ws.send = AsyncMock()
+    streaming: Final = RealTimeStreaming(
+        client_ws, backend_ws, MagicMock(), provider_config=GeminiRealtimeConfig(), model="gemini-live-2.5-flash"
+    )
+
+    await streaming.client_ack_messages()
+
+    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"]
+    assert frames[0]["setup"]["generationConfig"]["responseModalities"] == ["TEXT"]
+    assert frames[0]["setup"]["systemInstruction"]["parts"][0]["text"] == "Be brief."
+    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")]
+
+
+@pytest.mark.asyncio
+async def test_deferred_setup_sends_the_default_setup_before_text_content_that_comes_first(monkeypatch):
+    """Gemini Live closes the socket on content sent before any setup, so text that arrives
+    before a session.update in deferred mode gets the model's default setup first."""
+    monkeypatch.setattr(litellm, "gemini_live_defer_setup", True, raising=False)
+    from litellm.llms.gemini.realtime.transformation import GeminiRealtimeConfig
+
+    client_ws: Final = MagicMock()
+    client_ws.receive_text = AsyncMock(
+        side_effect=[
+            _TEXT_ITEM_CREATE,
+            json.dumps({"type": "session.update", "event_id": "evt_late", "session": {"modalities": ["text"]}}),
+            ConnectionClosed(None, None),
+        ]
+    )
+    client_ws.send_text = AsyncMock()
+    backend_ws: Final = MagicMock()
+    backend_ws.send = AsyncMock()
+    streaming: Final = RealTimeStreaming(
+        client_ws, backend_ws, MagicMock(), provider_config=GeminiRealtimeConfig(), model="gemini-live-2.5-flash"
+    )
+
+    await streaming.client_ack_messages()
+
+    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"]
+    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")]
+
+
+@pytest.mark.asyncio
+async def test_deferred_setup_sends_a_session_update_that_follows_buffered_audio_at_once(monkeypatch):
+    append: Final = json.dumps({"type": "input_audio_buffer.append", "audio": "AAAA"})
+    commit: Final = json.dumps({"type": "input_audio_buffer.commit"})
+    session_update: Final = json.dumps({"type": "session.update", "session": {"instructions": "Be a pirate."}})
+    streaming, backend_ws = _gemini_default_mode_streaming(
+        monkeypatch, [append, commit, session_update], model="gemini-live-2.5-flash", defer_setup=True
+    )
+
+    await streaming.client_ack_messages()
+
+    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"]
+    assert "Be a pirate." in json.dumps(frames[0]["setup"]["systemInstruction"])
+
+
+@pytest.mark.parametrize("defer_setup", [False, True])
+@pytest.mark.asyncio
+async def test_a_transcription_guardrail_refuses_gemini_audio_it_cannot_screen(monkeypatch, defer_setup):
+    monkeypatch.setattr(litellm, "callbacks", [_transcription_guardrail()])
+    session_update: Final = json.dumps(
+        {"type": "session.update", "session": {"turn_detection": {"type": "server_vad", "create_response": False}}}
+    )
+    append: Final = json.dumps({"type": "input_audio_buffer.append", "audio": "AAAA"})
+    commit: Final = json.dumps({"type": "input_audio_buffer.commit", "event_id": "commit_1"})
+    streaming, backend_ws = _gemini_default_mode_streaming(
+        monkeypatch,
+        [session_update, append, append, commit, json.dumps({"type": "response.create"}), _TEXT_ITEM_CREATE],
+        model="gemini-live-2.5-flash",
+        defer_setup=defer_setup,
+    )
+
+    await streaming.client_ack_messages()
+
+    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"]
+    refusals: Final = [
+        event["error"]["event_id"]
+        for event in _events_sent_to_client(streaming)
+        if event["type"] == "error" and event["error"]["code"] == "audio_input_not_screenable"
+    ]
+    assert refusals == [None, "commit_1"]
+
+
+@pytest.mark.parametrize("defer_setup", [False, True])
+@pytest.mark.parametrize("control_type", ["response.cancel", "conversation.item.delete", "input_audio_buffer.clear"])
+@pytest.mark.asyncio
+async def test_a_control_event_before_session_update_leaves_the_setup_open(monkeypatch, defer_setup, control_type):
+    control_event: Final = json.dumps({"type": control_type, "item_id": "item_x"})
+    session_update: Final = json.dumps({"type": "session.update", "session": {"modalities": ["text"]}})
+    streaming, backend_ws = _gemini_default_mode_streaming(
+        monkeypatch,
+        [control_event, session_update, _TEXT_ITEM_CREATE, _RESPONSE_CREATE],
+        model="gemini-live-2.5-flash",
+        defer_setup=defer_setup,
+    )
+
+    await streaming.client_ack_messages()
+
+    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"]
+    assert frames[0]["setup"]["generationConfig"]["responseModalities"] == ["TEXT"]
+    assert [event["type"] for event in _events_sent_to_client(streaming) if event["type"] == "error"] == []
+
+
 @pytest.mark.asyncio
 async def test_deferred_setup_buffers_audio_until_backend_setup_complete(monkeypatch):
     """Pipecat may send audio before session.update when setup is deferred."""
@@ -2533,14 +3173,18 @@ async def test_deferred_setup_buffers_audio_until_backend_setup_complete(monkeyp
 
 
 @pytest.mark.asyncio
-async def test_deferred_setup_sends_session_update_before_buffered_audio(monkeypatch):
+async def test_deferred_setup_sends_the_held_session_update_when_more_audio_arrives(monkeypatch):
     monkeypatch.setattr(litellm, "gemini_live_defer_setup", True, raising=False)
     from litellm.llms.gemini.realtime.transformation import GeminiRealtimeConfig
 
     client_ws = MagicMock()
+    client_ws.send_text = AsyncMock()
     audio_msg = json.dumps({"type": "input_audio_buffer.append", "audio": "AA=="})
     session_update = json.dumps({"type": "session.update", "session": {"modalities": ["audio"]}})
-    client_ws.receive_text = AsyncMock(side_effect=[audio_msg, session_update, ConnectionClosed(None, None)])
+    later_audio_msg = json.dumps({"type": "input_audio_buffer.append", "audio": "AQ=="})
+    client_ws.receive_text = AsyncMock(
+        side_effect=[audio_msg, session_update, later_audio_msg, ConnectionClosed(None, None)]
+    )
     backend_ws = MagicMock()
     backend_ws.send = AsyncMock()
     logging_obj: Final = _mock_logging()
@@ -2560,7 +3204,7 @@ async def test_deferred_setup_sends_session_update_before_buffered_audio(monkeyp
     sent_payload = json.loads(backend_ws.send.await_args_list[0].args[0])
     assert "setup" in sent_payload
     assert "realtimeInput" not in sent_payload
-    assert streaming._pending_messages_until_setup == [audio_msg]
+    assert streaming._pending_messages_until_setup == [audio_msg, later_audio_msg]
 
 
 @pytest.mark.asyncio
@@ -3523,6 +4167,24 @@ async def test_upstream_refusal_after_a_synthetic_session_created_still_logs_a_f
     assert session.logging.logged_sessions == ()
 
 
+class _LazySetupRealtimeConfig(_PassthroughRealtimeConfig):
+    def builds_setup_from_session_update(self) -> bool:
+        return True
+
+
+@pytest.mark.asyncio
+async def test_upstream_outage_before_a_held_setup_was_sent_logs_the_session_not_a_refusal():
+    upstream_close: Final = ConnectionClosed(Close(1011, "upstream outage"), None)
+    session: Final = _relay_session(
+        _client_ws_that_never_sends(), _backend_ws_closing_with(upstream_close), _LazySetupRealtimeConfig()
+    )
+
+    await session.run()
+
+    assert session.logging.logged_failures == ()
+    assert len(session.logging.logged_sessions) == 1
+
+
 @pytest.mark.asyncio
 async def test_upstream_close_after_relayed_events_still_logs_the_session_as_success():
     client_ws: Final = _client_ws_that_never_sends()

tests/unit/llms/custom_httpx/test_llm_http_handler.py

@@ -14,6 +14,7 @@ import httpcore
 import httpx
 import pytest
 import respx
+import websockets.exceptions
 from botocore.credentials import RefreshableCredentials
 from h2.events import ConnectionTerminated
 
@@ -2787,6 +2788,194 @@ async def test_async_realtime_error_event_send_failure_still_closes():
     assert client_ws.events == [("close", (1011, "Internal server error: vertex token refresh exploded"))]
 
 
+_GEMINI_LIVE_SETUP_COMPLETE = json.dumps({"setupComplete": {}})
+_GEMINI_LIVE_TEXT_TURN = json.dumps({"serverContent": {"modelTurn": {"parts": [{"text": "Blue."}]}}})
+_GEMINI_LIVE_TURN_COMPLETE = json.dumps({"serverContent": {"turnComplete": True}})
+
+
+class _ScriptedRealtimeBackend:
+    """Backend socket that answers the first setup frame with the scripted replies
+    and closes once the client is done, so both bridge loops run to completion."""
+
+    def __init__(self, replies_after_setup):
+        self.sent = []
+        self.closed = asyncio.Event()
+        self._replies_after_setup = list(replies_after_setup)
+        self._inbox = asyncio.Queue()
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return None
+
+    async def send(self, payload):
+        self.sent.append(payload)
+        if "setup" in json.loads(payload):
+            for reply in self._replies_after_setup:
+                self._inbox.put_nowait(reply)
+
+    def close_after_replies(self):
+        self._inbox.put_nowait(None)
+
+    async def recv(self, decode=True):
+        reply = await self._inbox.get()
+        if reply is None:
+            self.closed.set()
+            raise websockets.exceptions.ConnectionClosedOK(None, None)
+        return reply
+
+
+class _ScriptedRealtimeClient:
+    def __init__(self, messages, backend):
+        self.scope = {"headers": []}
+        self.exceptions = websockets.exceptions
+        self.sent = []
+        self._messages = list(messages)
+        self._backend = backend
+
+    async def send_text(self, payload):
+        self.sent.append(payload)
+
+    async def receive_text(self):
+        if self._messages:
+            return self._messages.pop(0)
+        self._backend.close_after_replies()
+        await self._backend.closed.wait()
+        raise websockets.exceptions.ConnectionClosedOK(None, None)
+
+    async def close(self, code=None, reason=None):
+        return None
+
+
+def _realtime_logging_obj():
+    logging_obj = Mock()
+    logging_obj.litellm_trace_id = "trace-lit-6974"
+    logging_obj.async_success_handler = AsyncMock()
+    logging_obj.dispatch_failure_handlers = AsyncMock()
+    return logging_obj
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "model, session, backend_modality, client_modality, backend_turn, drop_params",
+    [
+        (
+            "gemini-live-2.5-flash",
+            {"type": "realtime", "output_modalities": ["text"]},
+            "TEXT",
+            "text",
+            [_GEMINI_LIVE_TEXT_TURN],
+            False,
+        ),
+        ("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, 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. 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)
+    backend = _ScriptedRealtimeBackend([_GEMINI_LIVE_SETUP_COMPLETE, *backend_turn, _GEMINI_LIVE_TURN_COMPLETE])
+    client_ws = _ScriptedRealtimeClient(
+        [
+            json.dumps({"type": "session.update", "session": session}),
+            json.dumps(
+                {
+                    "type": "conversation.item.create",
+                    "item": {
+                        "type": "message",
+                        "role": "user",
+                        "content": [{"type": "input_text", "text": "What color is the sky? One word."}],
+                    },
+                }
+            ),
+            json.dumps({"type": "response.create"}),
+        ],
+        backend,
+    )
+    handler = BaseLLMHTTPHandler()
+
+    with patch.object(handler, "_open_realtime_backend_ws", AsyncMock(return_value=backend)):
+        await asyncio.wait_for(
+            handler.async_realtime(
+                model=model,
+                websocket=client_ws,
+                logging_obj=_realtime_logging_obj(),
+                provider_config=VertexAIRealtimeConfig(
+                    access_token="fake-token", project="fake-project", location="us-central1", drop_params=drop_params
+                ),
+                headers={},
+            ),
+            timeout=10,
+        )
+
+    backend_frames = [json.loads(frame) for frame in backend.sent]
+    setups = [frame["setup"] for frame in backend_frames if "setup" in frame]
+    assert len(setups) == 1
+    assert "setup" in backend_frames[0]
+    assert setups[0]["generationConfig"]["responseModalities"] == [backend_modality]
+    assert any("clientContent" in frame for frame in backend_frames[1:])
+
+    client_events = [json.loads(frame) for frame in client_ws.sent]
+    event_types = [event["type"] for event in client_events]
+    assert event_types[:2] == ["session.created", "session.updated"]
+    assert client_events[1]["session"]["modalities"] == [client_modality]
+    if backend_turn:
+        assert "response.output_text.delta" in event_types
+        assert "response.output_audio_transcript.delta" not in event_types
+
+
+@pytest.mark.asyncio
+async def test_async_realtime_sends_setup_at_connect_when_provider_cannot_build_it_from_session_update():
+    """Bedrock Nova Sonic style providers keep the connect-time setup and get no
+    synthetic session.created."""
+    setup = json.dumps({"event": {"sessionStart": {}}})
+    backend = _ScriptedRealtimeBackend([])
+    client_ws = _ScriptedRealtimeClient([], backend)
+    provider_config = Mock()
+    provider_config.get_complete_url.return_value = "wss://backend.example/live"
+    provider_config.validate_environment.return_value = {}
+    provider_config.requires_session_configuration.return_value = True
+    provider_config.builds_setup_from_session_update.return_value = False
+    provider_config.session_configuration_request.return_value = setup
+    provider_config.unbilled_usage_on_session_close.return_value = None
+    provider_config.open_backend = AsyncMock(return_value=None)
+    provider_config.wrap_backend.side_effect = lambda opened: opened
+    handler = BaseLLMHTTPHandler()
+
+    with patch.object(handler, "_open_realtime_backend_ws", AsyncMock(return_value=backend)):
+        await asyncio.wait_for(
+            handler.async_realtime(
+                model="amazon.nova-sonic-v1:0",
+                websocket=client_ws,
+                logging_obj=_realtime_logging_obj(),
+                provider_config=provider_config,
+                headers={},
+            ),
+            timeout=10,
+        )
+
+    assert backend.sent == [setup]
+    assert client_ws.sent == []
+
+
 class _JSONBodyAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
     def get_supported_openai_params(self, model):
         return []

tests/unit/llms/gemini/realtime/test_gemini_realtime_transformation.py

@@ -1,3 +1,4 @@
+import base64
 import json
 from collections.abc import Mapping
 from typing import Final, cast
@@ -7,7 +8,7 @@ import pytest
 
 
 import litellm
-from litellm.llms.gemini.realtime.transformation import GeminiRealtimeConfig
+from litellm.llms.gemini.realtime.transformation import GEMINI_LIVE_DOCUMENTED_VOICES, GeminiRealtimeConfig
 from litellm.types.llms.gemini import BidiGenerateContentServerMessage
 
 
@@ -439,6 +440,51 @@ def test_gemini_realtime_session_update_with_tools():
     assert "parameters" in function_decl
 
 
+def _schema_nodes(schema: object):
+    if isinstance(schema, dict):
+        yield schema
+        for value in schema.values():
+            yield from _schema_nodes(value)
+    elif isinstance(schema, list):
+        for value in schema:
+            yield from _schema_nodes(value)
+
+
+def test_gemini_session_update_converts_flat_realtime_tool_schemas_to_gemini_schemas():
+    session_update: Final = {
+        "type": "session.update",
+        "session": {
+            "tools": [
+                {
+                    "type": "function",
+                    "name": "lookup_weather",
+                    "description": "Look up the weather in a city.",
+                    "parameters": {
+                        "type": "object",
+                        "$defs": {
+                            "City": {"type": "object", "properties": {"name": {"type": "string"}}, "required": ["name"]}
+                        },
+                        "properties": {"city": {"$ref": "#/$defs/City"}, "note": {"type": ["string", "null"]}},
+                        "required": ["city"],
+                    },
+                }
+            ]
+        },
+    }
+
+    messages: Final = GeminiRealtimeConfig().transform_realtime_request(
+        json.dumps(session_update), "gemini-2.5-flash", session_configuration_request=None
+    )
+
+    declaration: Final = json.loads(messages[0])["setup"]["tools"][0]["function_declarations"][0]
+    assert declaration["name"] == "lookup_weather"
+    parameters: Final = declaration["parameters"]
+    assert parameters["properties"]["city"]["properties"]["name"]["type"] == "string"
+    nodes: Final = tuple(_schema_nodes(parameters))
+    assert not any("$defs" in node or "$ref" in node for node in nodes)
+    assert all(isinstance(node["type"], str) for node in nodes if "type" in node)
+
+
 def test_gemini_session_update_defaults_to_audio_modality():
     config = GeminiRealtimeConfig()
 
@@ -613,6 +659,35 @@ def test_gemini_audio_only_live_models_drop_text_from_text_audio_combo(patch_gem
     assert setup["generationConfig"]["responseModalities"] == ["AUDIO"]
 
 
+@pytest.mark.parametrize("output_modalities", ["text", ["text"]])
+def test_gemini_session_update_accepts_a_single_string_output_modality(output_modalities):
+    """Regression: a bare string was iterated per character, so output_modalities
+    "text" reached Vertex as responseModalities ["T"], which it rejects with 1007."""
+    config = GeminiRealtimeConfig()
+    session_update = {"type": "session.update", "session": {"type": "realtime", "output_modalities": output_modalities}}
+
+    messages = config.transform_realtime_request(
+        json.dumps(session_update), "gemini-live-2.5-flash", session_configuration_request=None
+    )
+
+    assert json.loads(messages[0])["setup"]["generationConfig"]["responseModalities"] == ["TEXT"]
+
+
+@pytest.mark.parametrize(
+    ("output_modalities", "expected"),
+    [(["video", "text"], ["TEXT"]), (["video"], ["AUDIO"]), (None, ["AUDIO"])],
+)
+def test_gemini_session_update_drops_modalities_live_cannot_produce(output_modalities, expected):
+    config = GeminiRealtimeConfig()
+    session_update = {"type": "session.update", "session": {"type": "realtime", "output_modalities": output_modalities}}
+
+    messages = config.transform_realtime_request(
+        json.dumps(session_update), "gemini-live-2.5-flash", session_configuration_request=None
+    )
+
+    assert json.loads(messages[0])["setup"]["generationConfig"]["responseModalities"] == expected
+
+
 def test_gemini_non_live_model_preserves_text_modality():
     config = GeminiRealtimeConfig()
     session_update = {
@@ -1396,22 +1471,71 @@ def test_gemini_input_audio_buffer_clear_is_local_noop():
     assert messages == []
 
 
-def test_gemini_input_audio_buffer_commit_maps_to_activity_end_when_manual_vad():
+_MANUAL_VAD_SETUP = json.dumps({"setup": {"realtimeInputConfig": {"automaticActivityDetection": {"disabled": True}}}})
+_APPEND = json.dumps({"type": "input_audio_buffer.append", "audio": base64.b64encode(b"\x00\x01" * 80).decode()})
+
+
+def _realtime_input_keys(config: GeminiRealtimeConfig, client_event: str, setup: str | None) -> list[str]:
+    frames = config.transform_realtime_request(client_event, "gemini-live-2.5-flash-native-audio", setup)
+    return [key for frame in frames for key in json.loads(frame)["realtimeInput"]]
+
+
+@pytest.mark.parametrize("turn_end", ["input_audio_buffer.commit", "response.create"])
+def test_gemini_manual_vad_wraps_each_audio_turn_in_activity_start_and_end(turn_end):
+    """With activity detection off, Gemini Live answers only after activityEnd closes an
+    activityStart, and it closes the socket on any other shape of those markers."""
     config = GeminiRealtimeConfig()
-    setup = {
-        "setup": {
-            "realtimeInputConfig": {
-                "automaticActivityDetection": {"disabled": True},
-            }
-        }
-    }
-    messages = config.transform_realtime_request(
-        json.dumps({"type": "input_audio_buffer.commit"}),
-        "gemini-live-2.5-flash-native-audio",
-        session_configuration_request=json.dumps(setup),
+    end = json.dumps({"type": turn_end})
+
+    turns = [
+        [_realtime_input_keys(config, event, _MANUAL_VAD_SETUP) for event in (_APPEND, _APPEND, end, end)]
+        for _ in range(2)
+    ]
+
+    assert turns == [[["activityStart", "audio"], ["audio"], ["activityEnd"], []]] * 2
+
+
+def test_gemini_manual_vad_activity_markers_are_empty_messages():
+    config = GeminiRealtimeConfig()
+    commit = json.dumps({"type": "input_audio_buffer.commit"})
+
+    first_append = config.transform_realtime_request(_APPEND, "gemini-live-2.5-flash", _MANUAL_VAD_SETUP)
+    turn_end = config.transform_realtime_request(commit, "gemini-live-2.5-flash", _MANUAL_VAD_SETUP)
+
+    assert json.loads(first_append[0]) == {"realtimeInput": {"activityStart": {}}}
+    assert [json.loads(frame) for frame in turn_end] == [{"realtimeInput": {"activityEnd": {}}}]
+
+
+@pytest.mark.parametrize(
+    ("session", "expected_keys"),
+    [
+        ({"turn_detection": None}, [["activityStart", "audio"], ["activityEnd"]]),
+        ({"type": "realtime", "audio": {"input": {"turn_detection": None}}}, [["activityStart", "audio"], ["activityEnd"]]),
+        ({"type": "realtime", "audio": {"input": {}}}, [["audio"], ["audioStreamEnd"]]),
+    ],
+)
+def test_gemini_null_turn_detection_makes_the_client_mark_each_turn(session, expected_keys):
+    """OpenAI's turn_detection null turns server VAD off, so the client commits each turn itself."""
+    config = GeminiRealtimeConfig()
+    setup = config.transform_realtime_request(
+        json.dumps({"type": "session.update", "session": session}), "gemini-live-2.5-flash-native-audio", None
     )
-    assert len(messages) == 1
-    assert json.loads(messages[0]) == {"realtimeInput": {"activityEnd": True}}
+    commit = json.dumps({"type": "input_audio_buffer.commit"})
+
+    keys = [_realtime_input_keys(config, event, setup[0]) for event in (_APPEND, commit)]
+
+    assert keys == expected_keys
+
+
+def test_gemini_automatic_vad_sends_no_activity_markers():
+    config = GeminiRealtimeConfig()
+    commit = json.dumps({"type": "input_audio_buffer.commit"})
+    response_create = json.dumps({"type": "response.create"})
+
+    keys = [_realtime_input_keys(config, event, None) for event in (_APPEND, commit)]
+
+    assert keys == [["audio"], ["audioStreamEnd"]]
+    assert config.transform_realtime_request(response_create, "gemini-live-2.5-flash", None) == []
 
 
 def test_gemini_tool_call_id_to_name_evicts_oldest_when_capped():
@@ -1883,13 +2007,75 @@ def test_map_openai_params_drops_stock_voice_case_insensitively():
     """
     cfg = GeminiRealtimeConfig()
 
-    dropped = cfg.map_openai_params(optional_params={}, non_default_params={"voice": "Alloy"})
+    dropped = cfg.map_openai_params(optional_params={}, non_default_params={"voice": "Alloy"}, model="gemini-3.8-live")
     assert "speechConfig" not in dropped.get("generationConfig", {})
 
-    passthrough = cfg.map_openai_params(optional_params={}, non_default_params={"voice": "Kore"})
+    passthrough = cfg.map_openai_params(
+        optional_params={}, non_default_params={"voice": "Kore"}, model="gemini-3.8-live"
+    )
     assert passthrough["generationConfig"]["speechConfig"]["voiceConfig"]["prebuiltVoiceConfig"]["voiceName"] == "Kore"
 
 
+_DOCUMENTED_VOICE: Final = sorted(GEMINI_LIVE_DOCUMENTED_VOICES)[0].capitalize()
+
+
+@pytest.mark.parametrize("drop_params", [False, True])
+def test_google_ai_studio_forwards_a_documented_voice(drop_params):
+    config: Final = GeminiRealtimeConfig(drop_params=drop_params)
+    mapped: Final = config.map_openai_params(
+        optional_params={}, non_default_params={"voice": _DOCUMENTED_VOICE}, model="gemini-3.8-live"
+    )
+
+    speech_config: Final = mapped["generationConfig"]["speechConfig"]
+    assert speech_config["voiceConfig"]["prebuiltVoiceConfig"]["voiceName"] == _DOCUMENTED_VOICE
+    assert config.setup_session_update_rejection("gemini-3.8-live", {"voice": _DOCUMENTED_VOICE}, setup_sent=False) is None
+
+
+@pytest.mark.parametrize(
+    ("session", "param"),
+    [
+        ({"voice": "Nova"}, "session.voice"),
+        ({"type": "realtime", "audio": {"output": {"voice": "Nova"}}}, "session.audio.output.voice"),
+    ],
+)
+def test_google_ai_studio_rejects_an_undocumented_voice_on_the_session_update(session, param):
+    rejection: Final = GeminiRealtimeConfig().setup_session_update_rejection(
+        "gemini-2.5-flash-native-audio-preview-12-2025", session, setup_sent=False
+    )
+
+    assert rejection is not None
+    assert (rejection.code, rejection.param) == ("invalid_value", param)
+
+
+def _setup_speech_config(config: GeminiRealtimeConfig, model: str, session: dict[str, object]) -> object:
+    messages: Final = config.transform_realtime_request(
+        json.dumps({"type": "session.update", "session": session}), model, session_configuration_request=None
+    )
+    return json.loads(messages[0])["setup"].get("generationConfig", {}).get("speechConfig")
+
+
+@pytest.mark.parametrize(
+    "session",
+    [{"voice": "Nova"}, {"type": "realtime", "audio": {"output": {"voice": "Nova"}}}],
+)
+def test_google_ai_studio_drops_an_undocumented_voice_with_drop_params(session):
+    config: Final = GeminiRealtimeConfig(drop_params=True)
+    model: Final = "gemini-2.5-flash-native-audio-preview-12-2025"
+
+    assert config.setup_session_update_rejection(model, session, setup_sent=False) is None
+    assert _setup_speech_config(config, model, session) is None
+
+
+@pytest.mark.parametrize("drop_params", [False, True])
+@pytest.mark.parametrize("model", ["gemini-3.8-live", "gemini-3.1-flash-live-preview"])
+def test_google_ai_studio_forwards_any_voice_to_models_that_accept_it(model, drop_params):
+    config: Final = GeminiRealtimeConfig(drop_params=drop_params)
+
+    assert config.setup_session_update_rejection(model, {"voice": "Nova"}, setup_sent=False) is None
+    speech_config: Final = _setup_speech_config(config, model, {"voice": "Nova"})
+    assert speech_config == {"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Nova"}}}
+
+
 @pytest.fixture(autouse=False)
 def patch_gemini_transcribe_live_cost_map_entry(monkeypatch):
     """Inject the gemini-3.5-transcribe-live registry entry locally.
@@ -1954,6 +2140,55 @@ 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(
+    ("model", "modalities", "producible"),
+    [
+        ("gemini-3.5-transcribe-live", ["audio"], False),
+        ("gemini-3.5-transcribe-live", ["audio", "text"], True),
+        ("gemini-live-2.5-flash-native-audio", ["text"], False),
+        ("gemini-live-2.5-flash-native-audio", ["text", "audio"], True),
+    ],
+)
+@pytest.mark.parametrize("drop_params", [False, True])
+def test_gemini_live_rejects_only_output_modalities_the_model_cannot_produce_at_all(
+    model,
+    modalities,
+    producible,
+    drop_params,
+    patch_gemini_transcribe_live_cost_map_entry,
+    patch_gemini_audio_cost_map_entries,
+):
+    rejection = GeminiRealtimeConfig(drop_params=drop_params).setup_session_update_rejection(
+        model, {"output_modalities": modalities}, setup_sent=False
+    )
+
+    assert (rejection is None) is (drop_params or producible)
+    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,
 ):
@@ -2169,6 +2404,47 @@ def test_unbilled_usage_on_session_close_flushes_trailing_audio(patch_gemini_tra
     assert config.unbilled_usage_on_session_close("gemini-3.5-transcribe-live") is None
 
 
+def test_gemini_realtime_two_output_modalities_collapse_to_audio():
+    """Gemini Live closes the socket with 1007 when the setup names two response
+    modalities, so a GA session.update asking for ["text", "audio"] must reach the
+    backend as AUDIO only, the same choice the beta shim makes."""
+    config = GeminiRealtimeConfig()
+    session_update = {
+        "type": "session.update",
+        "session": {"output_modalities": ["text", "audio"], "instructions": "Be brief."},
+    }
+
+    messages = config.transform_realtime_request(
+        json.dumps(session_update),
+        "gemini-live-2.5-flash",
+        session_configuration_request=None,
+    )
+
+    assert len(messages) == 1
+    assert json.loads(messages[0])["setup"]["generationConfig"]["responseModalities"] == ["AUDIO"]
+
+
+def test_gemini_realtime_inf_max_response_output_tokens_is_not_forwarded():
+    """OpenAI's default max_response_output_tokens is the string "inf", which Gemini
+    rejects with 1007 (TYPE_INT32), so only integer limits map to maxOutputTokens."""
+    config = GeminiRealtimeConfig()
+
+    def generation_config_for(limit):
+        session_update = {
+            "type": "session.update",
+            "session": {"modalities": ["text"], "max_response_output_tokens": limit},
+        }
+        messages = config.transform_realtime_request(
+            json.dumps(session_update),
+            "gemini-live-2.5-flash",
+            session_configuration_request=None,
+        )
+        return json.loads(messages[0])["setup"]["generationConfig"]
+
+    assert "maxOutputTokens" not in generation_config_for("inf")
+    assert generation_config_for(200)["maxOutputTokens"] == 200
+
+
 def _grounded_live_frame(grounding_metadata: Mapping[str, object] | None) -> Mapping[str, object]:
     """One Live server frame. Grounding metadata and usageMetadata arrive together, as Vertex sends them."""
     from typing import Final

tests/unit/llms/vertex_ai/realtime/test_vertex_ai_realtime_transformation.py

@@ -18,6 +18,7 @@ import websockets.exceptions  # registers websockets.exceptions on the websocket
 
 import litellm
 from litellm.llms.vertex_ai.common_utils import get_vertex_base_url
+from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import OPENAI_TO_GEMINI_TTS_VOICES
 from litellm.llms.vertex_ai.realtime.transformation import VertexAIRealtimeConfig
 
 # ---------------------------------------------------------------------------
@@ -179,6 +180,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,
 ):
@@ -253,6 +273,26 @@ def test_vertex_keeps_turn_detection_tuning_when_auto_response_is_off():
     }
 
 
+
+@pytest.mark.parametrize(
+    ("session", "disabled"),
+    [
+        ({"turn_detection": None}, True),
+        ({"type": "realtime", "audio": {"input": {"turn_detection": None}}}, True),
+        ({"type": "realtime", "audio": {"input": {}}}, False),
+    ],
+)
+def test_vertex_turns_activity_detection_off_for_null_turn_detection(session, disabled):
+    """OpenAI's turn_detection null is push-to-talk, so Vertex must wait for the client's activity markers."""
+    cfg: Final = VertexAIRealtimeConfig(access_token="tok", project="my-proj", location="us-central1")
+    messages: Final = cfg.transform_realtime_request(
+        json.dumps({"type": "session.update", "session": session}),
+        _NATIVE_AUDIO_MODEL,
+        session_configuration_request=None,
+    )
+    setup: Final = json.loads(messages[0])["setup"]
+    assert setup["realtimeInputConfig"]["automaticActivityDetection"]["disabled"] is disabled
+
 # ---------------------------------------------------------------------------
 # Round-trip test: text-in / text-out via RealTimeStreaming
 # ---------------------------------------------------------------------------
@@ -607,3 +647,73 @@ def test_vertex_native_audio_unmapped_voice_passes_through(patch_native_audio_co
 
     generation_config = json.loads(messages[0])["setup"]["generationConfig"]
     assert generation_config["speechConfig"]["voiceConfig"]["prebuiltVoiceConfig"]["voiceName"] == "Kore"
+
+
+def test_vertex_session_update_two_output_modalities_collapse_to_audio():
+    """Vertex Live rejects a setup naming two response modalities, so a GA
+    ["text", "audio"] request must be sent as AUDIO only."""
+    cfg = VertexAIRealtimeConfig(access_token="tok", project="my-proj", location="global")
+    session_update = {
+        "type": "session.update",
+        "session": {"output_modalities": ["text", "audio"], "instructions": "Be brief."},
+    }
+
+    messages = cfg.transform_realtime_request(
+        json.dumps(session_update),
+        "gemini-live-2.5-flash",
+        session_configuration_request=None,
+    )
+
+    assert len(messages) == 1
+    assert json.loads(messages[0])["setup"]["generationConfig"]["responseModalities"] == ["AUDIO"]
+
+
+@pytest.mark.parametrize(
+    ("session", "param"),
+    [
+        ({"voice": "Nova"}, "session.voice"),
+        ({"type": "realtime", "audio": {"output": {"voice": "Nova"}}}, "session.audio.output.voice"),
+    ],
+)
+def test_vertex_rejects_an_undocumented_voice_on_the_session_update(session, param):
+    cfg: Final = VertexAIRealtimeConfig(access_token="tok", project="my-proj", location="us-central1")
+
+    rejection: Final = cfg.setup_session_update_rejection("gemini-live-2.5-flash", session, setup_sent=False)
+
+    assert rejection is not None
+    assert (rejection.code, rejection.param) == ("invalid_value", param)
+
+
+@pytest.mark.parametrize("voice", ["Kore", "alloy"])
+def test_vertex_accepts_documented_and_openai_stock_voices_on_the_session_update(voice):
+    cfg: Final = VertexAIRealtimeConfig(access_token="tok", project="my-proj", location="us-central1")
+
+    assert cfg.setup_session_update_rejection("gemini-live-2.5-flash", {"voice": voice}, setup_sent=False) is None
+
+
+def test_vertex_drops_an_undocumented_voice_with_drop_params():
+    cfg: Final = VertexAIRealtimeConfig(access_token="tok", project="my-proj", location="us-central1", drop_params=True)
+    session_update: Final = {"type": "session.update", "session": {"voice": "Nova", "instructions": "Be brief."}}
+
+    assert cfg.setup_session_update_rejection("gemini-live-2.5-flash", session_update["session"], setup_sent=False) is None
+    messages: Final = cfg.transform_realtime_request(
+        json.dumps(session_update), "gemini-live-2.5-flash", session_configuration_request=None
+    )
+    assert "speechConfig" not in json.loads(messages[0])["setup"].get("generationConfig", {})
+
+
+@pytest.mark.parametrize("drop_params", [False, True])
+@pytest.mark.parametrize("voice", ["fable", "onyx"])
+def test_vertex_sends_the_documented_voice_an_openai_tts_voice_maps_to(voice, drop_params):
+    cfg: Final = VertexAIRealtimeConfig(
+        access_token="tok", project="my-proj", location="us-central1", drop_params=drop_params
+    )
+    session_update: Final = {"type": "session.update", "session": {"voice": voice}}
+
+    assert cfg.setup_session_update_rejection("gemini-live-2.5-flash", session_update["session"], setup_sent=False) is None
+    messages: Final = cfg.transform_realtime_request(
+        json.dumps(session_update), "gemini-live-2.5-flash", session_configuration_request=None
+    )
+    assert json.loads(messages[0])["setup"]["generationConfig"]["speechConfig"] == {
+        "voiceConfig": {"prebuiltVoiceConfig": {"voiceName": OPENAI_TO_GEMINI_TTS_VOICES[voice]}}
+    }

ui/litellm-dashboard/src/app/(dashboard)/playground/components/chat_ui/RealtimePlayground.test.tsx

@@ -173,6 +173,24 @@ describe("RealtimePlayground", () => {
     expect(update.session.type).toBe("realtime");
   });
 
+  it("starts recording on the configuration the session was created with", async () => {
+    const user = userEvent.setup();
+    render(<RealtimePlayground {...props} />);
+
+    await connect(user);
+    await act(async () => {
+      latestSocket().emit({ type: "session.created" });
+    });
+    await user.click(screen.getByTitle("Start recording"));
+
+    const updates = latestSocket()
+      .sent.map((raw) => JSON.parse(raw))
+      .filter((payload) => payload.type === "session.update");
+    expect(updates).toHaveLength(2);
+    expect(updates[1]).toEqual(updates[0]);
+    expect(updates[0].session.turn_detection).toEqual({ type: "server_vad" });
+  });
+
   it("sends what was typed and then asks for a response", async () => {
     const user = userEvent.setup();
     render(<RealtimePlayground {...props} />);

ui/litellm-dashboard/src/app/(dashboard)/playground/components/chat_ui/RealtimePlayground.tsx

@@ -22,6 +22,20 @@ interface RealtimePlaygroundProps {
   selectedGuardrails?: string[];
 }
 
+const sessionUpdate = (voice: string) =>
+  JSON.stringify({
+    type: "session.update",
+    session: {
+      type: "realtime",
+      modalities: ["text", "audio"],
+      voice,
+      input_audio_format: "pcm16",
+      output_audio_format: "pcm16",
+      input_audio_transcription: { model: "gpt-4o-mini-transcribe" },
+      turn_detection: { type: "server_vad" },
+    },
+  });
+
 const RealtimePlayground: React.FC<RealtimePlaygroundProps> = ({
   accessToken,
   selectedModel,
@@ -125,21 +139,7 @@ const RealtimePlayground: React.FC<RealtimePlaygroundProps> = ({
           const type = data.type;
 
           if (type === "session.created") {
-            // GA: session.type is required ("realtime" | "transcription")
-            ws.send(
-              JSON.stringify({
-                type: "session.update",
-                session: {
-                  type: "realtime",
-                  modalities: ["text", "audio"],
-                  voice: selectedVoice,
-                  input_audio_format: "pcm16",
-                  output_audio_format: "pcm16",
-                  input_audio_transcription: { model: "gpt-4o-mini-transcribe" },
-                  turn_detection: null,
-                },
-              }),
-            );
+            ws.send(sessionUpdate(selectedVoice));
           } else if (type === "session.updated") {
             // session configured
           } else if (
@@ -231,22 +231,7 @@ const RealtimePlayground: React.FC<RealtimePlaygroundProps> = ({
   const startRecording = useCallback(async () => {
     if (!wsRef.current || wsRef.current.readyState !== WebSocket.OPEN) return;
 
-    // Switch to server VAD mode for voice input
-    // GA: session.type is required
-    wsRef.current.send(
-      JSON.stringify({
-        type: "session.update",
-        session: {
-          type: "realtime",
-          modalities: ["text", "audio"],
-          voice: selectedVoice,
-          input_audio_format: "pcm16",
-          output_audio_format: "pcm16",
-          input_audio_transcription: { model: "gpt-4o-mini-transcribe" },
-          turn_detection: { type: "server_vad" },
-        },
-      }),
-    );
+    wsRef.current.send(sessionUpdate(selectedVoice));
 
     try {
       const stream = await navigator.mediaDevices.getUserMedia({ audio: true });
@@ -301,7 +286,7 @@ const RealtimePlayground: React.FC<RealtimePlaygroundProps> = ({
     } catch (err: any) {
       addMessage("status", `Microphone error: ${err.message}`);
     }
-  }, [addMessage]);
+  }, [addMessage, selectedVoice]);
 
   const stopRecording = useCallback(() => {
     processorRef.current?.disconnect();
@@ -317,21 +302,7 @@ const RealtimePlayground: React.FC<RealtimePlaygroundProps> = ({
     if (!wsRef.current || wsRef.current.readyState !== WebSocket.OPEN) return;
     if (configureSessionRef.current) return;
     configureSessionRef.current = true;
-    // GA: session.type is required
-    wsRef.current.send(
-      JSON.stringify({
-        type: "session.update",
-        session: {
-          type: "realtime",
-          modalities: ["text", "audio"],
-          voice: selectedVoice,
-          input_audio_format: "pcm16",
-          output_audio_format: "pcm16",
-          input_audio_transcription: { model: "gpt-4o-mini-transcribe" },
-          turn_detection: null,
-        },
-      }),
-    );
+    wsRef.current.send(sessionUpdate(selectedVoice));
   }, [selectedVoice]);
 
   const sendTextMessage = useCallback(() => {