fix(realtime): mark realtime sessions async so failure hooks fire once

The relay's failure dispatch runs the async handler and then the legacy sync
failure_handler for the proxy's callable callbacks. The realtime logging object
carried no async marker, so failure_handler treated the session as a sync SDK
call and fired every CustomLogger's sync failure hook on top of the async one:
Langfuse recorded two ERROR observations per refused session, and OpenTelemetry,
MLflow, Braintrust, Literal AI, DeepEval and New Relic implement the same sync
hook. Plant the _arealtime marker in litellm_params the way aanthropic_messages
and agenerate_content already do, so both dispatchers classify the session async.

litellm/litellm_core_utils/litellm_logging.py

@@ -1820,6 +1820,7 @@ class Logging(LiteLLMLoggingBaseClass):
             and litellm_params.get(CallTypes.aanthropic_messages.value, False) is not True
             and litellm_params.get(CallTypes.agenerate_content.value, False) is not True
             and litellm_params.get(CallTypes.agenerate_content_stream.value, False) is not True
+            and litellm_params.get(CallTypes.arealtime.value, False) is not True
         )
 
     def _is_assembled_stream_success(self, result=None) -> bool:

litellm/realtime_api/main.py

@@ -27,7 +27,7 @@ from litellm.types.realtime import (
     RealtimeTranscriptionSessionRequest,
 )
 from litellm.types.router import GenericLiteLLMParams
-from litellm.types.utils import LlmProviders
+from litellm.types.utils import CallTypes, LlmProviders
 from litellm.utils import ProviderConfigManager
 
 from ..litellm_core_utils.get_litellm_params import get_litellm_params
@@ -355,7 +355,7 @@ async def _arealtime(
     user: Final = kwargs.get("user", None)
     litellm_params: Final = GenericLiteLLMParams(**kwargs)
 
-    litellm_params_dict: Final = get_litellm_params(**kwargs)
+    litellm_params_dict: Final = {**get_litellm_params(**kwargs), CallTypes.arealtime.value: True}
 
     model, _custom_llm_provider, dynamic_api_key, dynamic_api_base = get_llm_provider(
         model=model,

tests/test_litellm/litellm_core_utils/test_litellm_logging.py

@@ -995,6 +995,35 @@ async def test_anthropic_messages_marks_litellm_params_async():
         litellm.callbacks = original_callbacks
 
 
+@pytest.mark.asyncio
+async def test_arealtime_marks_litellm_params_async(monkeypatch):
+    """LIT-6973: ``_arealtime`` must plant ``_arealtime`` in ``litellm_params`` so
+    ``_is_sync_litellm_request`` classifies the session async and a failed session
+    reaches a CustomLogger's failure hook once, through the async path only, even
+    though the sync ``failure_handler`` still runs ahead of the async one."""
+    captured = {}
+    async_logged = asyncio.Event()
+
+    class CaptureLogger(CustomLogger):
+        async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
+            captured["litellm_params"] = kwargs.get("litellm_params", {})
+            async_logged.set()
+
+    logger = CaptureLogger()
+    logger.log_failure_event = MagicMock()
+    monkeypatch.setattr(litellm, "callbacks", [logger])
+    monkeypatch.setattr(litellm, "failure_callback", [])
+    monkeypatch.setattr(litellm, "_async_failure_callback", [])
+    monkeypatch.setattr(litellm, "success_callback", [])
+    monkeypatch.setattr(litellm, "_async_success_callback", [])
+    with pytest.raises(ValueError, match="Unsupported model"):
+        await litellm._arealtime(model="anthropic/claude-x", websocket=MagicMock())
+    await asyncio.wait_for(async_logged.wait(), timeout=10)
+    logger.log_failure_event.assert_not_called()
+    assert captured["litellm_params"].get("_arealtime") is True
+    assert LitellmLogging._is_sync_litellm_request(captured["litellm_params"]) is False
+
+
 @pytest.mark.asyncio
 async def test_agenerate_content_marks_litellm_params_async():
     """LIT-4475: the async ``agenerate_content`` entrypoint must plant
@@ -1180,6 +1209,7 @@ def test_is_sync_litellm_request():
     assert LitellmLogging._is_sync_litellm_request({}) is True
     assert LitellmLogging._is_sync_litellm_request({"acompletion": True}) is False
     assert LitellmLogging._is_sync_litellm_request({"allm_passthrough_route": True}) is False
+    assert LitellmLogging._is_sync_litellm_request({"_arealtime": True}) is False
     assert LitellmLogging._is_sync_litellm_request({"aanthropic_messages": True}) is False
     assert LitellmLogging._is_sync_litellm_request({"agenerate_content": True}) is False
     assert LitellmLogging._is_sync_litellm_request({"agenerate_content_stream": True}) is False