diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index ea4af565d7a..2f2a9ca334a 100644 --- a/litellm/litellm_core_utils/realtime_streaming.py +++ b/litellm/litellm_core_utils/realtime_streaming.py @@ -877,6 +877,31 @@ class RealTimeStreaming: self._pending_guardrail_message = None continue + ## GUARDRAIL: Inject turn_detection into first session.update + # if needed. Done BEFORE the GA remap so the injected + # ``create_response`` rides along with any client-provided + # turn_detection fields (e.g. silence_duration_ms) into the + # nested ``audio.input.turn_detection`` path produced by the + # remap. Doing this after the remap would create a separate + # minimal root-level ``turn_detection`` and silently drop + # the client's nested settings. + if ( + msg_type == "session.update" + and self.session_configuration_request is None + and not self._guardrail_turn_detection_update_sent + and self._has_audio_transcription_guardrails() + ): + session = msg_obj.setdefault("session", {}) + if isinstance(session, dict): + session.setdefault("turn_detection", {})[ + "create_response" + ] = False + message = json.dumps(msg_obj) + self._guardrail_turn_detection_update_sent = True + verbose_logger.debug( + "Injected turn_detection into first session.update for audio transcription guardrails" + ) + # GA compatibility: remap beta-style session fields only when # the upstream is in GA mode. Beta upstreams expect the flat # session shape unchanged. @@ -893,28 +918,6 @@ class RealTimeStreaming: except (json.JSONDecodeError, AttributeError): pass - ## GUARDRAIL: Inject turn_detection into first session.update if needed - try: - msg_obj = json.loads(message) - if ( - msg_obj.get("type") == "session.update" - and self.session_configuration_request is None - and not self._guardrail_turn_detection_update_sent - and self._has_audio_transcription_guardrails() - ): - # Inject turn_detection into the first session.update - session = msg_obj.setdefault("session", {}) - session.setdefault("turn_detection", {})[ - "create_response" - ] = False - message = json.dumps(msg_obj) - self._guardrail_turn_detection_update_sent = True - verbose_logger.debug( - "Injected turn_detection into first session.update for audio transcription guardrails" - ) - except (json.JSONDecodeError, AttributeError): - pass - ## LOGGING # Log after any in-place modifications (GA remap, guardrail # turn_detection injection) so audit logs reflect what we @@ -922,16 +925,7 @@ class RealTimeStreaming: self.store_input(message=message) ## FORWARD TO BACKEND - if self.provider_config: - message = self.provider_config.transform_realtime_request( - message, self.model, self.session_configuration_request - ) - - for msg in message: - await self.backend_ws.send(msg) # type: ignore[union-attr] - self._cache_session_configuration_request(msg) - else: - await self.backend_ws.send(message) # type: ignore[union-attr] + await self._send_to_backend(message) except Exception as e: verbose_logger.debug(f"Error in client ack messages: {e}") diff --git a/litellm/llms/vertex_ai/realtime/transformation.py b/litellm/llms/vertex_ai/realtime/transformation.py index 94fdf82a6f7..555b04ab883 100644 --- a/litellm/llms/vertex_ai/realtime/transformation.py +++ b/litellm/llms/vertex_ai/realtime/transformation.py @@ -204,7 +204,7 @@ class VertexAIRealtimeConfig(GeminiRealtimeConfig): if session_configuration_request is None: # First session.update - send the setup with Vertex AI configuration setup_config = self._build_vertex_ai_setup_config( - model, json_message["session"] + model, json_message.get("session") or {} ) gemini_setup_msg = json.dumps({"setup": setup_config}) diff --git a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py index b4a7a05a136..c62ef6e23b0 100644 --- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py +++ b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py @@ -1353,11 +1353,18 @@ async def test_guardrail_turn_detection_injected_into_first_session_update_defer # Simulate first session.update in deferred mode await streaming.client_ack_messages() - # Verify turn_detection was injected into the session.update + # Verify turn_detection was injected into the session.update. The + # injection runs before the GA remap, so the create_response flag ends + # up nested under audio.input.turn_detection in the GA-shaped payload. assert len(transformed_messages) == 1 transformed_msg, session_config = transformed_messages[0] msg_obj = json.loads(transformed_msg) assert msg_obj["type"] == "session.update" - assert "turn_detection" in msg_obj["session"] - assert msg_obj["session"]["turn_detection"]["create_response"] is False + session_obj = msg_obj["session"] + injected_turn_detection = ( + session_obj.get("turn_detection") + or session_obj.get("audio", {}).get("input", {}).get("turn_detection") + ) + assert injected_turn_detection is not None + assert injected_turn_detection["create_response"] is False assert streaming._guardrail_turn_detection_update_sent is True