From 165fedd83ae0ff245fd5e17563946d3d2d09914e Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Fri, 22 May 2026 12:43:45 +0000 Subject: [PATCH] fix(realtime): correct guardrail flag and event-mapping fallback - realtime_streaming: only mark _guardrail_turn_detection_update_sent when the message was actually delivered to the backend. The provider transformation (e.g. Gemini after initial setup) may silently drop session.update; previously we set the flag anyway, falsely claiming the disable was sent and preventing any retry on subsequent session.created events. _send_to_backend now returns whether at least one transformed message was sent. - gemini realtime transformation: avoid shadowing the outer openai_event variable in map_openai_event's fallback loop. With the new toolCall entry now last in MAP_GEMINI_FIELD_TO_OPENAI_EVENT, an unmatched key would otherwise leak FUNCTION_CALL_ARGUMENTS_DONE and skip the ValueError raise. Use a distinct loop variable so the is-None check correctly raises for unknown Gemini messages. Co-authored-by: Yassin Kortam --- .../litellm_core_utils/realtime_streaming.py | 30 ++++++++++++++----- .../llms/gemini/realtime/transformation.py | 8 +++-- 2 files changed, 27 insertions(+), 11 deletions(-) diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index ea912ba62a0..eefa918926a 100644 --- a/litellm/litellm_core_utils/realtime_streaming.py +++ b/litellm/litellm_core_utils/realtime_streaming.py @@ -254,25 +254,32 @@ class RealTimeStreaming: ## SYNC LOGGING executor.submit(self.logging_obj.success_handler(self.messages)) - async def _send_to_backend(self, message: str) -> None: + async def _send_to_backend(self, message: str) -> bool: """Send a message to the backend WebSocket. If a provider_config is set the message is first passed through transform_realtime_request so that provider-specific translation (e.g. dropping session.update for Vertex AI) is applied even for guardrail-injected messages. + + Returns True if at least one message was actually delivered to the + backend, False if the provider transformation produced no output and + the message was effectively dropped. """ if self.provider_config: transformed = self.provider_config.transform_realtime_request( message, self.model, self.session_configuration_request ) + sent = False for msg in transformed: # Cache setup immediately once we send it so concurrent client # session.update messages don't emit duplicate setup packets. self._cache_session_configuration_request(msg) await self.backend_ws.send(msg) # type: ignore[union-attr, attr-defined] - else: - await self.backend_ws.send(message) # type: ignore[union-attr, attr-defined] + sent = True + return sent + await self.backend_ws.send(message) # type: ignore[union-attr, attr-defined] + return True def _cache_session_configuration_request(self, transformed_message: str) -> None: """Store setup payload once sent to backend.""" @@ -308,8 +315,13 @@ class RealTimeStreaming: return if not self._has_audio_transcription_guardrails(): return - await self._send_to_backend(self._make_disable_auto_response_message()) - self._guardrail_turn_detection_update_sent = True + sent = await self._send_to_backend(self._make_disable_auto_response_message()) + # Only mark as sent when the provider transformation actually delivered + # the update to the backend. Otherwise (e.g. Gemini drops session.update + # after the initial setup), leave the flag unset so future opportunities + # — such as a duplicate session.created — can retry. + if sent: + self._guardrail_turn_detection_update_sent = True def _has_realtime_guardrails(self) -> bool: """Return True if any callback is registered for realtime guardrail event types.""" @@ -879,7 +891,7 @@ class RealTimeStreaming: ## LOGGING self.store_input(message=message) - + ## GUARDRAIL: Inject turn_detection into first session.update if needed try: msg_obj = json.loads(message) @@ -891,7 +903,9 @@ class RealTimeStreaming: ): # Inject turn_detection into the first session.update session = msg_obj.setdefault("session", {}) - session.setdefault("turn_detection", {})["create_response"] = False + session.setdefault("turn_detection", {})[ + "create_response" + ] = False message = json.dumps(msg_obj) self._guardrail_turn_detection_update_sent = True verbose_logger.debug( @@ -899,7 +913,7 @@ class RealTimeStreaming: ) except (json.JSONDecodeError, AttributeError): pass - + ## FORWARD TO BACKEND if self.provider_config: message = self.provider_config.transform_realtime_request( diff --git a/litellm/llms/gemini/realtime/transformation.py b/litellm/llms/gemini/realtime/transformation.py index 0c6b04c41d4..917f13eefb6 100644 --- a/litellm/llms/gemini/realtime/transformation.py +++ b/litellm/llms/gemini/realtime/transformation.py @@ -972,14 +972,16 @@ class GeminiRealtimeConfig(BaseRealtimeConfig): delta_type=current_delta_type ) else: - # Check if this key or any nested key matches our mapping - for map_key, openai_event in MAP_GEMINI_FIELD_TO_OPENAI_EVENT.items(): + # Check if this key or any nested key matches our mapping. Use a + # distinct loop variable so we don't shadow ``openai_event`` and + # leak the last dict value when no entry matches. + for map_key, candidate_event in MAP_GEMINI_FIELD_TO_OPENAI_EVENT.items(): if map_key == key or ( "." in map_key and GeminiRealtimeConfig.get_nested_value(json_message, map_key) is not None ): - openai_event = openai_event + openai_event = candidate_event break if openai_event is None: raise ValueError(f"Unknown openai event: {key}, value: {value}")