diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index bd6406c6241..a1a070eb5b7 100644 --- a/litellm/litellm_core_utils/realtime_streaming.py +++ b/litellm/litellm_core_utils/realtime_streaming.py @@ -5,6 +5,7 @@ from typing import TYPE_CHECKING, Any, Dict, List, Optional, Protocol, Union, ca import litellm from litellm._logging import verbose_logger +from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig from litellm.types.llms.openai import ( OpenAIRealtimeEvents, @@ -315,8 +316,10 @@ class RealTimeStreaming: self.logging_obj.model_call_details["realtime_tools"] = self.session_tools self.logging_obj.model_call_details["realtime_tool_calls"] = self.tool_calls ## ASYNC LOGGING - # Create an event loop for the new thread - asyncio.create_task(self.logging_obj.async_success_handler(self.messages)) + # Route through the bounded logging worker (per-coroutine timeout + + # concurrency cap) instead of a bare create_task, so a slow callback + # can't leave suspended tasks pinning each call's response in memory. + GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue(self.logging_obj.async_success_handler(self.messages)) ## SYNC LOGGING executor.submit(self.logging_obj.success_handler(self.messages)) 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 2ad9b919a1f..766befd1a99 100644 --- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py +++ b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py @@ -2945,3 +2945,26 @@ def test_non_bidi_setup_left_untouched_for_followup_capable_providers(): assert streaming._maybe_inject_guardrail_auto_response_disable(msg) == msg finally: litellm.callbacks = [] + + +@pytest.mark.asyncio +async def test_log_messages_routes_async_logging_through_bounded_worker(): + """Realtime success logging must go through GLOBAL_LOGGING_WORKER (bounded + queue + per-coroutine timeout), not a bare asyncio.create_task. A bare task + has no timeout/concurrency cap, so when a logging callback is slow every + realtime turn leaves a suspended task pinning its response in memory -> an + unbounded leak. Regression for that fix.""" + logging_obj = MagicMock() + streaming = RealTimeStreaming(MagicMock(), MagicMock(), logging_obj) + streaming.messages = [{"type": "session.created"}] + + with ( + patch("litellm.litellm_core_utils.realtime_streaming.GLOBAL_LOGGING_WORKER") as mock_worker, + patch("litellm.litellm_core_utils.realtime_streaming.asyncio.create_task") as mock_create_task, + patch("litellm.litellm_core_utils.realtime_streaming.executor.submit"), + ): + await streaming.log_messages() + + mock_worker.ensure_initialized_and_enqueue.assert_called_once() + # the bare create_task path must no longer be used for success logging + mock_create_task.assert_not_called()