test(realtime): verify wire rates with typed helpers

litellm/llms/gemini/realtime/transformation.py

@@ -133,23 +133,13 @@ def _grounding_metadata_from_frame(frame: Mapping[str, object]) -> tuple[Mapping
 GEMINI_LIVE_TRANSCRIBE_AUDIO_TOKENS_PER_SECOND: Final = 25
 GEMINI_LIVE_TRANSCRIBE_OUTPUT_TEXT_TOKENS_PER_MINUTE: Final = 175
 
-# Live API input audio is natively 16kHz; 24kHz is the *output* rate. Per
-# ai.google.dev/gemini-api/docs/live-api/capabilities: "Audio output always uses a sample
-# rate of 24kHz. Input audio is natively 16kHz ... To convey the sample rate of input
-# audio, set the MIME type of each audio-containing Blob to a value like
-# audio/pcm;rate=16000." The MIME rate is what the server resamples against, so declaring
-# the output rate on the input path mislabels correctly-encoded audio.
+# The MIME rate controls server resampling and must describe the input bytes.
 GEMINI_LIVE_INPUT_AUDIO_SAMPLE_RATE_HZ: Final = 16000
 PCM16_BYTES_PER_SAMPLE: Final = 2
-# The declared rate is client-controlled and feeds the transcription spend estimate, so only accept
-# rates that real PCM audio actually uses. Outside this range the declaration is ignored and the
-# native default stands, which bounds how far a bogus rate can move a bill.
+# Bound client-controlled rates because they affect the duration billed.
 MIN_ACCEPTED_INPUT_AUDIO_SAMPLE_RATE_HZ: Final = 8000
 MAX_ACCEPTED_INPUT_AUDIO_SAMPLE_RATE_HZ: Final = 48000
-# The beta session shape has no rate field, but its ``input_audio_format`` codec name carries one
-# by definition. LiteLLM's own type stub for it says pcm16 input "must be 16-bit PCM at a 24kHz
-# sample rate" (``OpenAIRealtimeSession.input_audio_format`` in litellm/types/llms/openai.py), and
-# the beta-to-GA converter in realtime_streaming.py already expands the name to that rate.
+# The beta pcm16 codec implies 24kHz, matching the beta-to-GA conversion.
 BETA_PCM16_INPUT_AUDIO_SAMPLE_RATE_HZ: Final = 24000
 
 
@@ -167,11 +157,8 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
         # Gemini Live sometimes emits usageMetadata in a standalone frame between
         # turns; buffer it here so the next response.done carries the token counts.
         self._pending_usage_metadata: dict | None = None
-        # Seconds, not bytes: each chunk is converted at the rate declared when it arrived, so a
-        # later session.update cannot reprice audio the backend has already processed.
+        # Store per-chunk duration so later rate changes cannot reprice earlier audio.
         self._unbilled_input_audio_seconds: float = 0.0
-        # Overwritten from session.update when the client declares a rate; see
-        # _record_input_audio_sample_rate.
         self._input_audio_sample_rate_hz: int = GEMINI_LIVE_INPUT_AUDIO_SAMPLE_RATE_HZ
 
     def is_setup_message(self, msg_obj: dict) -> bool:
@@ -488,7 +475,6 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
 
     @staticmethod
     def _declared_rate_from_ga_shape(session_payload: Mapping[str, object]) -> object:
-        """Read ``audio.input.format.rate`` out of the GA nested session shape."""
         audio = session_payload.get("audio")
         if not isinstance(audio, dict):
             return None
@@ -502,40 +488,16 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
 
     @staticmethod
     def _declared_rate_from_beta_shape(session_payload: Mapping[str, object]) -> object:
-        """Read the rate implied by the flat beta ``input_audio_format`` codec name.
-
-        Only pcm16 is mapped. ``get_audio_mime_type`` labels every append as pcm16, so a rate
-        lifted from a g711 name would describe bytes with a codec they are not in.
-        """
+        """Only pcm16 implies a rate here because append frames are sent as pcm16."""
         if session_payload.get("input_audio_format") == "pcm16":
             return BETA_PCM16_INPUT_AUDIO_SAMPLE_RATE_HZ
         return None
 
     def _record_input_audio_sample_rate(self, session_payload: Mapping[str, object]) -> None:
-        """
-        Remember the input sample rate the client declared on session.update.
-
-        The rate reaches Gemini only through the per-blob MIME type, and the server resamples
-        against whatever that MIME type claims, so it has to describe the bytes actually sent.
-
-        Both session shapes can declare a rate. The GA shape states it outright in
-        ``audio.input.format.rate``. The beta shape has no rate field, but its
-        ``input_audio_format`` codec name implies one, and pcm16 is specified as 24kHz. Both are
-        read here because which shape reaches this method is decided upstream by the
-        ``OpenAI-Beta`` header: without it, ``RealTimeStreaming._remap_beta_session_to_ga``
-        rewrites the flat payload into the GA shape and supplies that same 24kHz for pcm16; with
-        it, the flat payload arrives untouched. Reading only the GA shape would label one
-        client's audio 24kHz and an identical client's 16kHz over a header that says nothing
-        about sample rates.
-
-        A change here only affects audio that arrives after it: already-buffered audio was
-        converted to seconds at the rate in force when it was appended, so a mid-stream
-        redeclaration cannot retroactively reprice it.
-        """
+        """Read both shapes so the OpenAI-Beta header cannot change the audio's declared rate."""
         rate = self._declared_rate_from_ga_shape(session_payload)
         if rate is None:
             rate = self._declared_rate_from_beta_shape(session_payload)
-        # bool is an int subclass, so exclude it explicitly.
         if isinstance(rate, bool) or not isinstance(rate, int):
             return
         if not (MIN_ACCEPTED_INPUT_AUDIO_SAMPLE_RATE_HZ <= rate <= MAX_ACCEPTED_INPUT_AUDIO_SAMPLE_RATE_HZ):

tests/unit/llms/gemini/realtime/test_gemini_realtime_transformation.py

@@ -1,9 +1,10 @@
 import json
 from collections.abc import Mapping
-from typing import cast
+from typing import Final, cast
 from unittest.mock import MagicMock
 
 import pytest
+from pydantic import TypeAdapter
 
 
 import litellm
@@ -2007,14 +2008,16 @@ def _input_audio_append_message(raw_byte_count: int) -> str:
     )
 
 
-def _session_update_message(session: dict) -> str:
-    return json.dumps({"type": "session.update", "session": session})
+def _session_update_message(session: Mapping[str, object]) -> str:
+    return json.dumps({"type": "session.update", "session": dict(session)})
 
 
-def _sent_audio_mime_type(config, raw_byte_count: int = 32000) -> str:
-    """Round-trips one input_audio_buffer.append and returns the mimeType actually put on the wire."""
-    sent = config.transform_realtime_request(_input_audio_append_message(raw_byte_count), "gemini-3.5-transcribe-live")
-    return json.loads(sent[0])["realtimeInput"]["audio"]["mimeType"]
+def _sent_audio_mime_type(config: GeminiRealtimeConfig, raw_byte_count: int = 32000) -> str:
+    sent: Final = config.transform_realtime_request(
+        _input_audio_append_message(raw_byte_count), "gemini-3.5-transcribe-live"
+    )
+    frame: Final = TypeAdapter(dict[str, dict[str, dict[str, str]]]).validate_json(sent[0])
+    return frame["realtimeInput"]["audio"]["mimeType"]
 
 
 def test_input_audio_mime_type_declares_the_native_16khz_input_rate():
@@ -2028,16 +2031,15 @@ def test_input_audio_mime_type_declares_the_native_16khz_input_rate():
     assert _sent_audio_mime_type(config) == "audio/pcm;rate=16000"
 
 
-def test_vertex_realtime_inherits_the_same_input_audio_rate():
-    """VertexAIRealtimeConfig used to carry its own copy of get_audio_mime_type, so a fix to the
-    parent had no effect on the Vertex path. It must resolve through the parent now."""
+def test_vertex_realtime_sends_the_native_16khz_input_rate() -> None:
+    """Vertex's default input rate must match the serialized audio frame."""
     from typing import Final
 
     from litellm.llms.vertex_ai.realtime.transformation import VertexAIRealtimeConfig
 
     config: Final = VertexAIRealtimeConfig(access_token="t", project="p", location="us-central1")
-    assert "get_audio_mime_type" not in VertexAIRealtimeConfig.__dict__
     assert config.get_audio_mime_type() == "audio/pcm;rate=16000"
+    assert _sent_audio_mime_type(config) == "audio/pcm;rate=16000"
 
 
 def test_client_declared_input_audio_rate_is_honored_on_the_wire():