fix(realtime): stop second Gemini Live setup, retry hung handshake, close guardrail bypass (#31519)

* fix(realtime): stop sending a second Gemini Live setup on follow-up session.update

Gemini Live (BidiGenerateContent) accepts setup as the first-and-only client
message; a second setup closes the socket with 1007 Request contains an invalid
argument. The AI Studio Gemini path forwarded every client session.update after
the first as a follow-up setup, and GA clients (pipecat) send several while
configuring the session, so the second one tore the session down before the
first turn. Callers saw silence after the first response, exponential per-turn
latency from reconnect/retry churn, and intermittent 1011 errors.

Drop subsequent session.updates instead of resending setup, matching what the
Vertex subclass already does. Tools and instructions must ride on the first
session.update before any conversation content.

Adds regression tests covering the plain follow-up, a follow-up that adds tools
(the case the previous identical-only dedup still forwarded), and the guardrail
create_response=False warning path.

* fix(realtime): retry the backend open handshake instead of failing with 1011

The upstream Live API open handshake (e.g. Gemini Live) intermittently hangs;
waiting longer never recovers a hung attempt, but a fresh attempt almost always
connects in ~1s. The proxy opened the backend websocket once with the default
open_timeout and no retry, so a single slow handshake surfaced to the caller as
a fatal 1011 internal error and dropped the call.

Bound each open attempt with a short open_timeout and retry; a bounded attempt
that already timed out spaces out the next try, so no backoff is needed.
Deterministic handshake-status rejections (auth/4xx) are not retried, and the
retry only ever wraps the open, never a live session.

Adds tests for retry-then-succeed, raise-after-max-attempts, and
no-retry-on-auth-failure.

* fix(realtime): close guardrail bypass + surface handshake status; drop obsolete tests

Three review fixes on the Gemini Live realtime path.

Transcription-guardrail bypass: Gemini Live rejects a second setup (1007), so once
the initial setup is sent the guardrail's automaticActivityDetection.disabled=true
can no longer be delivered as a follow-up session.update. With that follow-up now
dropped, the model's auto-response stayed enabled and a realtime_input_transcription
guardrail was bypassed (the model answered before the proxy could gate the turn).
Fold the disable into the one-and-only setup instead: the handler injects it into
the auto-sent setup (gemini_live_defer_setup false) and _send_to_backend injects it
into the deferred first setup. OpenAI sessions accept follow-up updates and are left
untouched.

Backend handshake status: the open-retry treated only InvalidStatusCode as
deterministic; websockets>=15 raises InvalidStatus for a rejected client handshake,
so a 401/403 fell into the broad WebSocketException branch and was retried before
the caller closed the client with 1011 instead of the upstream status. Treat both as
non-retryable.

Obsolete tests: the four tests asserting a follow-up session.update is merged and
re-sent as a second setup asserted behavior that crashes Gemini Live with 1007
(verified directly against the API). Removed; the drop is covered by new regression
tests.

* style(realtime): reformat changed files to ruff line-length 120

Post-merge with litellm_internal_staging, which unified ruff format width to 120
(#31518). The realtime change set was formatted at 88, so the changed lines
tripped the whole-file ruff format check. Reformat with ruff 0.15.3 at the repo's
120 width; no logic changes.

(cherry picked from commit ef5d05f137)

litellm/litellm_core_utils/realtime_streaming.py

@@ -357,6 +357,7 @@ class RealTimeStreaming:
                 # send, causing subsequent client session.update messages to
                 # be treated as "subsequent" and dropped even though the
                 # backend never received the original setup.
+                msg = self._maybe_inject_guardrail_auto_response_disable(msg)
                 await self.backend_ws.send(msg)  # type: ignore[union-attr, attr-defined]
                 self._cache_session_configuration_request(msg)
                 sent = True
@@ -617,6 +618,39 @@ class RealTimeStreaming:
         if sent:
             self._guardrail_turn_detection_update_sent = True
 
+    def _maybe_inject_guardrail_auto_response_disable(self, setup_message: str) -> str:
+        """Fold the transcription-guardrail auto-response disable into the setup.
+
+        Gemini/Vertex Live reject a second ``setup`` (1007), so the guardrail's
+        ``automaticActivityDetection.disabled=true`` cannot be delivered as a
+        follow-up session.update; it must live in the one-and-only setup, or a
+        ``realtime_input_transcription`` guardrail is bypassed (the model
+        auto-responds before the proxy can gate the turn). Applies only to the
+        bidi ``setup`` shape; OpenAI sessions accept follow-up updates and so are
+        left untouched (handled by ``_maybe_send_guardrail_turn_detection_update``).
+        """
+        if self._guardrail_turn_detection_update_sent:
+            return setup_message
+        if not self._has_audio_transcription_guardrails():
+            return setup_message
+        try:
+            obj = json.loads(setup_message)
+        except (json.JSONDecodeError, TypeError):
+            return setup_message
+        setup = obj.get("setup") if isinstance(obj, dict) else None
+        if not isinstance(setup, dict):
+            return setup_message
+        automatic = setup.setdefault("realtimeInputConfig", {}).setdefault(
+            "automaticActivityDetection", {}
+        )
+        automatic["disabled"] = True
+        self._guardrail_turn_detection_update_sent = True
+        verbose_logger.debug(
+            "Realtime: folded automaticActivityDetection.disabled=true into setup "
+            "for transcription-guardrail gating"
+        )
+        return json.dumps(obj)
+
     def _has_realtime_guardrails_for_event_hooks(
         self,
         event_hooks: List[Any],

litellm/llms/custom_httpx/llm_http_handler.py

@@ -5457,6 +5457,58 @@ class BaseLLMHTTPHandler:
         new_query = parsed.query + ("&" if parsed.query else "") + urlencode(extras)
         return urlunparse(parsed._replace(query=new_query))
 
+    @staticmethod
+    async def _open_realtime_backend_ws(
+        websockets_module: Any,
+        url: str,
+        headers: dict,
+        ssl_context: Any,
+        *,
+        open_timeout: float = 8.0,
+        max_attempts: int = 3,
+    ) -> Any:
+        """Open the backend realtime websocket, retrying a hung open handshake.
+
+        The upstream Live handshake (e.g. Gemini Live) intermittently hangs on
+        open; waiting longer never recovers a hung attempt, but a fresh attempt
+        almost always connects in ~1s. So bound each attempt with ``open_timeout``
+        and retry, instead of surfacing one slow handshake to the caller as a
+        fatal 1011. A bounded attempt that timed out already spaced out the
+        retry, so no extra backoff is needed. Deterministic rejections (auth /
+        handshake status) are not retried.
+        """
+        # Handshake-status rejections are deterministic (auth / 4xx): retrying
+        # cannot help and the caller must see the upstream status, not a generic
+        # 1011. websockets <15 raises InvalidStatusCode, >=15 raises InvalidStatus.
+        deterministic_errors = tuple(
+            exc
+            for exc in (
+                getattr(websockets_module.exceptions, "InvalidStatus", None),
+                getattr(websockets_module.exceptions, "InvalidStatusCode", None),
+            )
+            if exc is not None
+        )
+        last_exc: Optional[BaseException] = None
+        for _ in range(max_attempts):
+            try:
+                return await websockets_module.connect(
+                    url,
+                    additional_headers=headers,
+                    max_size=REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES,
+                    ssl=ssl_context,
+                    open_timeout=open_timeout,
+                )
+            except deterministic_errors:
+                raise
+            except (
+                TimeoutError,
+                OSError,
+                websockets_module.exceptions.WebSocketException,
+            ) as e:
+                last_exc = e
+        assert last_exc is not None  # loop only exits via return or a captured exc
+        raise last_exc
+
     async def async_realtime(
         self,
         model: str,
@@ -5491,22 +5543,10 @@ class BaseLLMHTTPHandler:
                 ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
                 ssl_context.check_hostname = False
                 ssl_context.verify_mode = ssl.CERT_NONE
-            async with websockets.connect(  # type: ignore
-                url,
-                additional_headers=headers,
-                max_size=REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES,
-                ssl=ssl_context,
-            ) as backend_ws:
-                # Auto-send session setup if the provider requires it
-                # (e.g. Gemini/Vertex AI Live needs a `setup` message before any realtime_input)
-                _session_config: Optional[str] = None
-                if provider_config.requires_session_configuration():
-                    _session_config = provider_config.session_configuration_request(
-                        model
-                    )
-                    if _session_config:
-                        await backend_ws.send(_session_config)
-
+            backend_ws = await self._open_realtime_backend_ws(
+                websockets, url, headers, ssl_context
+            )
+            async with backend_ws:
                 _request_data: Dict[str, Any] = {}
                 if litellm_metadata:
                     _request_data["litellm_metadata"] = litellm_metadata
@@ -5524,8 +5564,26 @@ class BaseLLMHTTPHandler:
                         else None
                     ),
                 )
-                if _session_config:
-                    realtime_streaming.session_configuration_request = _session_config
+
+                # Auto-send session setup if the provider requires it (e.g.
+                # Gemini/Vertex AI Live needs a `setup` before any realtime_input).
+                # Build the streaming handler first so a transcription guardrail's
+                # auto-response disable can be folded into this one setup: Gemini
+                # rejects a second setup, so a follow-up disable would be dropped
+                # and the guardrail bypassed.
+                _session_config: Optional[str] = None
+                if provider_config.requires_session_configuration():
+                    _session_config = provider_config.session_configuration_request(
+                        model
+                    )
+                    if _session_config:
+                        _session_config = (
+                            realtime_streaming._maybe_inject_guardrail_auto_response_disable(
+                                _session_config
+                            )
+                        )
+                        await backend_ws.send(_session_config)
+                        realtime_streaming.session_configuration_request = _session_config
 
                 # For providers that defer setup until client session.update, optionally
                 # send synthetic session.created to unblock clients waiting on connect.

litellm/llms/gemini/realtime/transformation.py

@@ -433,17 +433,11 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
         Handle session.update by sending setup to Gemini.
 
         On the FIRST session.update (when session_configuration_request is None),
-        the full setup with all configuration is sent.
-
-        Subsequent session.update messages are forwarded as a follow-up setup
-        with the new fields merged into the original setup. Gemini Live treats
-        a follow-up BidiGenerateContentSetup as a full session replacement
-        rather than a partial merge, so we carry forward the previous setup
-        (tools, generationConfig, inputAudioTranscription, systemInstruction,
-        ...) and overlay the new fields on top. This preserves the old
-        behavior where clients could refine the session via session.update
-        (e.g. add tools after the auto-setup on connect), and also keeps the
-        guardrail-driven turn_detection update working.
+        the full setup with all configuration is sent. Gemini Live accepts setup
+        as the first-and-only client message, so every later session.update is
+        dropped rather than forwarded as a second setup (which Gemini rejects
+        with a 1007, tearing the session down). To carry tools/instructions, send
+        them on the first session.update before any conversation content.
         """
         session_payload = json_message.get("session") or {}
         # Normalize GA-remapped fields (``output_modalities``,
@@ -472,82 +466,32 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
                 )
             ]
 
-        if not new_overrides:
+        # Gemini Live accepts exactly one ``setup`` message: the first and only
+        # client message. A second ``setup`` closes the socket with
+        # ``1007 Request contains an invalid argument``, so a session.update
+        # after the initial setup must not be forwarded as a follow-up setup.
+        # Every GA client (pipecat included) sends several session.updates while
+        # configuring the session; forwarding a second one tears the session down
+        # before the first turn, which surfaces to callers as silence after the
+        # first response, reconnect/retry latency churn, and 1011 errors. Drop
+        # it. The Vertex subclass already drops subsequent setups for this exact
+        # reason; the constraint is identical on AI Studio.
+        client_turn_detection = self._extract_turn_detection(session_payload)
+        if (
+            isinstance(client_turn_detection, dict)
+            and client_turn_detection.get("create_response") is False
+        ):
+            verbose_logger.warning(
+                "Gemini Realtime: Dropping subsequent session.update "
+                "(turn_detection.create_response=False) — Gemini Live rejects a "
+                "second setup message, so audio-transcription guardrails cannot "
+                "suppress the model's auto-response mid-session."
+            )
+        else:
             verbose_logger.debug(
-                "Gemini Realtime: Ignoring session.update (no mappable fields)"
+                "Gemini Realtime: Ignoring session.update (setup already sent)"
             )
-            return []
-
-        try:
-            original_setup = cast(
-                BidiGenerateContentSetup,
-                json.loads(session_configuration_request).get("setup", {}),
-            )
-        except (json.JSONDecodeError, AttributeError):
-            original_setup = {}
-
-        # Deep-merge ``generationConfig`` and ``realtimeInputConfig`` so a
-        # partial session.update (e.g. only ``temperature`` or only
-        # ``modalities``) does not silently drop unrelated sub-keys
-        # (``responseModalities``, ``maxOutputTokens``, ...) from the original
-        # setup.
-        follow_up_setup: BidiGenerateContentSetup = {
-            **original_setup,
-            **new_overrides,
-            "model": f"models/{model}",
-        }
-        original_generation_config = original_setup.get("generationConfig")
-        new_generation_config = new_overrides.get("generationConfig")
-        if isinstance(original_generation_config, dict) and isinstance(
-            new_generation_config, dict
-        ):
-            follow_up_setup["generationConfig"] = {
-                **original_generation_config,
-                **new_generation_config,
-            }
-        original_realtime_input_config = original_setup.get("realtimeInputConfig")
-        new_realtime_input_config = new_overrides.get("realtimeInputConfig")
-        if isinstance(original_realtime_input_config, dict) and isinstance(
-            new_realtime_input_config, dict
-        ):
-            merged_realtime_input_config = {
-                **original_realtime_input_config,
-                **new_realtime_input_config,
-            }
-            # Deep-merge ``automaticActivityDetection`` so a partial VAD
-            # update (e.g. the guardrail-injected ``disabled: True`` from
-            # ``create_response: False``) does not silently drop unrelated
-            # knobs like ``silenceDurationMs`` / ``prefixPaddingMs`` from
-            # the original setup.
-            original_automatic_activity_detection = original_realtime_input_config.get(
-                "automaticActivityDetection"
-            )
-            new_automatic_activity_detection = new_realtime_input_config.get(
-                "automaticActivityDetection"
-            )
-            if isinstance(original_automatic_activity_detection, dict) and isinstance(
-                new_automatic_activity_detection, dict
-            ):
-                merged_realtime_input_config["automaticActivityDetection"] = {
-                    **original_automatic_activity_detection,
-                    **new_automatic_activity_detection,
-                }
-            follow_up_setup["realtimeInputConfig"] = cast(
-                BidiGenerateContentRealtimeInputConfig,
-                merged_realtime_input_config,
-            )
-        verbose_logger.debug(
-            "Gemini Realtime: Forwarding session.update as follow-up setup"
-        )
-        return [
-            json.dumps(
-                {
-                    "setup": self._finalize_gemini_live_setup(
-                        model, cast(Dict[str, Any], follow_up_setup)
-                    )
-                }
-            )
-        ]
+        return []
 
     def _handle_conversation_item(self, json_message: dict) -> List[str]:
         """

tests/test_litellm/litellm_core_utils/test_realtime_streaming.py

@@ -2681,3 +2681,72 @@ async def test_deferred_setup_clear_drops_appends_when_buffered():
     streaming._buffer_pending_message_until_setup(new_audio)
 
     assert streaming._pending_messages_until_setup == [new_audio]
+
+
+def _transcription_guardrail():
+    """A minimal real CustomGuardrail registered for the realtime transcript hook."""
+    from litellm.integrations.custom_guardrail import CustomGuardrail
+    from litellm.types.guardrails import GuardrailEventHooks
+
+    class _TranscriptionGuardrail(CustomGuardrail):
+        async def apply_guardrail(
+            self, inputs, request_data, input_type, logging_obj=None
+        ):
+            return inputs
+
+    return _TranscriptionGuardrail(
+        guardrail_name="test_transcription_guard",
+        event_hook=GuardrailEventHooks.realtime_input_transcription,
+        default_on=True,
+    )
+
+
+def test_setup_folds_in_auto_response_disable_when_transcription_guardrail_active():
+    """Gemini rejects a second setup, so a transcription guardrail's auto-response
+    disable must be folded into the one-and-only setup; otherwise the model
+    auto-responds and the guardrail is bypassed."""
+    import litellm
+
+    litellm.callbacks = [_transcription_guardrail()]
+    try:
+        streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock())
+        setup = json.dumps(
+            {
+                "setup": {
+                    "model": "models/gemini-3.1-flash-live-preview",
+                    "generationConfig": {"responseModalities": ["AUDIO"]},
+                    "inputAudioTranscription": {},
+                }
+            }
+        )
+        out = json.loads(streaming._maybe_inject_guardrail_auto_response_disable(setup))
+        aad = out["setup"]["realtimeInputConfig"]["automaticActivityDetection"]
+        assert aad["disabled"] is True
+    finally:
+        litellm.callbacks = []
+
+
+def test_setup_unchanged_without_transcription_guardrail():
+    import litellm
+
+    litellm.callbacks = []
+    streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock())
+    setup = json.dumps(
+        {"setup": {"model": "x", "generationConfig": {"responseModalities": ["AUDIO"]}}}
+    )
+    out = streaming._maybe_inject_guardrail_auto_response_disable(setup)
+    assert json.loads(out) == json.loads(setup)
+
+
+def test_non_bidi_setup_left_untouched_for_followup_capable_providers():
+    """OpenAI realtime accepts a follow-up session.update, so a non-bidi message
+    (no top-level 'setup' key) must be left untouched even with a guardrail on."""
+    import litellm
+
+    litellm.callbacks = [_transcription_guardrail()]
+    try:
+        streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock())
+        msg = json.dumps({"type": "session.update", "session": {"instructions": "hi"}})
+        assert streaming._maybe_inject_guardrail_auto_response_disable(msg) == msg
+    finally:
+        litellm.callbacks = []

tests/test_litellm/llms/custom_httpx/test_llm_http_handler.py

@@ -1051,3 +1051,96 @@ def test_async_compact_handler_sends_json_when_not_signed():
     )
     assert kwargs.get("json") == {"model": "openai.gpt-5.5", "input": "hi"}
     assert "data" not in kwargs
+
+
+class _FakeWSExceptions:
+    class WebSocketException(Exception):
+        pass
+
+    class InvalidStatusCode(WebSocketException):
+        def __init__(self) -> None:
+            super().__init__("HTTP 403")
+
+    # websockets>=15 raises InvalidStatus (not InvalidStatusCode) for a rejected
+    # client handshake; both must be treated as deterministic.
+    class InvalidStatus(WebSocketException):
+        def __init__(self) -> None:
+            super().__init__("HTTP 401")
+
+
+class _FakeWebsocketsModule:
+    """Stand-in for the ``websockets`` module so the realtime backend-open retry
+    can be exercised without a real network handshake (dependency injection,
+    no monkeypatching)."""
+
+    def __init__(self, outcomes):
+        # outcomes: list where each item is either an Exception to raise or a
+        # sentinel object to return as the "connected" websocket.
+        self._outcomes = list(outcomes)
+        self.exceptions = _FakeWSExceptions
+        self.attempts = 0
+        self.open_timeouts: list = []
+
+    async def connect(self, *args, **kwargs):
+        self.attempts += 1
+        self.open_timeouts.append(kwargs.get("open_timeout"))
+        outcome = self._outcomes.pop(0)
+        if isinstance(outcome, Exception):
+            raise outcome
+        return outcome
+
+
+@pytest.mark.asyncio
+async def test_realtime_backend_open_retries_then_succeeds():
+    """A hung/slow open handshake is retried; a later fresh attempt connects.
+
+    Regression for intermittent ``1011 timed out during opening handshake``:
+    the proxy used to surface a single slow upstream handshake to the caller as
+    a fatal 1011 with no retry.
+    """
+    sentinel = object()
+    fake = _FakeWebsocketsModule(
+        [TimeoutError("timed out during opening handshake"), sentinel]
+    )
+
+    result = await BaseLLMHTTPHandler._open_realtime_backend_ws(
+        fake, "wss://backend.example/live", {"Authorization": "Bearer x"}, None
+    )
+
+    assert result is sentinel
+    assert fake.attempts == 2
+    # Each attempt must be bounded by a finite open_timeout (not the default/None).
+    assert all(t is not None and t > 0 for t in fake.open_timeouts)
+
+
+@pytest.mark.asyncio
+async def test_realtime_backend_open_raises_after_max_attempts():
+    """When every attempt times out, the final error propagates (so the caller
+    still closes the client socket) rather than looping forever."""
+    fake = _FakeWebsocketsModule([TimeoutError("hang")] * 2)
+
+    with pytest.raises(TimeoutError):
+        await BaseLLMHTTPHandler._open_realtime_backend_ws(
+            fake, "wss://backend.example/live", {}, None, max_attempts=2
+        )
+
+    assert fake.attempts == 2
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "rejection",
+    [_FakeWSExceptions.InvalidStatusCode, _FakeWSExceptions.InvalidStatus],
+)
+async def test_realtime_backend_open_does_not_retry_auth_failure(rejection):
+    """A deterministic handshake-status rejection (auth/4xx) must not be retried;
+    retrying cannot help and the upstream status must surface, not a 1011. Both
+    the websockets<15 (InvalidStatusCode) and >=15 (InvalidStatus) shapes apply."""
+    fake = _FakeWebsocketsModule([rejection()])
+
+    with pytest.raises(_FakeWSExceptions.WebSocketException):
+        await BaseLLMHTTPHandler._open_realtime_backend_ws(
+            fake, "wss://backend.example/live", {}, None
+        )
+
+    assert fake.attempts == 1

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

@@ -1145,60 +1145,6 @@ def test_gemini_function_call_output_includes_name():
     assert "response" in function_response
 
 
-def test_gemini_subsequent_session_update_forwards_tools_merged_with_original_setup():
-    """A client session.update sent after the auto-setup must forward tools/
-    instructions as a follow-up setup, merged with the original setup so we
-    don't drop the pre-existing config (model, generationConfig, etc.)."""
-    config = GeminiRealtimeConfig()
-
-    original_setup = {
-        "setup": {
-            "model": "models/gemini-2.5-flash-native-audio",
-            "generationConfig": {"responseModalities": ["AUDIO"]},
-            "inputAudioTranscription": {},
-            "systemInstruction": {"role": "user", "parts": [{"text": "original"}]},
-        }
-    }
-
-    session_update = {
-        "type": "session.update",
-        "session": {
-            "tools": [
-                {
-                    "type": "function",
-                    "function": {
-                        "name": "get_weather",
-                        "description": "Get weather.",
-                        "parameters": {
-                            "type": "object",
-                            "properties": {"location": {"type": "string"}},
-                            "required": ["location"],
-                        },
-                    },
-                }
-            ],
-            "instructions": "Be concise.",
-        },
-    }
-
-    messages = config.transform_realtime_request(
-        json.dumps(session_update),
-        "gemini-2.5-flash-native-audio",
-        session_configuration_request=json.dumps(original_setup),
-    )
-
-    assert len(messages) == 1
-    follow_up = json.loads(messages[0])["setup"]
-    assert "tools" in follow_up
-    assert follow_up["tools"][0]["function_declarations"][0]["name"] == "get_weather"
-    # systemInstruction overwritten by client's instructions
-    assert follow_up["systemInstruction"]["parts"][0]["text"] == "Be concise."
-    # Original generationConfig / model / inputAudioTranscription preserved
-    assert follow_up["generationConfig"]["responseModalities"] == ["AUDIO"]
-    assert follow_up["model"] == "models/gemini-2.5-flash-native-audio"
-    assert follow_up["inputAudioTranscription"] == {}
-
-
 def test_gemini_realtime_pipecat_ga_session_voice_and_tools():
     """Pipecat OpenAIRealtimeSessionProperties: output_modalities, nested tools,
     and audio.output.voice (e.g. Kore) must map into Gemini setup."""
@@ -1340,116 +1286,6 @@ def test_gemini_input_audio_buffer_commit_maps_to_activity_end_when_manual_vad()
     assert json.loads(messages[0]) == {"realtimeInput": {"activityEnd": True}}
 
 
-def test_gemini_subsequent_session_update_with_turn_detection_only_preserves_original_tools():
-    """A subsequent session.update carrying only turn_detection (the
-    guardrail-injected disable) must keep the original tools/generationConfig."""
-    config = GeminiRealtimeConfig()
-
-    original_setup = {
-        "setup": {
-            "model": "models/gemini-2.5-flash-native-audio",
-            "generationConfig": {"responseModalities": ["AUDIO"]},
-            "inputAudioTranscription": {},
-            "tools": [
-                {
-                    "function_declarations": [
-                        {"name": "lookup", "description": "x", "parameters": {}}
-                    ]
-                }
-            ],
-        }
-    }
-
-    session_update = {
-        "type": "session.update",
-        "session": {"turn_detection": {"create_response": False}},
-    }
-
-    messages = config.transform_realtime_request(
-        json.dumps(session_update),
-        "gemini-2.5-flash-native-audio",
-        session_configuration_request=json.dumps(original_setup),
-    )
-
-    assert len(messages) == 1
-    follow_up = json.loads(messages[0])["setup"]
-    assert follow_up["tools"] == original_setup["setup"]["tools"]
-    assert (
-        follow_up["realtimeInputConfig"]["automaticActivityDetection"]["disabled"]
-        is True
-    )
-
-
-def test_gemini_follow_up_session_update_preserves_response_modalities_on_partial_generation_config():
-    """A follow-up session.update that only sets `temperature` (or any other
-    generationConfig sub-field) must not wipe `responseModalities` from the
-    original setup."""
-    config = GeminiRealtimeConfig()
-
-    original_setup = {
-        "setup": {
-            "model": "models/gemini-2.5-flash-native-audio",
-            "generationConfig": {
-                "responseModalities": ["AUDIO"],
-                "maxOutputTokens": 2048,
-            },
-            "inputAudioTranscription": {},
-        }
-    }
-
-    session_update = {
-        "type": "session.update",
-        "session": {"temperature": 0.7},
-    }
-
-    messages = config.transform_realtime_request(
-        json.dumps(session_update),
-        "gemini-2.5-flash-native-audio",
-        session_configuration_request=json.dumps(original_setup),
-    )
-
-    follow_up = json.loads(messages[0])["setup"]
-    assert follow_up["generationConfig"]["responseModalities"] == ["AUDIO"]
-    assert follow_up["generationConfig"]["maxOutputTokens"] == 2048
-    assert follow_up["generationConfig"]["temperature"] == 0.7
-
-
-def test_gemini_subsequent_session_update_preserves_automatic_activity_detection_subfields():
-    config = GeminiRealtimeConfig()
-
-    original_setup = {
-        "setup": {
-            "model": "models/gemini-2.5-flash-native-audio",
-            "generationConfig": {"responseModalities": ["AUDIO"]},
-            "realtimeInputConfig": {
-                "automaticActivityDetection": {
-                    "disabled": False,
-                    "silenceDurationMs": 500,
-                    "prefixPaddingMs": 100,
-                }
-            },
-        }
-    }
-
-    session_update = {
-        "type": "session.update",
-        "session": {"turn_detection": {"create_response": False}},
-    }
-
-    messages = config.transform_realtime_request(
-        json.dumps(session_update),
-        "gemini-2.5-flash-native-audio",
-        session_configuration_request=json.dumps(original_setup),
-    )
-
-    automatic_activity_detection = json.loads(messages[0])["setup"][
-        "realtimeInputConfig"
-    ]["automaticActivityDetection"]
-    assert automatic_activity_detection["disabled"] is True
-    assert automatic_activity_detection["silenceDurationMs"] == 500
-    assert automatic_activity_detection["prefixPaddingMs"] == 100
-
-
 def test_gemini_tool_call_id_to_name_evicts_oldest_when_capped():
     """The call_id → name LRU must evict the oldest entry once the cap is
     reached so long sessions with many tool calls don't grow unboundedly,
@@ -1732,3 +1568,106 @@ def test_gemini_in_frame_usage_metadata_clears_pending_buffer():
     assert usage["output_tokens"] == 2
     assert usage["total_tokens"] == 5
     assert config._pending_usage_metadata is None
+
+
+def test_gemini_subsequent_session_update_is_dropped_not_resent_as_setup():
+    """Regression: Gemini Live accepts exactly one ``setup`` message; a second
+    one closes the socket with ``1007 Request contains an invalid argument``.
+
+    Once the initial setup has been sent (``session_configuration_request`` is
+    set), a follow-up session.update must be dropped rather than forwarded as
+    another setup. Forwarding it tore the session down before the first turn,
+    which surfaced to callers as silence after the first response and 1011s.
+    """
+    config = GeminiRealtimeConfig()
+    initial_setup = json.dumps(
+        {
+            "setup": {
+                "model": "models/gemini-2.5-flash",
+                "generationConfig": {"responseModalities": ["AUDIO"]},
+                "inputAudioTranscription": {},
+            }
+        }
+    )
+    follow_up = {
+        "type": "session.update",
+        "session": {"instructions": "Updated instructions", "temperature": 0.4},
+    }
+
+    messages = config.transform_realtime_request(
+        json.dumps(follow_up),
+        "gemini-2.5-flash",
+        session_configuration_request=initial_setup,
+    )
+
+    assert messages == [], (
+        "a session.update after the initial setup must be dropped, never "
+        "forwarded as a second setup (Gemini Live rejects it with 1007)"
+    )
+
+
+def test_gemini_subsequent_session_update_with_new_tools_is_dropped():
+    """Regression: even a follow-up session.update that *differs* from the initial
+    setup (e.g. registers tools after connect) must be dropped.
+
+    The previous dedup only skipped follow-ups identical to the initial setup; a
+    changed one was merged and re-sent as a second setup, still hitting the 1007.
+    Tools must instead ride on the first session.update.
+    """
+    config = GeminiRealtimeConfig()
+    initial_setup = json.dumps(
+        {
+            "setup": {
+                "model": "models/gemini-2.5-flash",
+                "generationConfig": {"responseModalities": ["AUDIO"]},
+            }
+        }
+    )
+    follow_up = {
+        "type": "session.update",
+        "session": {
+            "tools": [
+                {
+                    "type": "function",
+                    "function": {
+                        "name": "terminate_call",
+                        "description": "End the call.",
+                        "parameters": {"type": "object", "properties": {}},
+                    },
+                }
+            ]
+        },
+    }
+
+    messages = config.transform_realtime_request(
+        json.dumps(follow_up),
+        "gemini-2.5-flash",
+        session_configuration_request=initial_setup,
+    )
+
+    assert messages == []
+
+
+def test_gemini_subsequent_guardrail_session_update_dropped_with_warning(caplog):
+    """A dropped follow-up carrying ``turn_detection.create_response=False`` (the
+    transcription-guardrail signal) is still dropped, but warns so operators know
+    the guardrail cannot gate the model's auto-response mid-session on Gemini.
+    """
+    import logging
+
+    config = GeminiRealtimeConfig()
+    initial_setup = json.dumps({"setup": {"model": "models/gemini-2.5-flash"}})
+    follow_up = {
+        "type": "session.update",
+        "session": {"turn_detection": {"type": "server_vad", "create_response": False}},
+    }
+
+    with caplog.at_level(logging.WARNING, logger="LiteLLM"):
+        messages = config.transform_realtime_request(
+            json.dumps(follow_up),
+            "gemini-2.5-flash",
+            session_configuration_request=initial_setup,
+        )
+
+    assert messages == []
+    assert any("Dropping subsequent session.update" in record.message for record in caplog.records)