Fix greptile review

litellm/litellm_core_utils/realtime_streaming.py

@@ -75,6 +75,9 @@ class RealTimeStreaming:
         # When a text message is blocked, hold the guardrail reason so the next
         # response.create can be rewritten to include the failure context.
         self._pending_guardrail_message: Optional[str] = None
+        # Track whether session.created has already been sent to the client
+        # (e.g. synthetic event in deferred setup mode).
+        self._session_created_sent_to_client: bool = False
 
     def _should_store_message(
         self,
@@ -441,6 +444,13 @@ class RealTimeStreaming:
         )
         for event in events:
             event_str = json.dumps(event)
+            if isinstance(event, dict) and event.get("type") == "session.created":
+                if self._session_created_sent_to_client:
+                    verbose_logger.debug(
+                        "Skipping duplicate session.created from provider stream"
+                    )
+                    continue
+                self._session_created_sent_to_client = True
             ## For audio/VAD guardrail path: forward session.created first, then inject.
             if (
                 isinstance(event, dict)

litellm/llms/custom_httpx/llm_http_handler.py

@@ -5166,6 +5166,7 @@ class BaseLLMHTTPHandler:
                     )
                     if synthetic_session is not None:
                         await websocket.send_text(json.dumps(synthetic_session))
+                        realtime_streaming._session_created_sent_to_client = True
                         verbose_logger.debug(
                             "Sent synthetic session.created to client to unblock connection"
                         )

tests/test_litellm/litellm_core_utils/test_realtime_streaming.py

@@ -968,3 +968,57 @@ async def test_on_violation_end_session_closes_on_first_fail():
     assert streaming._violation_count == 1
 
     litellm.callbacks = []  # cleanup
+
+
+@pytest.mark.asyncio
+async def test_provider_path_suppresses_duplicate_session_created_after_synthetic():
+    client_ws = MagicMock()
+    client_ws.send_text = AsyncMock()
+
+    backend_ws = MagicMock()
+    backend_ws.recv = AsyncMock(
+        side_effect=[b'{"setupComplete": {}}', ConnectionClosed(None, None)]
+    )
+    backend_ws.send = AsyncMock()
+
+    provider_config = MagicMock()
+    provider_config.transform_realtime_response = MagicMock(
+        return_value={
+            "response": [
+                {
+                    "type": "session.created",
+                    "event_id": "event_1",
+                    "session": {"id": "sess_1", "modalities": ["audio"]},
+                }
+            ],
+            "current_output_item_id": None,
+            "current_response_id": None,
+            "current_delta_chunks": [],
+            "current_conversation_id": None,
+            "current_item_chunks": [],
+            "current_delta_type": None,
+            "session_configuration_request": None,
+        }
+    )
+
+    logging_obj = MagicMock()
+    logging_obj.litellm_trace_id = "trace_1"
+    logging_obj.async_success_handler = AsyncMock()
+    logging_obj.success_handler = MagicMock()
+
+    streaming = RealTimeStreaming(
+        websocket=client_ws,
+        backend_ws=backend_ws,
+        logging_obj=logging_obj,
+        provider_config=provider_config,
+        model="gemini-2.5-flash",
+    )
+    # Simulate synthetic session.created already sent by llm_http_handler.
+    streaming._session_created_sent_to_client = True
+
+    await streaming.backend_to_client_send_messages()
+
+    sent_payloads = [json.loads(c.args[0]) for c in client_ws.send_text.call_args_list]
+    assert not any(
+        payload.get("type") == "session.created" for payload in sent_payloads
+    ), f"Expected duplicate session.created to be suppressed, got: {sent_payloads}"