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 <yassin@berri.ai>

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(

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}")