diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index a35d113fc9d..97a87ca1faf 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, Union, cast 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, @@ -327,8 +328,10 @@ class RealTimeStreaming: 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 18547f2f6a3..cad5f6d5b38 100644 --- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py +++ b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py @@ -2750,3 +2750,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()