Merge pull request #38563 from BerriAI/litellm_lit6324_flush_trailing_live_audio

fix(realtime): bill trailing audio when a Gemini transcribe Live session closes

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

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,

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,

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
+    )

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