From 6ee33df952ef9102f14960ed04c46e8f49900d66 Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Fri, 4 Sep 2026 19:25:12 -0700
Subject: [PATCH 01/10] fix(realtime): relay the upstream websocket close to
 the client instead of hanging

When the provider closes the realtime websocket (for example Vertex Live
refusing the session with 1008 "Publisher model ... was not found"), the
proxy swallowed the close and kept waiting on the client, so the client
sat on an open socket with nothing coming back and the session was logged
as a $0 success

The backend relay now returns the upstream close, and bidirectional_forward
sends the client an OpenAI-style error event naming the upstream code and
reason, then closes the client socket with the same code (or 1011 when the
upstream code is one a server may not send). A session the upstream refused
before sending any frame is logged through the failure handlers instead of
as a success
---
 litellm/litellm_core_utils/realtime_errors.py |   8 +
 .../litellm_core_utils/realtime_streaming.py  | 186 ++++++++++++------
 .../test_realtime_errors.py                   |  10 +
 .../test_realtime_streaming.py                | 169 +++++++++++++++-
 4 files changed, 303 insertions(+), 70 deletions(-)

diff --git a/litellm/litellm_core_utils/realtime_errors.py b/litellm/litellm_core_utils/realtime_errors.py
index e1b957f4325..3c064728a66 100644
--- a/litellm/litellm_core_utils/realtime_errors.py
+++ b/litellm/litellm_core_utils/realtime_errors.py
@@ -29,3 +29,11 @@ def websocket_close_reason(message: str, fallback: str) -> str:
     if len(encoded) <= WEBSOCKET_CLOSE_REASON_MAX_BYTES:
         return message
     return encoded[:WEBSOCKET_CLOSE_REASON_MAX_BYTES].decode("utf-8", errors="ignore")
+
+
+def client_close_code(upstream_code: int) -> int:
+    from websockets.frames import EXTERNAL_CLOSE_CODES, CloseCode
+
+    if upstream_code in EXTERNAL_CLOSE_CODES or 3000 <= upstream_code < 5000:
+        return upstream_code
+    return int(CloseCode.INTERNAL_ERROR)
diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py
index 8479e108d17..746343026ed 100644
--- a/litellm/litellm_core_utils/realtime_streaming.py
+++ b/litellm/litellm_core_utils/realtime_streaming.py
@@ -1,7 +1,9 @@
 import asyncio
 import json
-from collections.abc import Mapping, Sequence
-from typing import TYPE_CHECKING, Any, Final, Protocol, TypedDict, cast
+import traceback
+from collections.abc import Coroutine, Mapping, Sequence
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any, Final, NoReturn, Protocol, TypedDict, cast
 
 from typing_extensions import ReadOnly
 
@@ -19,9 +21,11 @@ from litellm.types.llms.openai import (
 from litellm.types.realtime import ALL_DELTA_TYPES
 
 from .litellm_logging import Logging as LiteLLMLogging
+from .realtime_errors import client_close_code, realtime_error_event, websocket_close_reason
 
 if TYPE_CHECKING:
     from websockets.asyncio.client import ClientConnection
+    from websockets.exceptions import ConnectionClosed
 
     from litellm.types.guardrails import GuardrailEventHooks
 
@@ -30,8 +34,22 @@ else:
     CLIENT_CONNECTION_CLASS = Any
 
 
-class _ClientWebSocketExceptions(Protocol):
-    ConnectionClosed: type[Exception]
+@dataclass(frozen=True, slots=True)
+class BackendClose:
+    code: int
+    reason: str
+
+    @property
+    def message(self) -> str:
+        if not self.reason:
+            return f"upstream websocket closed with code {self.code}"
+        return f"upstream websocket closed with code {self.code}: {self.reason}"
+
+
+def backend_close_from(error: "ConnectionClosed") -> BackendClose:
+    if error.rcvd is None:
+        return BackendClose(code=1006, reason=str(error))
+    return BackendClose(code=error.rcvd.code, reason=error.rcvd.reason)
 
 
 class _ASGIScope(TypedDict, total=False):
@@ -69,10 +87,13 @@ class _ScopedWebSocket(Protocol):
 
 
 class _ClientWebSocket(_ScopedWebSocket, Protocol):
-    exceptions: _ClientWebSocketExceptions
-
     async def send_text(self, data: str) -> None: ...
     async def receive_text(self) -> str: ...
+    async def close(self, code: int = 1000, reason: str | None = None) -> None: ...
+
+
+class _LoggingWorker(Protocol):
+    def ensure_initialized_and_enqueue(self, async_coroutine: Coroutine[object, object, None]) -> None: ...
 
 
 def _decode_json_object(payload: str) -> Mapping[str, object]:
@@ -108,11 +129,14 @@ class RealTimeStreaming:
         backend_uses_beta_protocol: bool | None = None,
         force_transcription_model: str | None = None,
         event_normalizer: RealtimeEventNormalizer | None = None,
+        logging_worker: _LoggingWorker = GLOBAL_LOGGING_WORKER,
     ):
         self.websocket: _ClientWebSocket = websocket
         self.backend_ws = backend_ws
         self.logging_obj = logging_obj
+        self._logging_worker = logging_worker
         self.messages: list[OpenAIRealtimeEvents] = []
+        self._backend_sent_frames: bool = False
         self.input_message: dict = {}
         self.input_messages: list[dict[str, str]] = []
         self.session_tools: list[dict] = []
@@ -388,7 +412,7 @@ class RealTimeStreaming:
             # 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_worker.ensure_initialized_and_enqueue(
                 self.logging_obj.dispatch_success_handlers(self.messages, prefer_async_handlers=True)
             )
 
@@ -1035,60 +1059,84 @@ class RealTimeStreaming:
             return True
         return False
 
-    async def backend_to_client_send_messages(self):
+    async def _relay_backend_messages(self) -> NoReturn:
+        while True:
+            try:
+                raw_response = await self.backend_ws.recv(decode=False)
+            except TypeError:
+                raw_response = await self.backend_ws.recv()
+            self._backend_sent_frames = True
+
+            if isinstance(raw_response, bytes):
+                try:
+                    raw_response = raw_response.decode("utf-8")
+                except UnicodeDecodeError:
+                    verbose_logger.warning("Received non-UTF-8 binary frame from backend, skipping.")
+                    continue
+
+            if self.provider_config:
+                try:
+                    await self._handle_provider_config_message(raw_response)
+                except Exception as e:
+                    verbose_logger.exception("Error processing backend message, skipping: %s", e)
+                    continue
+            else:
+                event = self._parse_backend_event(raw_response)
+                if event is None:
+                    await self.websocket.send_text(raw_response)
+                    continue
+
+                if self._should_drop_event_from_client(event):
+                    continue
+
+                if await self._handle_raw_backend_message(event, raw_response):
+                    continue
+
+                event = self._normalize_event_for_ga_client(event)
+                self.store_message(event)
+
+                if not self._client_wants_beta:
+                    await self.websocket.send_text(json.dumps(event))
+                    continue
+
+                translated = self._translate_event_to_beta(event)
+                if translated is None:
+                    continue
+                await self.websocket.send_text(json.dumps(translated))
+
+    async def backend_to_client_send_messages(self) -> BackendClose:
         import websockets
 
         try:
-            while True:
-                try:
-                    raw_response = await self.backend_ws.recv(decode=False)
-                except TypeError:
-                    raw_response = await self.backend_ws.recv()
-
-                if isinstance(raw_response, bytes):
-                    try:
-                        raw_response = raw_response.decode("utf-8")
-                    except UnicodeDecodeError:
-                        verbose_logger.warning("Received non-UTF-8 binary frame from backend, skipping.")
-                        continue
-
-                if self.provider_config:
-                    try:
-                        await self._handle_provider_config_message(raw_response)
-                    except Exception as e:
-                        verbose_logger.exception("Error processing backend message, skipping: %s", e)
-                        continue
-                else:
-                    event = self._parse_backend_event(raw_response)
-                    if event is None:
-                        await self.websocket.send_text(raw_response)
-                        continue
-
-                    if self._should_drop_event_from_client(event):
-                        continue
-
-                    if await self._handle_raw_backend_message(event, raw_response):
-                        continue
-
-                    event = self._normalize_event_for_ga_client(event)
-                    self.store_message(event)
-
-                    if not self._client_wants_beta:
-                        await self.websocket.send_text(json.dumps(event))
-                        continue
-
-                    translated = self._translate_event_to_beta(event)
-                    if translated is None:
-                        continue
-                    await self.websocket.send_text(json.dumps(translated))
-
+            await self._relay_backend_messages()
         except websockets.exceptions.ConnectionClosed as e:
             verbose_logger.exception("Connection closed in backend to client send messages - %s", e)
-        except Exception as e:
-            verbose_logger.exception("Error in backend to client send messages: %s", e)
-        finally:
+            close: Final = backend_close_from(e)
+            self._flush_unbilled_transcription_usage()
+            if self._backend_refused_session(close):
+                await self.log_backend_refusal(e)
+            else:
+                await self.log_messages()
+            return close
+        except asyncio.CancelledError:
             self._flush_unbilled_transcription_usage()
             await self.log_messages()
+            raise
+        except Exception as e:
+            verbose_logger.exception("Error in backend to client send messages: %s", e)
+            self._flush_unbilled_transcription_usage()
+            await self.log_messages()
+            return BackendClose(code=1011, reason="proxy failed while relaying the upstream websocket")
+
+    def _backend_refused_session(self, close: BackendClose) -> bool:
+        return close.code != 1000 and not self._backend_sent_frames and not self.messages
+
+    async def log_backend_refusal(self, error: Exception) -> None:
+        if not self.logging_obj:
+            return
+        self._logging_worker.ensure_initialized_and_enqueue(
+            self.logging_obj.dispatch_failure_handlers(error, traceback.format_exc(), prefer_async_handlers=True)
+        )
 
     @staticmethod
     def _detect_beta_header(websocket: _ScopedWebSocket) -> bool:
@@ -1484,20 +1532,28 @@ class RealTimeStreaming:
         except Exception as e:
             verbose_logger.debug("Error in client ack messages: %s", e)
 
-    async def bidirectional_forward(self):
+    async def bidirectional_forward(self) -> None:
         forward_task: Final = asyncio.create_task(self.backend_to_client_send_messages())
+        client_task: Final = asyncio.create_task(self.client_ack_messages())
         try:
-            await self.client_ack_messages()
-        except self.websocket.exceptions.ConnectionClosed:
-            verbose_logger.debug("Connection closed")
-            forward_task.cancel()
+            await asyncio.wait((forward_task, client_task), return_when=asyncio.FIRST_COMPLETED)
+            if not client_task.done():
+                await self._close_client(forward_task.result())
         finally:
-            if not forward_task.done():
-                forward_task.cancel()
-                try:
-                    await forward_task
-                except asyncio.CancelledError:
-                    pass
+            forward_task.cancel()
+            client_task.cancel()
+            await asyncio.gather(forward_task, client_task, return_exceptions=True)
+
+    async def _close_client(self, close: BackendClose) -> None:
+        try:
+            if close.code != 1000:
+                await self.websocket.send_text(realtime_error_event(close.message, error_type="server_error"))
+            await self.websocket.close(
+                code=client_close_code(close.code),
+                reason=websocket_close_reason(close.reason, fallback=close.message),
+            )
+        except Exception as e:  # noqa: BLE001  # the client may already be gone; the session is over either way
+            verbose_logger.debug("Could not relay the upstream close to the client: %s", e)
 
 
 def client_sent_openai_beta_realtime_header(websocket: _ScopedWebSocket) -> bool:
diff --git a/tests/test_litellm/litellm_core_utils/test_realtime_errors.py b/tests/test_litellm/litellm_core_utils/test_realtime_errors.py
index 494d16b0b9b..1d2cf905f4e 100644
--- a/tests/test_litellm/litellm_core_utils/test_realtime_errors.py
+++ b/tests/test_litellm/litellm_core_utils/test_realtime_errors.py
@@ -1,8 +1,10 @@
 import json
 
+import pytest
 
 from litellm.litellm_core_utils.realtime_errors import (
     WEBSOCKET_CLOSE_REASON_MAX_BYTES,
+    client_close_code,
     realtime_error_event,
     websocket_close_reason,
 )
@@ -42,3 +44,11 @@ def test_websocket_close_reason_truncates_multibyte_message_by_bytes():
     assert len(reason.encode("utf-8")) <= WEBSOCKET_CLOSE_REASON_MAX_BYTES
     assert reason == "あ" * (WEBSOCKET_CLOSE_REASON_MAX_BYTES // 3)
     assert "�" not in reason
+
+
+@pytest.mark.parametrize(
+    ("upstream_code", "expected"),
+    [(1000, 1000), (1008, 1008), (1011, 1011), (4001, 4001), (1005, 1011), (1006, 1011), (1015, 1011), (2999, 1011)],
+)
+def test_client_close_code_only_forwards_codes_a_server_may_send(upstream_code, expected):
+    assert client_close_code(upstream_code) == expected
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 52e88db753a..1e0456079e9 100644
--- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
+++ b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
@@ -1,8 +1,13 @@
+import asyncio
 import json
+from collections.abc import Coroutine
+from dataclasses import dataclass
+from typing import Final
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from websockets.exceptions import ConnectionClosed
+from websockets.frames import Close
 
 import litellm
 
@@ -2941,13 +2946,11 @@ async def test_log_messages_routes_async_logging_through_bounded_worker():
     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)
+    mock_worker = MagicMock()
+    streaming = RealTimeStreaming(MagicMock(), MagicMock(), logging_obj, logging_worker=mock_worker)
     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,
-    ):
+    with patch("litellm.litellm_core_utils.realtime_streaming.asyncio.create_task") as mock_create_task:
         await streaming.log_messages()
 
         mock_worker.ensure_initialized_and_enqueue.assert_called_once()
@@ -3111,3 +3114,159 @@ async def test_session_close_flush_noop_without_unbilled_usage():
         isinstance(message, dict) and message.get("type") == "conversation.item.input_audio_transcription.completed"
         for message in streaming.messages
     )
+
+
+
+_UPSTREAM_REFUSAL: Final = "Publisher model `publishers/google/models/gemini-live-2.5-flash` was not found"
+
+
+class _InlineLoggingWorker:
+    def __init__(self) -> None:
+        self.enqueued: tuple[Coroutine[object, object, None], ...] = ()
+
+    def ensure_initialized_and_enqueue(self, async_coroutine: Coroutine[object, object, None]) -> None:
+        self.enqueued = (*self.enqueued, async_coroutine)
+
+    async def drain(self) -> None:
+        for coroutine in self.enqueued:
+            await coroutine
+
+
+class _RecordingLogging:
+    def __init__(self) -> None:
+        self.logged_sessions: tuple[tuple[dict, ...], ...] = ()
+        self.logged_failures: tuple[Exception, ...] = ()
+
+    async def dispatch_success_handlers(self, result: list[dict], prefer_async_handlers: bool = False) -> None:
+        self.logged_sessions = (*self.logged_sessions, tuple(result))
+
+    async def dispatch_failure_handlers(
+        self, exception: Exception, traceback_exception: str, prefer_async_handlers: bool = False
+    ) -> None:
+        self.logged_failures = (*self.logged_failures, exception)
+
+
+@dataclass(frozen=True, slots=True)
+class _RelaySession:
+    streaming: RealTimeStreaming
+    logging: _RecordingLogging
+    worker: _InlineLoggingWorker
+
+    async def run(self) -> None:
+        await asyncio.wait_for(self.streaming.bidirectional_forward(), timeout=2)
+        await self.worker.drain()
+
+
+async def _wait_forever() -> str:
+    await asyncio.Event().wait()
+    raise AssertionError("unreachable")
+
+
+def _client_ws_that_never_sends() -> MagicMock:
+    client_ws: Final = MagicMock()
+    client_ws.headers = {}
+    client_ws.receive_text = AsyncMock(side_effect=_wait_forever)
+    client_ws.send_text = AsyncMock()
+    client_ws.close = AsyncMock()
+    return client_ws
+
+
+def _backend_ws_closing_with(*frames: bytes | Exception) -> MagicMock:
+    backend_ws: Final = MagicMock()
+    backend_ws.recv = AsyncMock(side_effect=list(frames))
+    return backend_ws
+
+
+def _relay_session(client_ws: MagicMock, backend_ws: MagicMock) -> _RelaySession:
+    logging: Final = _RecordingLogging()
+    worker: Final = _InlineLoggingWorker()
+    streaming: Final = RealTimeStreaming(
+        client_ws, backend_ws, logging, model="gpt-realtime", logging_worker=worker
+    )
+    return _RelaySession(streaming=streaming, logging=logging, worker=worker)
+
+
+def _error_events_sent_to(client_ws: MagicMock) -> list[dict]:
+    events: Final = (json.loads(call.args[0]) for call in client_ws.send_text.await_args_list)
+    return [event for event in events if event.get("type") == "error"]
+
+
+@pytest.mark.asyncio
+async def test_bidirectional_forward_relays_upstream_policy_close_to_client():
+    client_ws: Final = _client_ws_that_never_sends()
+    upstream_close: Final = ConnectionClosed(Close(1008, _UPSTREAM_REFUSAL), None)
+    session: Final = _relay_session(client_ws, _backend_ws_closing_with(upstream_close))
+
+    await session.run()
+
+    (error_event,) = _error_events_sent_to(client_ws)
+    assert error_event["error"]["type"] == "server_error"
+    assert "1008" in error_event["error"]["message"]
+    assert _UPSTREAM_REFUSAL in error_event["error"]["message"]
+    client_ws.close.assert_awaited_once_with(code=1008, reason=_UPSTREAM_REFUSAL)
+
+
+@pytest.mark.asyncio
+async def test_bidirectional_forward_maps_abnormal_upstream_close_to_internal_error():
+    client_ws: Final = _client_ws_that_never_sends()
+    session: Final = _relay_session(client_ws, _backend_ws_closing_with(ConnectionClosed(None, None)))
+
+    await session.run()
+
+    (error_event,) = _error_events_sent_to(client_ws)
+    assert "1006" in error_event["error"]["message"]
+    client_ws.close.assert_awaited_once()
+    assert client_ws.close.await_args.kwargs["code"] == 1011
+
+
+@pytest.mark.asyncio
+async def test_bidirectional_forward_relays_normal_upstream_close_without_error_event():
+    client_ws: Final = _client_ws_that_never_sends()
+    session: Final = _relay_session(client_ws, _backend_ws_closing_with(ConnectionClosed(Close(1000, ""), None)))
+
+    await session.run()
+
+    assert _error_events_sent_to(client_ws) == []
+    client_ws.close.assert_awaited_once()
+    assert client_ws.close.await_args.kwargs["code"] == 1000
+
+
+@pytest.mark.asyncio
+async def test_upstream_refusal_before_any_frame_logs_a_failure_not_a_success():
+    upstream_close: Final = ConnectionClosed(Close(1008, _UPSTREAM_REFUSAL), None)
+    session: Final = _relay_session(_client_ws_that_never_sends(), _backend_ws_closing_with(upstream_close))
+
+    await session.run()
+
+    assert session.logging.logged_failures == (upstream_close,)
+    assert session.logging.logged_sessions == ()
+
+
+@pytest.mark.asyncio
+async def test_upstream_close_after_relayed_events_still_logs_the_session_as_success():
+    client_ws: Final = _client_ws_that_never_sends()
+    session_created: Final = json.dumps({"type": "session.created", "session": {"id": "sess_1"}}).encode()
+    upstream_close: Final = ConnectionClosed(Close(1008, _UPSTREAM_REFUSAL), None)
+    session: Final = _relay_session(client_ws, _backend_ws_closing_with(session_created, upstream_close))
+
+    await session.run()
+
+    (logged_session,) = session.logging.logged_sessions
+    assert [event["type"] for event in logged_session] == ["session.created"]
+    assert session.logging.logged_failures == ()
+    client_ws.close.assert_awaited_once_with(code=1008, reason=_UPSTREAM_REFUSAL)
+
+
+@pytest.mark.asyncio
+async def test_client_hanging_up_first_ends_the_session_without_a_relayed_close():
+    client_ws: Final = _client_ws_that_never_sends()
+    client_ws.receive_text = AsyncMock(side_effect=RuntimeError("client went away"))
+    backend_ws: Final = MagicMock()
+    backend_ws.recv = AsyncMock(side_effect=_wait_forever)
+    session: Final = _relay_session(client_ws, backend_ws)
+
+    await session.run()
+
+    assert session.logging.logged_sessions == ((),)
+    assert session.logging.logged_failures == ()
+    client_ws.close.assert_not_awaited()

From 85d45fbb4b6f9299af75b6891af61664176ad69f Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Fri, 4 Sep 2026 19:33:30 -0700
Subject: [PATCH 02/10] fix(realtime): relay the upstream close even when a
 client message hit the closed socket first

When the upstream closes while the proxy is forwarding a client message,
the client loop ends before the backend relay sees the close, and the
relay skipped closing the client because it read the client loop's exit
as the client hanging up. The client loop now reports why it stopped, so
a close observed on the backend send still reaches the client with the
error event and the upstream close code
---
 .../litellm_core_utils/realtime_streaming.py  | 19 ++++++++--
 .../test_realtime_streaming.py                | 37 +++++++++++++++++++
 2 files changed, 53 insertions(+), 3 deletions(-)

diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py
index 746343026ed..530391c7b57 100644
--- a/litellm/litellm_core_utils/realtime_streaming.py
+++ b/litellm/litellm_core_utils/realtime_streaming.py
@@ -3,6 +3,7 @@ import json
 import traceback
 from collections.abc import Coroutine, Mapping, Sequence
 from dataclasses import dataclass
+from enum import Enum, auto
 from typing import TYPE_CHECKING, Any, Final, NoReturn, Protocol, TypedDict, cast
 
 from typing_extensions import ReadOnly
@@ -46,6 +47,11 @@ class BackendClose:
         return f"upstream websocket closed with code {self.code}: {self.reason}"
 
 
+class ClientLoopExit(Enum):
+    CLIENT_DISCONNECTED = auto()
+    BACKEND_CLOSED = auto()
+
+
 def backend_close_from(error: "ConnectionClosed") -> BackendClose:
     if error.rcvd is None:
         return BackendClose(code=1006, reason=str(error))
@@ -1291,7 +1297,9 @@ class RealTimeStreaming:
         item["content"] = new_content
         return item
 
-    async def client_ack_messages(self):
+    async def client_ack_messages(self) -> ClientLoopExit:
+        import websockets
+
         client_event: _ClientEventFrame
         try:
             while True:
@@ -1529,16 +1537,21 @@ class RealTimeStreaming:
                 if guardrail_turn_detection_injected and sent:
                     self._guardrail_turn_detection_update_sent = True
 
+        except websockets.exceptions.ConnectionClosed as e:
+            verbose_logger.debug("Backend closed while forwarding a client message: %s", e)
+            return ClientLoopExit.BACKEND_CLOSED
         except Exception as e:
             verbose_logger.debug("Error in client ack messages: %s", e)
+            return ClientLoopExit.CLIENT_DISCONNECTED
 
     async def bidirectional_forward(self) -> None:
         forward_task: Final = asyncio.create_task(self.backend_to_client_send_messages())
         client_task: Final = asyncio.create_task(self.client_ack_messages())
         try:
             await asyncio.wait((forward_task, client_task), return_when=asyncio.FIRST_COMPLETED)
-            if not client_task.done():
-                await self._close_client(forward_task.result())
+            if client_task.done() and client_task.result() is ClientLoopExit.CLIENT_DISCONNECTED:
+                return
+            await self._close_client(await forward_task)
         finally:
             forward_task.cancel()
             client_task.cancel()
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 1e0456079e9..41b7557f6b2 100644
--- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
+++ b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
@@ -3134,9 +3134,13 @@ class _InlineLoggingWorker:
 
 class _RecordingLogging:
     def __init__(self) -> None:
+        self.model_call_details: dict[str, object] = {}
         self.logged_sessions: tuple[tuple[dict, ...], ...] = ()
         self.logged_failures: tuple[Exception, ...] = ()
 
+    def pre_call(self, input: str | dict, api_key: str) -> None:
+        return None
+
     async def dispatch_success_handlers(self, result: list[dict], prefer_async_handlers: bool = False) -> None:
         self.logged_sessions = (*self.logged_sessions, tuple(result))
 
@@ -3257,6 +3261,39 @@ async def test_upstream_close_after_relayed_events_still_logs_the_session_as_suc
     client_ws.close.assert_awaited_once_with(code=1008, reason=_UPSTREAM_REFUSAL)
 
 
+@pytest.mark.asyncio
+async def test_upstream_closing_while_a_client_message_is_forwarded_still_reaches_the_client():
+    upstream_close: Final = ConnectionClosed(Close(1008, _UPSTREAM_REFUSAL), None)
+    backend_closed: Final = asyncio.Event()
+    client_messages: Final = iter((json.dumps({"type": "response.create"}),))
+
+    async def receive_text() -> str:
+        message = next(client_messages, None)
+        return message if message is not None else await _wait_forever()
+
+    async def send_to_backend(_message: str) -> None:
+        backend_closed.set()
+        raise upstream_close
+
+    async def recv_from_backend() -> bytes:
+        await backend_closed.wait()
+        raise upstream_close
+
+    client_ws: Final = _client_ws_that_never_sends()
+    client_ws.receive_text = receive_text
+    backend_ws: Final = MagicMock()
+    backend_ws.send = send_to_backend
+    backend_ws.recv = recv_from_backend
+    session: Final = _relay_session(client_ws, backend_ws)
+
+    await session.run()
+
+    (error_event,) = _error_events_sent_to(client_ws)
+    assert _UPSTREAM_REFUSAL in error_event["error"]["message"]
+    client_ws.close.assert_awaited_once_with(code=1008, reason=_UPSTREAM_REFUSAL)
+    assert session.logging.logged_failures == (upstream_close,)
+
+
 @pytest.mark.asyncio
 async def test_client_hanging_up_first_ends_the_session_without_a_relayed_close():
     client_ws: Final = _client_ws_that_never_sends()

From da9dbdba961ce2981f9d82669ab9e3eb3a9d90a0 Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Fri, 4 Sep 2026 19:52:48 -0700
Subject: [PATCH 03/10] fix(realtime): treat any client receive failure as a
 client hangup

client_ack_messages classified a websockets ConnectionClosed raised by
the client socket as the backend closing, so bidirectional_forward kept
waiting on the upstream instead of ending the session. Starlette clients
raise WebSocketDisconnect, but the realtime test client in
tests/llm_translation/realtime raises websockets.exceptions.ConnectionClosed,
which hung test_openai_realtime_simple.py until the run was killed.

Only the receive_text call now maps every exception to
CLIENT_DISCONNECTED; the loop body keeps ConnectionClosed as
BACKEND_CLOSED, since the backend socket is the only websockets socket
touched there.
---
 litellm/litellm_core_utils/realtime_streaming.py  | 11 ++++++++++-
 .../litellm_core_utils/test_realtime_streaming.py | 15 +++++++++++++++
 2 files changed, 25 insertions(+), 1 deletion(-)

diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py
index 530391c7b57..bb7fbd81146 100644
--- a/litellm/litellm_core_utils/realtime_streaming.py
+++ b/litellm/litellm_core_utils/realtime_streaming.py
@@ -1297,13 +1297,22 @@ class RealTimeStreaming:
         item["content"] = new_content
         return item
 
+    async def _receive_client_message(self) -> str | None:
+        try:
+            return await self.websocket.receive_text()
+        except Exception as e:  # noqa: BLE001  # whatever the client socket raises, the client is gone
+            verbose_logger.debug("Client disconnected: %s", e)
+            return None
+
     async def client_ack_messages(self) -> ClientLoopExit:
         import websockets
 
         client_event: _ClientEventFrame
         try:
             while True:
-                message = await self.websocket.receive_text()
+                message = await self._receive_client_message()
+                if message is None:
+                    return ClientLoopExit.CLIENT_DISCONNECTED
 
                 ## GUARDRAIL: intercept conversation.item.create for text-based injection.
                 guardrail_turn_detection_injected = False
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 41b7557f6b2..00addb613c2 100644
--- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
+++ b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
@@ -3307,3 +3307,18 @@ async def test_client_hanging_up_first_ends_the_session_without_a_relayed_close(
     assert session.logging.logged_sessions == ((),)
     assert session.logging.logged_failures == ()
     client_ws.close.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_client_hanging_up_with_a_websockets_close_is_not_mistaken_for_the_backend_closing():
+    client_ws: Final = _client_ws_that_never_sends()
+    client_ws.receive_text = AsyncMock(side_effect=ConnectionClosed(None, None))
+    backend_ws: Final = MagicMock()
+    backend_ws.recv = AsyncMock(side_effect=_wait_forever)
+    session: Final = _relay_session(client_ws, backend_ws)
+
+    await session.run()
+
+    assert session.logging.logged_sessions == ((),)
+    assert session.logging.logged_failures == ()
+    client_ws.close.assert_not_awaited()

From 14f8677bfcdc160d3b3b424dc84a9c1727734939 Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Fri, 4 Sep 2026 20:57:51 -0700
Subject: [PATCH 04/10] 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 |  1 +
 litellm/realtime_api/main.py                  |  4 +--
 .../test_litellm_logging.py                   | 30 +++++++++++++++++++
 3 files changed, 33 insertions(+), 2 deletions(-)

diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py
index df579f6df5b..15585c64efb 100644
--- a/litellm/litellm_core_utils/litellm_logging.py
+++ b/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:
diff --git a/litellm/realtime_api/main.py b/litellm/realtime_api/main.py
index 3862aec445f..9de91dfcaa5 100644
--- a/litellm/realtime_api/main.py
+++ b/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,
diff --git a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py b/tests/test_litellm/litellm_core_utils/test_litellm_logging.py
index f1de7390b5b..af75691eb10 100644
--- a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py
+++ b/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

From 412c36bb8e0663fd27e6c635d94f35d7407eabd3 Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Fri, 4 Sep 2026 21:24:36 -0700
Subject: [PATCH 05/10] fix(realtime): detect an upstream refusal from received
 frames, not the session log

The refusal predicate also required the session log to be empty, but that
log is not limited to upstream frames. With gemini_live_defer_setup the
handler stores a synthetic session.created before the relay starts, and
the transcription usage flush appends a usage event before the check
runs, so an upstream policy close with no received frames was still
logged as a $0 success. Key the check off the received-frames flag only
---
 .../litellm_core_utils/realtime_streaming.py  |  2 +-
 .../test_realtime_streaming.py                | 39 +++++++++++++++++--
 2 files changed, 37 insertions(+), 4 deletions(-)

diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py
index bb7fbd81146..984934daaac 100644
--- a/litellm/litellm_core_utils/realtime_streaming.py
+++ b/litellm/litellm_core_utils/realtime_streaming.py
@@ -1135,7 +1135,7 @@ class RealTimeStreaming:
             return BackendClose(code=1011, reason="proxy failed while relaying the upstream websocket")
 
     def _backend_refused_session(self, close: BackendClose) -> bool:
-        return close.code != 1000 and not self._backend_sent_frames and not self.messages
+        return close.code != 1000 and not self._backend_sent_frames
 
     async def log_backend_refusal(self, error: Exception) -> None:
         if not self.logging_obj:
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 00addb613c2..09757c35570 100644
--- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
+++ b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
@@ -3031,12 +3031,12 @@ async def test_session_close_flushes_unbilled_transcription_usage():
     messages before log_messages runs, and never forwarded to the client."""
     from typing import Final
 
-    from litellm.types.realtime import RealtimeInputAudioTranscriptionUsage
+    from litellm.types.realtime import RealtimeInputAudioTranscriptionUsage, RealtimeResponseTypedDict
 
     client_ws: Final = MagicMock()
     client_ws.send_text = AsyncMock()
     backend_ws: Final = MagicMock()
-    backend_ws.recv = AsyncMock(side_effect=ConnectionClosed(None, None))
+    backend_ws.recv = AsyncMock(side_effect=[b'{"serverContent": {}}', ConnectionClosed(None, None)])
     logging_obj: Final = MagicMock()
     logging_obj.async_success_handler = AsyncMock()
     logging_obj.success_handler = MagicMock()
@@ -3048,7 +3048,24 @@ async def test_session_close_flushes_unbilled_transcription_usage():
         "total_tokens": 171,
         "input_token_details": {"text_tokens": 0, "audio_tokens": 153},
     }
+    transcript_frame: Final[RealtimeResponseTypedDict] = {
+        "response": {
+            "type": "conversation.item.input_audio_transcription.completed",
+            "event_id": "event_1",
+            "transcript": "ahoy",
+            "item_id": "item_1",
+            "content_index": 0,
+        },
+        "current_output_item_id": None,
+        "current_response_id": None,
+        "current_delta_chunks": None,
+        "current_conversation_id": None,
+        "current_item_chunks": None,
+        "current_delta_type": None,
+        "session_configuration_request": None,
+    }
     provider_config: Final = MagicMock()
+    provider_config.transform_realtime_response = MagicMock(return_value=transcript_frame)
     provider_config.unbilled_usage_on_session_close = MagicMock(return_value=usage)
 
     streaming: Final = RealTimeStreaming(
@@ -3080,7 +3097,9 @@ async def test_session_close_flushes_unbilled_transcription_usage():
     )
     assert len(flushed) == 1
     assert flushed[0] in logged_snapshots[0]
-    assert not client_ws.send_text.called
+    forwarded: Final = tuple(json.loads(call.args[0]) for call in client_ws.send_text.await_args_list)
+    assert [event.get("transcript") for event in forwarded] == ["ahoy"]
+    assert all("usage" not in event for event in forwarded)
 
 
 @pytest.mark.asyncio
@@ -3246,6 +3265,20 @@ async def test_upstream_refusal_before_any_frame_logs_a_failure_not_a_success():
     assert session.logging.logged_sessions == ()
 
 
+@pytest.mark.asyncio
+async def test_upstream_refusal_after_a_synthetic_session_created_still_logs_a_failure():
+    """LIT-6973: deferred Gemini Live setup stores a synthetic ``session.created`` before
+    the relay starts. It is not an upstream frame, so a refusal after it is still a refusal."""
+    upstream_close: Final = ConnectionClosed(Close(1008, _UPSTREAM_REFUSAL), None)
+    session: Final = _relay_session(_client_ws_that_never_sends(), _backend_ws_closing_with(upstream_close))
+    session.streaming.store_message(json.dumps({"type": "session.created", "session": {"id": "sess_synthetic"}}))
+
+    await session.run()
+
+    assert session.logging.logged_failures == (upstream_close,)
+    assert session.logging.logged_sessions == ()
+
+
 @pytest.mark.asyncio
 async def test_upstream_close_after_relayed_events_still_logs_the_session_as_success():
     client_ws: Final = _client_ws_that_never_sends()

From 74613f9bd47d8e3068e6e2f1f519675ac15b7ab8 Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Fri, 4 Sep 2026 22:52:26 -0700
Subject: [PATCH 06/10] fix(realtime): redact credentials from the relayed
 upstream close

The handshake error path already runs client-facing error strings through
_redact_string; the relay's _close_client did not, so a secret echoed in an
upstream close reason could reach the client verbatim. Mirror the handshake
path and scrub the close message and reason before relaying them.
---
 .../litellm_core_utils/realtime_streaming.py   |  8 +++++---
 .../test_realtime_streaming.py                 | 18 ++++++++++++++++++
 2 files changed, 23 insertions(+), 3 deletions(-)

diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py
index 984934daaac..b448cb7c9ff 100644
--- a/litellm/litellm_core_utils/realtime_streaming.py
+++ b/litellm/litellm_core_utils/realtime_streaming.py
@@ -9,7 +9,7 @@ from typing import TYPE_CHECKING, Any, Final, NoReturn, Protocol, TypedDict, cas
 from typing_extensions import ReadOnly
 
 import litellm
-from litellm._logging import verbose_logger
+from litellm._logging import _redact_string, 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 (
@@ -1567,12 +1567,14 @@ class RealTimeStreaming:
             await asyncio.gather(forward_task, client_task, return_exceptions=True)
 
     async def _close_client(self, close: BackendClose) -> None:
+        redacted_message: Final = _redact_string(close.message)
+        redacted_reason: Final = _redact_string(close.reason)
         try:
             if close.code != 1000:
-                await self.websocket.send_text(realtime_error_event(close.message, error_type="server_error"))
+                await self.websocket.send_text(realtime_error_event(redacted_message, error_type="server_error"))
             await self.websocket.close(
                 code=client_close_code(close.code),
-                reason=websocket_close_reason(close.reason, fallback=close.message),
+                reason=websocket_close_reason(redacted_reason, fallback=redacted_message),
             )
         except Exception as e:  # noqa: BLE001  # the client may already be gone; the session is over either way
             verbose_logger.debug("Could not relay the upstream close to the client: %s", e)
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 09757c35570..bcfacf16205 100644
--- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
+++ b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
@@ -3229,6 +3229,24 @@ async def test_bidirectional_forward_relays_upstream_policy_close_to_client():
     client_ws.close.assert_awaited_once_with(code=1008, reason=_UPSTREAM_REFUSAL)
 
 
+@pytest.mark.asyncio
+async def test_upstream_close_reason_with_a_secret_is_redacted_before_reaching_the_client():
+    """LIT-6973: the relayed close mirrors the handshake path and scrubs credential
+    patterns, so an upstream error echoing a token never reaches the client verbatim."""
+    secret: Final = "sk-live-abcdef0123456789abcdef0123"
+    client_ws: Final = _client_ws_that_never_sends()
+    upstream_close: Final = ConnectionClosed(Close(1008, f"auth failed for {secret}"), None)
+    session: Final = _relay_session(client_ws, _backend_ws_closing_with(upstream_close))
+
+    await session.run()
+
+    (error_event,) = _error_events_sent_to(client_ws)
+    assert secret not in error_event["error"]["message"]
+    relayed_reason: Final = client_ws.close.await_args.kwargs["reason"]
+    assert secret not in relayed_reason
+    assert "REDACTED" in relayed_reason
+
+
 @pytest.mark.asyncio
 async def test_bidirectional_forward_maps_abnormal_upstream_close_to_internal_error():
     client_ws: Final = _client_ws_that_never_sends()

From af3ddb477a852f20898aeefd7bc35713188f98da Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Sat, 5 Sep 2026 00:22:20 -0700
Subject: [PATCH 07/10] fix(realtime): release the budget reservation on a
 failed session and scrub relayed close details

A refused or failed /v1/realtime session never ran the success cost callback
or a failure hook, so its pre-call budget reservation stayed open and kept the
key/team/user spend counters pinned above real spend, 429ing later requests on
the same key until the counter's TTL expired. The endpoint now reconciles the
reservation in a finally, reusing a shared release_or_invalidate_budget_reservation
helper that mirrors the success/failure paths (release to zero, else invalidate
the reserved counters and finalize).

The relayed upstream close message and reason also go through the proxy's
client-facing redaction, so a credential, internal hostname, private IP, or
server path echoed by the upstream never reaches the client verbatim.
---
 .../litellm_core_utils/realtime_streaming.py  |  6 +-
 litellm/proxy/proxy_server.py                 | 12 +++
 .../spend_tracking/budget_reservation.py      | 25 ++++++
 .../test_realtime_streaming.py                | 21 +++--
 tests/test_litellm/proxy/test_proxy_server.py | 83 +++++++++++++++++++
 5 files changed, 137 insertions(+), 10 deletions(-)

diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py
index b448cb7c9ff..c670278d3fb 100644
--- a/litellm/litellm_core_utils/realtime_streaming.py
+++ b/litellm/litellm_core_utils/realtime_streaming.py
@@ -9,7 +9,7 @@ from typing import TYPE_CHECKING, Any, Final, NoReturn, Protocol, TypedDict, cas
 from typing_extensions import ReadOnly
 
 import litellm
-from litellm._logging import _redact_string, verbose_logger
+from litellm._logging import redact_internal_details_from_client_message, 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 (
@@ -1567,8 +1567,8 @@ class RealTimeStreaming:
             await asyncio.gather(forward_task, client_task, return_exceptions=True)
 
     async def _close_client(self, close: BackendClose) -> None:
-        redacted_message: Final = _redact_string(close.message)
-        redacted_reason: Final = _redact_string(close.reason)
+        redacted_message: Final = redact_internal_details_from_client_message(close.message)
+        redacted_reason: Final = redact_internal_details_from_client_message(close.reason)
         try:
             if close.code != 1000:
                 await self.websocket.send_text(realtime_error_event(redacted_message, error_type="server_error"))
diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py
index 61bcdea94d4..06cfee45918 100644
--- a/litellm/proxy/proxy_server.py
+++ b/litellm/proxy/proxy_server.py
@@ -11453,6 +11453,16 @@ def _realtime_query_params_template(model: str | None, intent: str | None) -> tu
     return tuple(params)
 
 
+async def _release_realtime_budget_reservation(user_api_key_dict: UserAPIKeyAuth) -> None:
+    from litellm.proxy.spend_tracking.budget_reservation import (
+        release_or_invalidate_budget_reservation,
+    )
+
+    await release_or_invalidate_budget_reservation(
+        budget_reservation=user_api_key_dict.budget_reservation,
+    )
+
+
 @app.websocket("/openai/v1/realtime")
 @app.websocket("/v1/realtime")
 @app.websocket("/realtime")
@@ -11592,6 +11602,8 @@ async def realtime_websocket_endpoint(
             )
         except Exception:  # noqa: BLE001  # the lower layer may have closed the socket already; closing twice is not an error
             verbose_proxy_logger.debug("Could not close realtime client websocket; it is already gone")
+    finally:
+        await _release_realtime_budget_reservation(user_api_key_dict)
 
 
 ######################################################################
diff --git a/litellm/proxy/spend_tracking/budget_reservation.py b/litellm/proxy/spend_tracking/budget_reservation.py
index 91d2ece7a51..2ee7320b82c 100644
--- a/litellm/proxy/spend_tracking/budget_reservation.py
+++ b/litellm/proxy/spend_tracking/budget_reservation.py
@@ -373,6 +373,31 @@ async def invalidate_budget_reservation_counters(
         await _invalidate_spend_counter(counter_key=counter_key)
 
 
+async def release_or_invalidate_budget_reservation(
+    budget_reservation: dict | None,  # mutable-ok: stamps finalized on the caller's shared reservation dict
+) -> None:
+    """Reconcile a still-open reservation on a terminal path that settles no cost.
+
+    A failed or upstream-refused request never runs the success cost callback, so
+    its pre-call reservation stays open and keeps the spend counter pinned above
+    real spend until the counter's TTL expires, 429ing later requests on the same
+    key. Release it to zero; if the release itself fails (e.g. the counter store is
+    unreachable) drop the reserved counters directly and mark the reservation
+    finalized so nothing reprocesses it. Idempotent: the finalized guard makes a
+    second call a no-op once success or failure handling already reconciled.
+    """
+    if budget_reservation is None or budget_reservation.get("finalized") is True:
+        return
+    try:
+        await release_budget_reservation(budget_reservation=budget_reservation)
+    except Exception:  # noqa: BLE001  # a cleanup failure must not pin the counter; drop it directly instead
+        verbose_proxy_logger.exception("Failed to release budget reservation; invalidating counters")
+        try:
+            await invalidate_budget_reservation_counters(budget_reservation=budget_reservation)
+        finally:
+            budget_reservation["finalized"] = True
+
+
 async def _get_budget_counters(
     request_body: dict,
     valid_token: UserAPIKeyAuth,
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 bcfacf16205..5352c894f87 100644
--- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
+++ b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
@@ -3229,21 +3229,28 @@ async def test_bidirectional_forward_relays_upstream_policy_close_to_client():
     client_ws.close.assert_awaited_once_with(code=1008, reason=_UPSTREAM_REFUSAL)
 
 
+@pytest.mark.parametrize(
+    "leaked_detail",
+    (
+        pytest.param("sk-live-abcdef0123456789abcdef0123", id="credential"),
+        pytest.param("vertex-int.svc.cluster.local", id="internal-hostname"),
+        pytest.param("/etc/litellm/service-account.json", id="filesystem-path"),
+    ),
+)
 @pytest.mark.asyncio
-async def test_upstream_close_reason_with_a_secret_is_redacted_before_reaching_the_client():
-    """LIT-6973: the relayed close mirrors the handshake path and scrubs credential
-    patterns, so an upstream error echoing a token never reaches the client verbatim."""
-    secret: Final = "sk-live-abcdef0123456789abcdef0123"
+async def test_upstream_close_details_are_scrubbed_before_reaching_the_client(leaked_detail: str):
+    """LIT-6973: the relayed close goes through the proxy's client-facing redaction, so an upstream
+    error echoing a credential, an internal host, or a server path never reaches the client verbatim."""
     client_ws: Final = _client_ws_that_never_sends()
-    upstream_close: Final = ConnectionClosed(Close(1008, f"auth failed for {secret}"), None)
+    upstream_close: Final = ConnectionClosed(Close(1008, f"upstream rejected: {leaked_detail}"), None)
     session: Final = _relay_session(client_ws, _backend_ws_closing_with(upstream_close))
 
     await session.run()
 
     (error_event,) = _error_events_sent_to(client_ws)
-    assert secret not in error_event["error"]["message"]
+    assert leaked_detail not in error_event["error"]["message"]
     relayed_reason: Final = client_ws.close.await_args.kwargs["reason"]
-    assert secret not in relayed_reason
+    assert leaked_detail not in relayed_reason
     assert "REDACTED" in relayed_reason
 
 
diff --git a/tests/test_litellm/proxy/test_proxy_server.py b/tests/test_litellm/proxy/test_proxy_server.py
index d91928a203e..57e2cfc3332 100644
--- a/tests/test_litellm/proxy/test_proxy_server.py
+++ b/tests/test_litellm/proxy/test_proxy_server.py
@@ -9521,6 +9521,89 @@ def test_realtime_websocket_route_aliases_registered():
         )
 
 
+def _lit6973_fake_realtime_ws() -> MagicMock:
+    ws = MagicMock()
+    ws.headers = {}
+    ws.scope = {"headers": [], "type": "websocket"}
+    ws.url = "ws://testserver/v1/realtime"
+    ws.accept = AsyncMock()
+    ws.send_text = AsyncMock()
+    ws.close = AsyncMock()
+    return ws
+
+
+async def _lit6973_drive_refused_realtime_session(reservation: dict) -> None:
+    """Drive realtime_websocket_endpoint through a session the upstream refused.
+
+    route_request resolves normally because the relay handles the refusal
+    internally (sends the error event, closes the client), so neither the
+    success cost callback nor a failure hook runs on _ProxyDBLogger. The
+    endpoint itself must reconcile the pre-call budget reservation, so the
+    real release runs (entries is empty, so it touches no counter store) and
+    the caller asserts on the observable reservation state afterwards."""
+    from litellm.proxy import proxy_server as ps
+
+    user_api_key_dict: Final = UserAPIKeyAuth(api_key="sk-test", token="hashed-token")
+    user_api_key_dict.budget_reservation = reservation
+
+    completed: Final = asyncio.get_running_loop().create_future()
+    completed.set_result(None)
+
+    pre_call: Final = AsyncMock(return_value=({"model": "vertex_ai/gemini-live-2.5-flash"}, MagicMock()))
+    can_call = patch.object(ps, "can_key_call_resolved_model", new=AsyncMock())  # test-quality-ok: no HTTP boundary; fakes in-process auth to reach the finally under test
+    pre = patch.object(ps.ProxyBaseLLMRequestProcessing, "common_processing_pre_call_logic", new=pre_call)  # test-quality-ok: fakes phase-1 wiring; assertion checks observable reservation state
+    route = patch.object(ps, "route_request", new=AsyncMock(return_value=completed))  # test-quality-ok: fakes the relay that already handled the refusal so the session returns normally
+    with can_call, pre, route:
+        await ps.realtime_websocket_endpoint(
+            websocket=_lit6973_fake_realtime_ws(),
+            model="vertex_ai/gemini-live-2.5-flash",
+            intent=None,
+            guardrails=None,
+            user_api_key_dict=user_api_key_dict,
+        )
+
+
+@pytest.mark.asyncio
+async def test_refused_realtime_session_releases_the_budget_reservation():
+    """LIT-6973: reclassifying a refused realtime session as a failure removed the
+    success-path reservation release, so the pre-call reservation stayed open and
+    pinned the key/team/user spend counters, locking the key after a couple of
+    refusals. The endpoint must reconcile it: the reservation ends up finalized."""
+    reservation: Final = {"reserved_cost": 0.55, "input_cost": 0.0, "finalized": False, "entries": []}
+
+    await _lit6973_drive_refused_realtime_session(reservation)
+
+    assert reservation["finalized"] is True
+
+
+@pytest.mark.asyncio
+async def test_release_or_invalidate_falls_back_to_invalidating_the_counters():
+    """If releasing the reservation itself fails (e.g. the counter store is down),
+    the reserved counters must be invalidated directly so the estimate does not
+    stay pinned, and the reservation is finalized so nothing reprocesses it."""
+    from litellm.proxy import proxy_server as ps
+    from litellm.proxy.spend_tracking import budget_reservation as br
+
+    reservation: Final = {
+        "reserved_cost": 0.55,
+        "input_cost": 0.0,
+        "finalized": False,
+        "entries": [{"counter_key": "spend:key:hashed-token"}],
+    }
+    invalidated: Final[list[str]] = []
+
+    async def _record(counter_key: str) -> None:
+        invalidated.append(counter_key)
+
+    failing_release = patch.object(br, "release_budget_reservation", new=AsyncMock(side_effect=RuntimeError("counter store down")))  # test-quality-ok: forces the failure branch; assertion observes which counter key got invalidated
+    sink = patch.object(ps, "_invalidate_spend_counter", new=_record)  # test-quality-ok: fakes the counter-store sink so the invalidated key is observable
+    with failing_release, sink:
+        await br.release_or_invalidate_budget_reservation(budget_reservation=reservation)
+
+    assert invalidated == ["spend:key:hashed-token"]
+    assert reservation["finalized"] is True
+
+
 class TestTransformRequestBannedParams:
     """
     /utils/transform_request applies the same banned-param check as LLM endpoints.

From 1fe87e8e25206c039be5e87a5808a30f47cc3183 Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Sat, 5 Sep 2026 01:11:24 -0700
Subject: [PATCH 08/10] fix(realtime): settle the budget reservation only for
 sessions the success log does not own

The blanket finally release from the previous commit also zeroed the reservation
of successful sessions. Success settlement is enqueued on the logging worker, not
awaited, so the endpoint's finally ran first and released the reservation the cost
callback still had to reconcile, dropping the real spend from the key/team/user
counters.

The relay now stamps a synchronous marker (REALTIME_SESSION_SUCCESS_LOGGED_KEY) on
the shared logging object at the single success-dispatch site, and the endpoint
releases the reservation only when that marker is absent. Refused or failed
sessions, which never log success, still release; successful sessions leave the
reservation for the cost callback to settle to actual spend. Exactly one settler
touches each reservation, so the idempotent reconcile never double-adjusts.
---
 .../litellm_core_utils/realtime_streaming.py  |  4 ++
 litellm/proxy/proxy_server.py                 |  7 ++-
 .../test_realtime_streaming.py                | 32 +++++++++++
 tests/test_litellm/proxy/test_proxy_server.py | 55 +++++++++++++------
 4 files changed, 80 insertions(+), 18 deletions(-)

diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py
index c670278d3fb..75046f2cf87 100644
--- a/litellm/litellm_core_utils/realtime_streaming.py
+++ b/litellm/litellm_core_utils/realtime_streaming.py
@@ -35,6 +35,9 @@ else:
     CLIENT_CONNECTION_CLASS = Any
 
 
+REALTIME_SESSION_SUCCESS_LOGGED_KEY: Final = "realtime_session_success_logged"
+
+
 @dataclass(frozen=True, slots=True)
 class BackendClose:
     code: int
@@ -421,6 +424,7 @@ class RealTimeStreaming:
             self._logging_worker.ensure_initialized_and_enqueue(
                 self.logging_obj.dispatch_success_handlers(self.messages, prefer_async_handlers=True)
             )
+            self.logging_obj.model_call_details[REALTIME_SESSION_SUCCESS_LOGGED_KEY] = True
 
     async def _send_to_backend(self, message: str) -> bool:
         """Send a message to the backend WebSocket.
diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py
index 06cfee45918..7d59dfa86c4 100644
--- a/litellm/proxy/proxy_server.py
+++ b/litellm/proxy/proxy_server.py
@@ -11603,7 +11603,12 @@ async def realtime_websocket_endpoint(
         except Exception:  # noqa: BLE001  # the lower layer may have closed the socket already; closing twice is not an error
             verbose_proxy_logger.debug("Could not close realtime client websocket; it is already gone")
     finally:
-        await _release_realtime_budget_reservation(user_api_key_dict)
+        from litellm.litellm_core_utils.realtime_streaming import (
+            REALTIME_SESSION_SUCCESS_LOGGED_KEY,
+        )
+
+        if not litellm_logging_obj.model_call_details.get(REALTIME_SESSION_SUCCESS_LOGGED_KEY):
+            await _release_realtime_budget_reservation(user_api_key_dict)
 
 
 ######################################################################
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 5352c894f87..9c0f6f59463 100644
--- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
+++ b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
@@ -14,6 +14,7 @@ import litellm
 
 from litellm.integrations.custom_guardrail import CustomGuardrail
 from litellm.litellm_core_utils.realtime_streaming import (
+    REALTIME_SESSION_SUCCESS_LOGGED_KEY,
     RealTimeStreaming,
     client_sent_openai_beta_realtime_header,
 )
@@ -3380,3 +3381,34 @@ async def test_client_hanging_up_with_a_websockets_close_is_not_mistaken_for_the
     assert session.logging.logged_sessions == ((),)
     assert session.logging.logged_failures == ()
     client_ws.close.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_success_logging_stamps_the_reservation_ownership_marker():
+    """LIT-6973: only the success path enqueues the cost callback that settles the
+    session's budget reservation, so it stamps REALTIME_SESSION_SUCCESS_LOGGED_KEY on
+    the shared logging object. The proxy endpoint reads that stamp to decide whether to
+    release the reservation itself, so a logged-as-success session must carry it."""
+    client_ws: Final = _client_ws_that_never_sends()
+    session_created: Final = json.dumps({"type": "session.created", "session": {"id": "sess_1"}}).encode()
+    upstream_close: Final = ConnectionClosed(Close(1000, ""), None)
+    session: Final = _relay_session(client_ws, _backend_ws_closing_with(session_created, upstream_close))
+
+    await session.run()
+
+    assert session.logging.logged_sessions != ()
+    assert session.logging.model_call_details.get(REALTIME_SESSION_SUCCESS_LOGGED_KEY) is True
+
+
+@pytest.mark.asyncio
+async def test_refused_session_does_not_stamp_the_reservation_ownership_marker():
+    """A refused session logs a failure, not a success, so it must not stamp
+    REALTIME_SESSION_SUCCESS_LOGGED_KEY. If it did, the proxy endpoint would skip its
+    own reservation release and the refused session's reservation would stay pinned."""
+    upstream_close: Final = ConnectionClosed(Close(1008, _UPSTREAM_REFUSAL), None)
+    session: Final = _relay_session(_client_ws_that_never_sends(), _backend_ws_closing_with(upstream_close))
+
+    await session.run()
+
+    assert session.logging.logged_failures == (upstream_close,)
+    assert REALTIME_SESSION_SUCCESS_LOGGED_KEY not in session.logging.model_call_details
diff --git a/tests/test_litellm/proxy/test_proxy_server.py b/tests/test_litellm/proxy/test_proxy_server.py
index 57e2cfc3332..697d4c182c7 100644
--- a/tests/test_litellm/proxy/test_proxy_server.py
+++ b/tests/test_litellm/proxy/test_proxy_server.py
@@ -9532,27 +9532,34 @@ def _lit6973_fake_realtime_ws() -> MagicMock:
     return ws
 
 
-async def _lit6973_drive_refused_realtime_session(reservation: dict) -> None:
-    """Drive realtime_websocket_endpoint through a session the upstream refused.
+async def _lit6973_drive_realtime_session(reservation: dict, *, backend_logged_success: bool) -> None:
+    """Drive realtime_websocket_endpoint to just before its budget-reservation finally.
 
-    route_request resolves normally because the relay handles the refusal
-    internally (sends the error event, closes the client), so neither the
-    success cost callback nor a failure hook runs on _ProxyDBLogger. The
-    endpoint itself must reconcile the pre-call budget reservation, so the
-    real release runs (entries is empty, so it touches no counter store) and
-    the caller asserts on the observable reservation state afterwards."""
+    route_request resolves normally in both cases: the relay owns the session
+    once route_request returns. A successful session enqueues its success cost
+    callback and stamps REALTIME_SESSION_SUCCESS_LOGGED_KEY on the shared logging
+    object; a refused one does neither. The endpoint keys its reservation cleanup
+    off that stamp, so backend_logged_success reproduces both branches. The fake
+    logging object carries a real model_call_details dict so the stamp is
+    observable, and the reservation has empty entries so the real release touches
+    no counter store."""
+    from litellm.litellm_core_utils.realtime_streaming import REALTIME_SESSION_SUCCESS_LOGGED_KEY
     from litellm.proxy import proxy_server as ps
 
     user_api_key_dict: Final = UserAPIKeyAuth(api_key="sk-test", token="hashed-token")
     user_api_key_dict.budget_reservation = reservation
 
-    completed: Final = asyncio.get_running_loop().create_future()
-    completed.set_result(None)
+    logging_obj: Final = MagicMock()
+    logging_obj.model_call_details = {}
 
-    pre_call: Final = AsyncMock(return_value=({"model": "vertex_ai/gemini-live-2.5-flash"}, MagicMock()))
+    async def fake_llm_call() -> None:
+        if backend_logged_success:
+            logging_obj.model_call_details[REALTIME_SESSION_SUCCESS_LOGGED_KEY] = True
+
+    pre_call: Final = AsyncMock(return_value=({"model": "vertex_ai/gemini-live-2.5-flash"}, logging_obj))
     can_call = patch.object(ps, "can_key_call_resolved_model", new=AsyncMock())  # test-quality-ok: no HTTP boundary; fakes in-process auth to reach the finally under test
     pre = patch.object(ps.ProxyBaseLLMRequestProcessing, "common_processing_pre_call_logic", new=pre_call)  # test-quality-ok: fakes phase-1 wiring; assertion checks observable reservation state
-    route = patch.object(ps, "route_request", new=AsyncMock(return_value=completed))  # test-quality-ok: fakes the relay that already handled the refusal so the session returns normally
+    route = patch.object(ps, "route_request", new=AsyncMock(return_value=fake_llm_call()))  # test-quality-ok: fakes the relay whose success/refusal outcome the endpoint reads off the logging object
     with can_call, pre, route:
         await ps.realtime_websocket_endpoint(
             websocket=_lit6973_fake_realtime_ws(),
@@ -9565,17 +9572,31 @@ async def _lit6973_drive_refused_realtime_session(reservation: dict) -> None:
 
 @pytest.mark.asyncio
 async def test_refused_realtime_session_releases_the_budget_reservation():
-    """LIT-6973: reclassifying a refused realtime session as a failure removed the
-    success-path reservation release, so the pre-call reservation stayed open and
-    pinned the key/team/user spend counters, locking the key after a couple of
-    refusals. The endpoint must reconcile it: the reservation ends up finalized."""
+    """LIT-6973: a refused realtime session enqueues no success cost callback, so
+    the pre-call reservation would stay open and pin the key/team/user spend
+    counters, locking the key after a couple of refusals. The endpoint sees no
+    success stamp and reconciles it: the reservation ends up finalized."""
     reservation: Final = {"reserved_cost": 0.55, "input_cost": 0.0, "finalized": False, "entries": []}
 
-    await _lit6973_drive_refused_realtime_session(reservation)
+    await _lit6973_drive_realtime_session(reservation, backend_logged_success=False)
 
     assert reservation["finalized"] is True
 
 
+@pytest.mark.asyncio
+async def test_successful_realtime_session_leaves_the_reservation_for_the_cost_callback():
+    """A billable realtime session settles its reservation through the enqueued
+    success cost callback, not the endpoint. The endpoint must not finalize it in
+    its finally, or it would reconcile the reservation to zero before the cost
+    callback applies real spend, so billable sessions stop counting against budget.
+    With the success stamp present, the endpoint leaves the reservation untouched."""
+    reservation: Final = {"reserved_cost": 0.55, "input_cost": 0.0, "finalized": False, "entries": []}
+
+    await _lit6973_drive_realtime_session(reservation, backend_logged_success=True)
+
+    assert reservation["finalized"] is False
+
+
 @pytest.mark.asyncio
 async def test_release_or_invalidate_falls_back_to_invalidating_the_counters():
     """If releasing the reservation itself fails (e.g. the counter store is down),

From 5a35e6d41f76d2b09a258b3e2b051e3e7e745c79 Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Sat, 5 Sep 2026 02:18:33 -0700
Subject: [PATCH 09/10] fix(realtime): release the budget reservation when a
 session is rejected before the relay starts

The three pre-relay exits of realtime_websocket_endpoint (missing model,
key/model access denied, pre-call rejection such as a rate limit or a
guardrail) returned before the finally that releases the auth-time budget
reservation, so a rejected session pinned the key at the reserved amount
until the counter TTL expired and its next requests got budget_exceeded
while /key/info showed spend 0. A single _reject_realtime_session helper
now releases the reservation before sending the error event and closing,
and release_or_invalidate_budget_reservation shields the release from a
second cancellation and logs, rather than raises, a failing invalidate
fallback so it can never mask the session's own outcome.
---
 litellm/proxy/proxy_server.py                 | 43 ++++++----
 .../spend_tracking/budget_reservation.py      |  4 +-
 tests/test_litellm/proxy/test_proxy_server.py | 79 +++++++++++++++++--
 3 files changed, 103 insertions(+), 23 deletions(-)

diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py
index 7d59dfa86c4..65fe3ede822 100644
--- a/litellm/proxy/proxy_server.py
+++ b/litellm/proxy/proxy_server.py
@@ -11463,6 +11463,25 @@ async def _release_realtime_budget_reservation(user_api_key_dict: UserAPIKeyAuth
     )
 
 
+async def _reject_realtime_session(
+    websocket: WebSocket,
+    user_api_key_dict: UserAPIKeyAuth,
+    *,
+    code: int,
+    reason: str,
+    error_message: str | None = None,
+) -> None:
+    await _release_realtime_budget_reservation(user_api_key_dict)
+    if error_message is not None:
+        try:
+            await websocket.send_text(
+                json.dumps({"type": "error", "error": {"type": "guardrail_error", "message": error_message}})
+            )
+        except Exception:  # noqa: BLE001  # best-effort notice: a dead client socket must not skip the close below
+            verbose_proxy_logger.debug("Could not send realtime pre-call error event to client; closing anyway")
+    await websocket.close(code=code, reason=reason)
+
+
 @app.websocket("/openai/v1/realtime")
 @app.websocket("/v1/realtime")
 @app.websocket("/realtime")
@@ -11488,7 +11507,9 @@ async def realtime_websocket_endpoint(
         if intent == "transcription":
             route_model = "gpt-realtime-whisper"
         else:
-            await websocket.close(code=1008, reason="model query parameter is required")
+            await _reject_realtime_session(
+                websocket, user_api_key_dict, code=1008, reason="model query parameter is required"
+            )
             return
     assert route_model is not None
     try:
@@ -11499,7 +11520,7 @@ async def realtime_websocket_endpoint(
             llm_router=llm_router,
         )
     except ProxyException as e:
-        await websocket.close(code=1008, reason=e.message[:120])
+        await _reject_realtime_session(websocket, user_api_key_dict, code=1008, reason=e.message[:120])
         return
     await websocket.accept(**accept_kwargs)
 
@@ -11558,21 +11579,9 @@ async def realtime_websocket_endpoint(
         )
     except Exception as e:
         verbose_proxy_logger.exception("Realtime pre-call error")
-        try:
-            await websocket.send_text(
-                json.dumps(
-                    {
-                        "type": "error",
-                        "error": {
-                            "type": "guardrail_error",
-                            "message": str(e),
-                        },
-                    }
-                )
-            )
-        except Exception:
-            pass
-        await websocket.close(code=1011, reason="Pre-call error")
+        await _reject_realtime_session(
+            websocket, user_api_key_dict, code=1011, reason="Pre-call error", error_message=str(e)
+        )
         return
 
     # Phase 2: route to upstream LLM.
diff --git a/litellm/proxy/spend_tracking/budget_reservation.py b/litellm/proxy/spend_tracking/budget_reservation.py
index 2ee7320b82c..ed2bc87597c 100644
--- a/litellm/proxy/spend_tracking/budget_reservation.py
+++ b/litellm/proxy/spend_tracking/budget_reservation.py
@@ -389,11 +389,13 @@ async def release_or_invalidate_budget_reservation(
     if budget_reservation is None or budget_reservation.get("finalized") is True:
         return
     try:
-        await release_budget_reservation(budget_reservation=budget_reservation)
+        await asyncio.shield(release_budget_reservation(budget_reservation=budget_reservation))
     except Exception:  # noqa: BLE001  # a cleanup failure must not pin the counter; drop it directly instead
         verbose_proxy_logger.exception("Failed to release budget reservation; invalidating counters")
         try:
             await invalidate_budget_reservation_counters(budget_reservation=budget_reservation)
+        except Exception:  # noqa: BLE001  # nothing left to try; the finalized stamp below keeps it from being reprocessed
+            verbose_proxy_logger.exception("Failed to invalidate budget reservation counters after release failed")
         finally:
             budget_reservation["finalized"] = True
 
diff --git a/tests/test_litellm/proxy/test_proxy_server.py b/tests/test_litellm/proxy/test_proxy_server.py
index 697d4c182c7..8b151d9e1f6 100644
--- a/tests/test_litellm/proxy/test_proxy_server.py
+++ b/tests/test_litellm/proxy/test_proxy_server.py
@@ -9532,8 +9532,15 @@ def _lit6973_fake_realtime_ws() -> MagicMock:
     return ws
 
 
-async def _lit6973_drive_realtime_session(reservation: dict, *, backend_logged_success: bool) -> None:
-    """Drive realtime_websocket_endpoint to just before its budget-reservation finally.
+async def _lit6973_drive_realtime_session(
+    reservation: dict, *, backend_logged_success: bool, phase_one_exit: str | None = None
+) -> MagicMock:
+    """Drive realtime_websocket_endpoint through one of its reservation-settling exits.
+
+    phase_one_exit picks a rejection before the relay: "model_access" makes the
+    key/model check raise ProxyException, "pre_call" makes pre-call processing
+    (rate limits, guardrails) raise. Neither reaches route_request, so no success
+    log can own the reservation and the endpoint has to release it on that exit.
 
     route_request resolves normally in both cases: the relay owns the session
     once route_request returns. A successful session enqueues its success cost
@@ -9556,18 +9563,30 @@ async def _lit6973_drive_realtime_session(reservation: dict, *, backend_logged_s
         if backend_logged_success:
             logging_obj.model_call_details[REALTIME_SESSION_SUCCESS_LOGGED_KEY] = True
 
-    pre_call: Final = AsyncMock(return_value=({"model": "vertex_ai/gemini-live-2.5-flash"}, logging_obj))
-    can_call = patch.object(ps, "can_key_call_resolved_model", new=AsyncMock())  # test-quality-ok: no HTTP boundary; fakes in-process auth to reach the finally under test
+    from litellm.proxy._types import ProxyException
+
+    model_access_error: Final = (
+        ProxyException(message="key cannot access model", type="auth_error", param="model", code=401)
+        if phase_one_exit == "model_access"
+        else None
+    )
+    pre_call_error: Final = Exception("Rate limit exceeded") if phase_one_exit == "pre_call" else None
+    pre_call: Final = AsyncMock(
+        side_effect=pre_call_error, return_value=({"model": "vertex_ai/gemini-live-2.5-flash"}, logging_obj)
+    )
+    ws: Final = _lit6973_fake_realtime_ws()
+    can_call = patch.object(ps, "can_key_call_resolved_model", new=AsyncMock(side_effect=model_access_error))  # test-quality-ok: no HTTP boundary; fakes in-process auth to reach the exit under test
     pre = patch.object(ps.ProxyBaseLLMRequestProcessing, "common_processing_pre_call_logic", new=pre_call)  # test-quality-ok: fakes phase-1 wiring; assertion checks observable reservation state
     route = patch.object(ps, "route_request", new=AsyncMock(return_value=fake_llm_call()))  # test-quality-ok: fakes the relay whose success/refusal outcome the endpoint reads off the logging object
     with can_call, pre, route:
         await ps.realtime_websocket_endpoint(
-            websocket=_lit6973_fake_realtime_ws(),
+            websocket=ws,
             model="vertex_ai/gemini-live-2.5-flash",
             intent=None,
             guardrails=None,
             user_api_key_dict=user_api_key_dict,
         )
+    return ws
 
 
 @pytest.mark.asyncio
@@ -9583,6 +9602,39 @@ async def test_refused_realtime_session_releases_the_budget_reservation():
     assert reservation["finalized"] is True
 
 
+@pytest.mark.asyncio
+async def test_realtime_session_rejected_in_pre_call_releases_the_budget_reservation():
+    """A rate-limit or guardrail rejection happens before route_request, so the
+    relay never runs and no success log can own the reservation. The endpoint
+    must release it on that exit too, or the key stays pinned at the reserved
+    amount and its next requests 429 with budget_exceeded while /key/info shows
+    spend 0 (reproduced live with rpm_limit=1). The client still gets the
+    pre-call error event and the 1011 close it got before."""
+    reservation: Final = {"reserved_cost": 0.55, "input_cost": 0.0, "finalized": False, "entries": []}
+
+    ws: Final = await _lit6973_drive_realtime_session(
+        reservation, backend_logged_success=False, phase_one_exit="pre_call"
+    )
+
+    assert reservation["finalized"] is True
+    assert json.loads(ws.send_text.await_args.args[0])["error"]["message"] == "Rate limit exceeded"
+    ws.close.assert_awaited_once_with(code=1011, reason="Pre-call error")
+
+
+@pytest.mark.asyncio
+async def test_realtime_session_denied_model_access_releases_the_budget_reservation():
+    """The key/model access check rejects before the socket is even accepted;
+    that exit skipped the release as well, pinning the reservation."""
+    reservation: Final = {"reserved_cost": 0.55, "input_cost": 0.0, "finalized": False, "entries": []}
+
+    ws: Final = await _lit6973_drive_realtime_session(
+        reservation, backend_logged_success=False, phase_one_exit="model_access"
+    )
+
+    assert reservation["finalized"] is True
+    ws.close.assert_awaited_once_with(code=1008, reason="key cannot access model")
+
+
 @pytest.mark.asyncio
 async def test_successful_realtime_session_leaves_the_reservation_for_the_cost_callback():
     """A billable realtime session settles its reservation through the enqueued
@@ -9625,6 +9677,23 @@ async def test_release_or_invalidate_falls_back_to_invalidating_the_counters():
     assert reservation["finalized"] is True
 
 
+@pytest.mark.asyncio
+async def test_release_or_invalidate_finalizes_even_when_the_invalidate_fallback_fails():
+    """Both counter-store calls failing must not raise out of the realtime
+    endpoint's finally (it would mask the session's own outcome) and must still
+    stamp finalized so nothing retries the same reservation."""
+    from litellm.proxy.spend_tracking import budget_reservation as br
+
+    reservation: Final = {"reserved_cost": 0.55, "input_cost": 0.0, "finalized": False, "entries": []}
+    failing_release = patch.object(br, "release_budget_reservation", new=AsyncMock(side_effect=RuntimeError("counter store down")))  # test-quality-ok: forces the fallback branch
+    failing_invalidate = patch.object(br, "invalidate_budget_reservation_counters", new=AsyncMock(side_effect=RuntimeError("still down")))  # test-quality-ok: forces the fallback itself to fail
+
+    with failing_release, failing_invalidate:
+        await br.release_or_invalidate_budget_reservation(budget_reservation=reservation)
+
+    assert reservation["finalized"] is True
+
+
 class TestTransformRequestBannedParams:
     """
     /utils/transform_request applies the same banned-param check as LLM endpoints.

From 37722eba68149c5f3e59ed0f3c12798a84aa1bc4 Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Sat, 5 Sep 2026 02:35:26 -0700
Subject: [PATCH 10/10] fix(realtime): close a rejected client before releasing
 its budget reservation

A slow or unreachable counter store made a pre-relay rejection wait behind
the reservation release before the client saw the error event and the close.
Close first and release in finally, mirroring the relay's own failure path,
so a client that already hung up still gets its reservation released.
---
 litellm/proxy/proxy_server.py                 | 20 ++++----
 tests/test_litellm/proxy/test_proxy_server.py | 47 ++++++++++++++++++-
 2 files changed, 56 insertions(+), 11 deletions(-)

diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py
index 65fe3ede822..a5c60d8e976 100644
--- a/litellm/proxy/proxy_server.py
+++ b/litellm/proxy/proxy_server.py
@@ -11471,15 +11471,17 @@ async def _reject_realtime_session(
     reason: str,
     error_message: str | None = None,
 ) -> None:
-    await _release_realtime_budget_reservation(user_api_key_dict)
-    if error_message is not None:
-        try:
-            await websocket.send_text(
-                json.dumps({"type": "error", "error": {"type": "guardrail_error", "message": error_message}})
-            )
-        except Exception:  # noqa: BLE001  # best-effort notice: a dead client socket must not skip the close below
-            verbose_proxy_logger.debug("Could not send realtime pre-call error event to client; closing anyway")
-    await websocket.close(code=code, reason=reason)
+    try:
+        if error_message is not None:
+            try:
+                await websocket.send_text(
+                    json.dumps({"type": "error", "error": {"type": "guardrail_error", "message": error_message}})
+                )
+            except Exception:  # noqa: BLE001  # best-effort notice: a dead client socket must not skip the close below
+                verbose_proxy_logger.debug("Could not send realtime pre-call error event to client; closing anyway")
+        await websocket.close(code=code, reason=reason)
+    finally:
+        await _release_realtime_budget_reservation(user_api_key_dict)
 
 
 @app.websocket("/openai/v1/realtime")
diff --git a/tests/test_litellm/proxy/test_proxy_server.py b/tests/test_litellm/proxy/test_proxy_server.py
index 8b151d9e1f6..4d70c9d436f 100644
--- a/tests/test_litellm/proxy/test_proxy_server.py
+++ b/tests/test_litellm/proxy/test_proxy_server.py
@@ -9533,7 +9533,11 @@ def _lit6973_fake_realtime_ws() -> MagicMock:
 
 
 async def _lit6973_drive_realtime_session(
-    reservation: dict, *, backend_logged_success: bool, phase_one_exit: str | None = None
+    reservation: dict,
+    *,
+    backend_logged_success: bool,
+    phase_one_exit: str | None = None,
+    websocket: MagicMock | None = None,
 ) -> MagicMock:
     """Drive realtime_websocket_endpoint through one of its reservation-settling exits.
 
@@ -9574,7 +9578,7 @@ async def _lit6973_drive_realtime_session(
     pre_call: Final = AsyncMock(
         side_effect=pre_call_error, return_value=({"model": "vertex_ai/gemini-live-2.5-flash"}, logging_obj)
     )
-    ws: Final = _lit6973_fake_realtime_ws()
+    ws: Final = websocket if websocket is not None else _lit6973_fake_realtime_ws()
     can_call = patch.object(ps, "can_key_call_resolved_model", new=AsyncMock(side_effect=model_access_error))  # test-quality-ok: no HTTP boundary; fakes in-process auth to reach the exit under test
     pre = patch.object(ps.ProxyBaseLLMRequestProcessing, "common_processing_pre_call_logic", new=pre_call)  # test-quality-ok: fakes phase-1 wiring; assertion checks observable reservation state
     route = patch.object(ps, "route_request", new=AsyncMock(return_value=fake_llm_call()))  # test-quality-ok: fakes the relay whose success/refusal outcome the endpoint reads off the logging object
@@ -9635,6 +9639,45 @@ async def test_realtime_session_denied_model_access_releases_the_budget_reservat
     ws.close.assert_awaited_once_with(code=1008, reason="key cannot access model")
 
 
+@pytest.mark.asyncio
+async def test_rejected_realtime_session_closes_the_client_before_releasing_the_reservation():
+    """The counter release can block on a slow or unreachable store, and a
+    rejected client must not sit behind it: the relay's own failure path closes
+    the client first and releases in its finally, so the pre-relay rejection
+    has to close first as well. The fake close checks the reservation is still
+    open when the client is closed."""
+    reservation: Final = {"reserved_cost": 0.55, "input_cost": 0.0, "finalized": False, "entries": []}
+    ws: Final = _lit6973_fake_realtime_ws()
+
+    async def close_while_reservation_is_still_open(**_: object) -> None:
+        assert reservation["finalized"] is False, "client was closed only after the reservation release"
+
+    ws.close = AsyncMock(side_effect=close_while_reservation_is_still_open)
+
+    await _lit6973_drive_realtime_session(
+        reservation, backend_logged_success=False, phase_one_exit="pre_call", websocket=ws
+    )
+
+    ws.close.assert_awaited_once_with(code=1011, reason="Pre-call error")
+    assert reservation["finalized"] is True
+
+
+@pytest.mark.asyncio
+async def test_rejected_realtime_session_releases_the_reservation_when_the_client_is_already_gone():
+    """A client that hung up before the rejection makes the close raise; the
+    reservation must still be released, or the key stays pinned."""
+    reservation: Final = {"reserved_cost": 0.55, "input_cost": 0.0, "finalized": False, "entries": []}
+    ws: Final = _lit6973_fake_realtime_ws()
+    ws.close = AsyncMock(side_effect=RuntimeError("client already disconnected"))
+
+    with pytest.raises(RuntimeError, match="client already disconnected"):
+        await _lit6973_drive_realtime_session(
+            reservation, backend_logged_success=False, phase_one_exit="model_access", websocket=ws
+        )
+
+    assert reservation["finalized"] is True
+
+
 @pytest.mark.asyncio
 async def test_successful_realtime_session_leaves_the_reservation_for_the_cost_callback():
     """A billable realtime session settles its reservation through the enqueued