diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index 2da63554b75..9125ed6e70a 100644 --- a/litellm/litellm_core_utils/realtime_streaming.py +++ b/litellm/litellm_core_utils/realtime_streaming.py @@ -330,6 +330,24 @@ class RealTimeStreaming: except (AttributeError, TypeError): pass + def _flush_unbilled_transcription_usage(self) -> None: + if self.provider_config is None: + return + usage: Final = self.provider_config.unbilled_usage_on_session_close(self.model) + if usage is None: + return + flush_event: Final = ( + cast( # cast-ok: usage-only partial event, the same shape _capture_transcription_usage logs + OpenAIRealtimeEvents, + { + "type": "conversation.item.input_audio_transcription.completed", + "usage": usage, + }, + ) + ) + self.store_message(flush_event) + self._capture_transcription_usage(flush_event) + def _collect_tool_calls_from_response_done(self, event_obj: dict | OpenAIRealtimeEvents) -> None: """Extract function_call items from response.done events for spend logging.""" try: @@ -1069,6 +1087,7 @@ class RealTimeStreaming: except Exception as e: verbose_logger.exception("Error in backend to client send messages: %s", e) finally: + self._flush_unbilled_transcription_usage() await self.log_messages() @staticmethod diff --git a/litellm/llms/base_llm/realtime/transformation.py b/litellm/llms/base_llm/realtime/transformation.py index 26c189504df..cfcde7c6e9e 100644 --- a/litellm/llms/base_llm/realtime/transformation.py +++ b/litellm/llms/base_llm/realtime/transformation.py @@ -5,6 +5,7 @@ import httpx from litellm.types.llms.openai import OpenAIRealtimeStreamSessionEvents from litellm.types.realtime import ( + RealtimeInputAudioTranscriptionUsage, RealtimeResponseTransformInput, RealtimeResponseTypedDict, ) @@ -70,6 +71,9 @@ class BaseRealtimeConfig(ABC): def session_configuration_request(self, model: str) -> str | None: # message sent to setup the realtime session return None + def unbilled_usage_on_session_close(self, model: str) -> RealtimeInputAudioTranscriptionUsage | None: + return None + def transform_session_created_event( self, model: str, diff --git a/litellm/llms/gemini/realtime/transformation.py b/litellm/llms/gemini/realtime/transformation.py index a3b6381306e..367619db37d 100644 --- a/litellm/llms/gemini/realtime/transformation.py +++ b/litellm/llms/gemini/realtime/transformation.py @@ -1191,6 +1191,9 @@ class GeminiRealtimeConfig(BaseRealtimeConfig): } return usage + def unbilled_usage_on_session_close(self, model: str) -> RealtimeInputAudioTranscriptionUsage | None: + return self._consume_input_transcription_usage_estimate(model) + def transform_realtime_response( self, message: str | bytes, 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 1b71c2f1f9b..52e88db753a 100644 --- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py +++ b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py @@ -3019,3 +3019,95 @@ async def test_provider_config_path_captures_transcription_usage(): and message.get("usage") == usage ) assert len(usage_events) == 1 + + +@pytest.mark.asyncio +async def test_session_close_flushes_unbilled_transcription_usage(): + """Trailing audio appended after the last transcript frame must still be billed: + on session close the provider's unbilled estimate is flushed into the logged + messages before log_messages runs, and never forwarded to the client.""" + from typing import Final + + from litellm.types.realtime import RealtimeInputAudioTranscriptionUsage + + client_ws: Final = MagicMock() + client_ws.send_text = AsyncMock() + backend_ws: Final = MagicMock() + backend_ws.recv = AsyncMock(side_effect=ConnectionClosed(None, None)) + logging_obj: Final = MagicMock() + logging_obj.async_success_handler = AsyncMock() + logging_obj.success_handler = MagicMock() + + usage: Final[RealtimeInputAudioTranscriptionUsage] = { + "type": "tokens", + "input_tokens": 153, + "output_tokens": 18, + "total_tokens": 171, + "input_token_details": {"text_tokens": 0, "audio_tokens": 153}, + } + provider_config: Final = MagicMock() + provider_config.unbilled_usage_on_session_close = MagicMock(return_value=usage) + + streaming: Final = RealTimeStreaming( + client_ws, + backend_ws, + logging_obj, + provider_config=provider_config, + model="gemini-3.5-transcribe-live", + ) + logged_snapshots: Final[list[tuple]] = [] + + original_log_messages: Final = streaming.log_messages + + async def _snapshot_then_log(): + logged_snapshots.append(tuple(streaming.messages)) + await original_log_messages() + + streaming.log_messages = _snapshot_then_log + + await streaming.backend_to_client_send_messages() + + provider_config.unbilled_usage_on_session_close.assert_called_once_with("gemini-3.5-transcribe-live") + flushed: Final = tuple( + message + for message in streaming.messages + if isinstance(message, dict) + and message.get("type") == "conversation.item.input_audio_transcription.completed" + and message.get("usage") == usage + ) + assert len(flushed) == 1 + assert flushed[0] in logged_snapshots[0] + assert not client_ws.send_text.called + + +@pytest.mark.asyncio +async def test_session_close_flush_noop_without_unbilled_usage(): + """Everything already billed mid-stream: the session-close flush must not append + a duplicate transcription event.""" + from typing import Final + + client_ws: Final = MagicMock() + client_ws.send_text = AsyncMock() + backend_ws: Final = MagicMock() + backend_ws.recv = AsyncMock(side_effect=ConnectionClosed(None, None)) + logging_obj: Final = MagicMock() + logging_obj.async_success_handler = AsyncMock() + logging_obj.success_handler = MagicMock() + + provider_config: Final = MagicMock() + provider_config.unbilled_usage_on_session_close = MagicMock(return_value=None) + + streaming: Final = RealTimeStreaming( + client_ws, + backend_ws, + logging_obj, + provider_config=provider_config, + model="gemini-3.5-transcribe-live", + ) + + await streaming.backend_to_client_send_messages() + + assert not any( + isinstance(message, dict) and message.get("type") == "conversation.item.input_audio_transcription.completed" + for message in streaming.messages + ) diff --git a/tests/test_litellm/llms/gemini/realtime/test_gemini_realtime_transformation.py b/tests/test_litellm/llms/gemini/realtime/test_gemini_realtime_transformation.py index c362efbfffa..42994dbd2af 100644 --- a/tests/test_litellm/llms/gemini/realtime/test_gemini_realtime_transformation.py +++ b/tests/test_litellm/llms/gemini/realtime/test_gemini_realtime_transformation.py @@ -2119,3 +2119,27 @@ def test_non_transcription_live_model_completed_event_has_no_usage(patch_gemini_ ) assert len(completed) == 1 assert "usage" not in completed[0] + + +def test_unbilled_usage_on_session_close_flushes_trailing_audio(patch_gemini_transcribe_live_cost_map_entry): + """Audio appended after the last transcript frame is still unbilled when the + session closes; the session-close hook must hand back the estimate exactly once + so the streaming layer can bill it (144000 pcm16 bytes = 3s -> 75 in / 9 out).""" + from typing import Final + + from litellm.types.realtime import RealtimeInputAudioTranscriptionUsage + + config: Final = GeminiRealtimeConfig() + config.transform_realtime_request(_input_audio_append_message(144000), "gemini-3.5-transcribe-live") + + usage: Final = config.unbilled_usage_on_session_close("gemini-3.5-transcribe-live") + + expected: Final[RealtimeInputAudioTranscriptionUsage] = { + "type": "tokens", + "input_tokens": 75, + "output_tokens": 9, + "total_tokens": 84, + "input_token_details": {"text_tokens": 0, "audio_tokens": 75}, + } + assert usage == expected + assert config.unbilled_usage_on_session_close("gemini-3.5-transcribe-live") is None