diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index adf91509011..c640c1e52ad 100644 --- a/litellm/litellm_core_utils/realtime_streaming.py +++ b/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) diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py index 270aaaa84da..7149e16fdfe 100644 --- a/litellm/llms/custom_httpx/llm_http_handler.py +++ b/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" ) 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 dab344d45e7..78443664d54 100644 --- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py +++ b/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}"