fix(realtime): force create_response=False in all client session.update turn_detection when audio guardrails active

Prevents a client from re-enabling Gemini/GA VAD auto-response (and thereby
bypassing the audio transcription guardrail) by sending a later
session.update with turn_detection.create_response: true.

litellm/litellm_core_utils/realtime_streaming.py

@@ -911,6 +911,42 @@ class RealTimeStreaming:
                                 "Injected turn_detection into first session.update for audio transcription guardrails"
                             )
 
+                    ## GUARDRAIL: Force ``create_response`` to False in any
+                    # client-provided ``turn_detection`` so a later
+                    # ``session.update`` cannot re-enable VAD auto-response
+                    # and bypass the transcription guardrail after the
+                    # initial disable. Covers both the flat beta key and the
+                    # nested GA ``audio.input.turn_detection`` shape, since
+                    # the GA remap below also accepts either form.
+                    if (
+                        msg_type == "session.update"
+                        and self._has_audio_transcription_guardrails()
+                    ):
+                        session = msg_obj.get("session")
+                        if isinstance(session, dict):
+                            td_overridden = False
+                            flat_td = session.get("turn_detection")
+                            if (
+                                isinstance(flat_td, dict)
+                                and flat_td.get("create_response") is not False
+                            ):
+                                flat_td["create_response"] = False
+                                td_overridden = True
+                            audio = session.get("audio")
+                            if isinstance(audio, dict):
+                                audio_input = audio.get("input")
+                                if isinstance(audio_input, dict):
+                                    nested_td = audio_input.get("turn_detection")
+                                    if (
+                                        isinstance(nested_td, dict)
+                                        and nested_td.get("create_response")
+                                        is not False
+                                    ):
+                                        nested_td["create_response"] = False
+                                        td_overridden = True
+                            if td_overridden:
+                                message = json.dumps(msg_obj)
+
                     # GA compatibility: remap beta-style session fields only when
                     # the upstream is in GA mode. Beta upstreams expect the flat
                     # session shape unchanged.

tests/test_litellm/litellm_core_utils/test_realtime_streaming.py

@@ -1426,3 +1426,75 @@ async def test_guardrail_turn_detection_injection_tolerates_non_dict_value(
     assert isinstance(injected_turn_detection, dict)
     assert injected_turn_detection["create_response"] is False
     assert streaming._guardrail_turn_detection_update_sent is True
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "client_session",
+    [
+        {"turn_detection": {"type": "server_vad", "create_response": True}},
+        {
+            "audio": {
+                "input": {
+                    "turn_detection": {"type": "server_vad", "create_response": True}
+                }
+            }
+        },
+    ],
+)
+async def test_subsequent_session_update_cannot_reenable_vad_when_guardrails_active(
+    client_session,
+):
+    """A subsequent client session.update must not be allowed to flip
+    ``create_response`` back to True once audio transcription guardrails have
+    disabled VAD auto-response. Covers both the flat beta shape and the
+    nested GA ``audio.input.turn_detection`` shape.
+    """
+    client_ws = AsyncMock()
+    client_ws.receive_text = AsyncMock(
+        side_effect=[
+            json.dumps({"type": "session.update", "session": client_session}),
+            ConnectionClosed(None, None),
+        ]
+    )
+    backend_ws = MagicMock()
+    backend_ws.send = AsyncMock()
+
+    logging_obj = MagicMock()
+    logging_obj.litellm_trace_id = "trace_1"
+    logging_obj.async_success_handler = AsyncMock()
+    logging_obj.success_handler = MagicMock()
+
+    provider_config = MagicMock()
+    transformed_messages = []
+
+    def mock_transform(msg, model, session_config):
+        transformed_messages.append((msg, session_config))
+        return [msg]
+
+    provider_config.transform_realtime_request = MagicMock(side_effect=mock_transform)
+
+    streaming = RealTimeStreaming(
+        websocket=client_ws,
+        backend_ws=backend_ws,
+        logging_obj=logging_obj,
+        provider_config=provider_config,
+        model="gemini-2.5-flash",
+    )
+    streaming._has_audio_transcription_guardrails = MagicMock(return_value=True)  # type: ignore[method-assign]
+    # Simulate that initial setup + guardrail disable have already happened.
+    streaming.session_configuration_request = json.dumps({"setup": {"model": "x"}})
+    streaming._guardrail_turn_detection_update_sent = True
+
+    await streaming.client_ack_messages()
+
+    assert len(transformed_messages) == 1
+    forwarded_msg, _ = transformed_messages[0]
+    msg_obj = json.loads(forwarded_msg)
+    session_obj = msg_obj["session"]
+    forwarded_turn_detection = (
+        session_obj.get("turn_detection")
+        or session_obj.get("audio", {}).get("input", {}).get("turn_detection")
+    )
+    assert isinstance(forwarded_turn_detection, dict)
+    assert forwarded_turn_detection["create_response"] is False