Merge pull request #39395 from seyeong-han/litellm_meta_muse_voice_realtime

feat(realtime): add Meta Muse Voice transcription

litellm/litellm_core_utils/realtime_streaming.py

@@ -445,6 +445,12 @@ class RealTimeStreaming:
             )
             sent = False
             for msg in transformed:
+                if isinstance(msg, bytes):
+                    await self.provider_config.pace_backend_send(msg)
+                    await self.backend_ws.send(msg)
+                    self._content_sent_after_setup = True
+                    sent = True
+                    continue
                 try:
                     msg_obj = _decode_json_object(msg)
                 except (json.JSONDecodeError, TypeError):
@@ -1013,7 +1019,7 @@ class RealTimeStreaming:
                     cast(str, transcript),
                     item_id=cast(str | None, event.get("item_id")),
                 )
-                if not blocked:
+                if not blocked and not self._is_transcription_session:
                     await self._send_to_backend(json.dumps({"type": "response.create"}))
                 continue
             ## LOGGING

litellm/llms/base_llm/realtime/transformation.py

@@ -1,4 +1,5 @@
 from abc import ABC, abstractmethod
+from collections.abc import Mapping, Sequence
 from typing import TYPE_CHECKING, Any
 
 import httpx
@@ -54,9 +55,12 @@ class BaseRealtimeConfig(ABC):
         message: str,
         model: str,
         session_configuration_request: str | None = None,
-    ) -> list[str]:
+    ) -> Sequence[str | bytes]:
         pass
 
+    async def pace_backend_send(self, message: bytes) -> None:
+        return None
+
     def is_setup_message(self, msg_obj: dict) -> bool:
         return False
 
@@ -79,7 +83,7 @@ class BaseRealtimeConfig(ABC):
         model: str,
         logging_session_id: str,
         session_configuration_request: str | None = None,
-    ) -> dict | OpenAIRealtimeStreamSessionEvents | None:
+    ) -> Mapping[str, object] | OpenAIRealtimeStreamSessionEvents | None:
         """
         Optional hook for providers that defer session setup until client `session.update`.
 

litellm/llms/meta/realtime/transformation.py

@@ -0,0 +1,719 @@
+import asyncio
+import base64
+import binascii
+import json
+import math
+import time
+from collections.abc import Awaitable, Callable, Iterator, Mapping
+from dataclasses import dataclass
+from types import MappingProxyType
+from typing import Final, Literal
+from urllib.parse import urlparse, urlunparse
+
+from pydantic import JsonValue, TypeAdapter, ValidationError
+
+from litellm import verbose_logger
+from litellm._uuid import uuid
+from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
+from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig
+from litellm.secret_managers.main import get_secret_str
+from litellm.types.llms.meta import MuseAudioEncoding, MuseHandshake, MuseMode, MuseSampleRate
+from litellm.types.llms.openai import (
+    OpenAIRealtimeErrorEvent,
+    OpenAIRealtimeEvents,
+    OpenAIRealtimeInputAudioBufferSpeechEvent,
+    OpenAIRealtimeInputAudioTranscriptionCompleted,
+    OpenAIRealtimeInputAudioTranscriptionDelta,
+    OpenAIRealtimeServerVadTurnDetection,
+    OpenAIRealtimeTranscriptionSession,
+    OpenAIRealtimeTranscriptionSessionCreated,
+    OpenAIRealtimeTranscriptionSettings,
+)
+from litellm.types.realtime import (
+    RealtimeInputAudioTranscriptionDurationUsage,
+    RealtimeInputAudioTranscriptionUsage,
+    RealtimeResponseTransformInput,
+    RealtimeResponseTypedDict,
+)
+
+MUSE_MODEL: Final = "muse-voice-transcribe-1.0"
+DEFAULT_MUSE_REALTIME_URL: Final = "wss://api.meta.ai/v1/asr/realtime"
+SUPPORTED_SAMPLE_RATES: Final = frozenset((16_000, 24_000))
+SUPPORTED_LANGUAGES: Final = (
+    "Arabic",
+    "Bengali",
+    "Dutch",
+    "English",
+    "French",
+    "German",
+    "Hebrew",
+    "Hindi",
+    "Indonesian",
+    "Italian",
+    "Japanese",
+    "Kannada",
+    "Korean",
+    "Malay",
+    "Mandarin Chinese",
+    "Marathi",
+    "Polish",
+    "Portuguese",
+    "Spanish",
+    "Tagalog",
+    "Tamil",
+    "Telugu",
+    "Thai",
+    "Turkish",
+    "Vietnamese",
+)
+_LANGUAGE_NAMES: Final = MappingProxyType({language.casefold(): language for language in SUPPORTED_LANGUAGES})
+_LANGUAGE_CODES: Final = MappingProxyType(
+    {
+        "ar": "Arabic",
+        "bn": "Bengali",
+        "de": "German",
+        "en": "English",
+        "es": "Spanish",
+        "fil": "Tagalog",
+        "fr": "French",
+        "he": "Hebrew",
+        "hi": "Hindi",
+        "id": "Indonesian",
+        "it": "Italian",
+        "iw": "Hebrew",
+        "ja": "Japanese",
+        "kn": "Kannada",
+        "ko": "Korean",
+        "ms": "Malay",
+        "mr": "Marathi",
+        "nl": "Dutch",
+        "pl": "Polish",
+        "pt": "Portuguese",
+        "ta": "Tamil",
+        "te": "Telugu",
+        "th": "Thai",
+        "tl": "Tagalog",
+        "tr": "Turkish",
+        "vi": "Vietnamese",
+        "zh": "Mandarin Chinese",
+    }
+)
+_SUPPORTED_TRANSCRIPTION_KEYS: Final = frozenset(("model", "language"))
+_MAX_AUDIO_BACKLOG_SECONDS: Final = 4
+_PACKET_MS: Final = 80
+_END_STREAM: Final = '{"type":"endStream"}'
+_PROVIDER_ERROR_MESSAGE: Final = "Meta Muse realtime transcription failed"
+_JSON_ADAPTER: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue)
+_EMPTY_OBJECT: Final[Mapping[str, JsonValue]] = MappingProxyType({})
+_SERVER_VAD: Final[OpenAIRealtimeServerVadTurnDetection] = {"type": "server_vad"}
+
+
+class MuseProtocolError(ValueError):
+    pass
+
+
+@dataclass(frozen=True, slots=True)
+class MuseSessionConfig:
+    model: str
+    mode: MuseMode
+    sample_rate: MuseSampleRate
+    language_bias: tuple[str, ...]
+
+    @property
+    def audio_encoding(self) -> MuseAudioEncoding:
+        return "PCM_16KHZ" if self.sample_rate == 16_000 else "PCM_24KHZ"
+
+    @property
+    def bytes_per_second(self) -> int:
+        return self.sample_rate * 2
+
+    @property
+    def packet_bytes(self) -> int:
+        return self.bytes_per_second * _PACKET_MS // 1000
+
+    @property
+    def max_encoded_append_bytes(self) -> int:
+        return 4 * ((self.bytes_per_second * _MAX_AUDIO_BACKLOG_SECONDS + 2) // 3)
+
+    def handshake(self, access_token: str) -> MuseHandshake:
+        base: Final[MuseHandshake] = {
+            "authorization": {"accessToken": access_token},
+            "audioEncoding": self.audio_encoding,
+            "model": self.model,
+            "mode": self.mode,
+            "partialMode": "CUMULATIVE",
+            "emitAudioProgress": True,
+        }
+        if not self.language_bias:
+            return base
+        biased: Final[MuseHandshake] = {**base, "languageBias": self.language_bias}
+        return biased
+
+    def openai_session(self, session_id: str) -> OpenAIRealtimeTranscriptionSession:
+        session: Final[OpenAIRealtimeTranscriptionSession] = {
+            "id": session_id,
+            "object": "realtime.transcription_session",
+            "type": "transcription",
+            "audio": {
+                "input": {
+                    "format": {"type": "audio/pcm", "rate": self.sample_rate},
+                    "transcription": self._transcription_settings(),
+                    "turn_detection": None if self.mode == "PUSH_TO_TALK" else _SERVER_VAD,
+                }
+            },
+        }
+        return session
+
+    def _transcription_settings(self) -> OpenAIRealtimeTranscriptionSettings:
+        base: Final[OpenAIRealtimeTranscriptionSettings] = {"model": self.model}
+        if not self.language_bias:
+            return base
+        localized: Final[OpenAIRealtimeTranscriptionSettings] = {**base, "language": self.language_bias[0]}
+        return localized
+
+
+_DEFAULT_SESSION_CONFIG: Final = MuseSessionConfig(
+    model=MUSE_MODEL, mode="ENDPOINTING", sample_rate=24_000, language_bias=()
+)
+
+
+def _json_object(payload: str) -> Mapping[str, JsonValue]:
+    try:
+        value: Final = _JSON_ADAPTER.validate_json(payload)
+    except ValidationError:
+        raise MuseProtocolError("invalid JSON object") from None
+    if not isinstance(value, dict):
+        raise MuseProtocolError("message must be a JSON object")
+    return value
+
+
+def _mapping(value: JsonValue | None, name: str) -> Mapping[str, JsonValue]:
+    if value is None:
+        return _EMPTY_OBJECT
+    if not isinstance(value, dict):
+        raise MuseProtocolError(f"{name} must be an object")
+    return value
+
+
+def _string(value: JsonValue | None, name: str) -> str | None:
+    if value is None:
+        return None
+    if not isinstance(value, str):
+        raise MuseProtocolError(f"{name} must be a string")
+    return value
+
+
+def _normalize_model(model: str) -> str:
+    return model.removeprefix("meta/").strip()
+
+
+def _event_id() -> str:
+    return f"event_{uuid.uuid4().hex}"
+
+
+def normalize_language(language: str) -> str:
+    value: Final = language.strip()
+    if not value:
+        raise MuseProtocolError("language must be non-empty")
+    documented_name: Final = _LANGUAGE_NAMES.get(value.casefold())
+    if documented_name is not None:
+        return documented_name
+    primary: Final = value.replace("_", "-").split("-", 1)[0].casefold()
+    mapped_name: Final = _LANGUAGE_CODES.get(primary)
+    if mapped_name is None:
+        raise MuseProtocolError("unsupported Muse Voice language")
+    return mapped_name
+
+
+def normalize_access_token(api_key: str) -> str:
+    stripped: Final = api_key.strip()
+    if not stripped:
+        raise ValueError("Meta API key is required")
+    parts: Final = stripped.split(None, 1)
+    if parts[0].casefold() != "bearer":
+        return f"Bearer {stripped}"
+    if len(parts) != 2 or not parts[1].strip():
+        raise ValueError("Meta API key must include a token after Bearer")
+    return f"Bearer {parts[1].strip()}"
+
+
+def build_muse_realtime_url(api_base: str | None) -> str:
+    if api_base is None:
+        return DEFAULT_MUSE_REALTIME_URL
+    parsed: Final = urlparse(api_base.strip())
+    scheme: Final = "wss" if parsed.scheme == "https" else parsed.scheme
+    if (
+        scheme != "wss"
+        or not parsed.hostname
+        or parsed.username is not None
+        or parsed.password is not None
+        or parsed.fragment
+    ):
+        raise ValueError("Meta api_base must be an absolute wss:// or https:// URL without credentials or a fragment")
+    netloc: Final = f"{parsed.hostname}:{parsed.port}" if parsed.port is not None else parsed.hostname
+    return urlunparse((scheme, netloc, "/v1/asr/realtime", "", "", ""))
+
+
+def _parse_sample_rate(session: Mapping[str, JsonValue]) -> MuseSampleRate:
+    beta_format: Final = session.get("input_audio_format")
+    audio: Final = _mapping(session.get("audio"), "session.audio")
+    audio_input: Final = _mapping(audio.get("input"), "session.audio.input")
+    ga_format: Final = audio_input.get("format")
+    if beta_format is not None and ga_format is not None:
+        raise MuseProtocolError("input audio format must use either beta or GA layout")
+    if beta_format is not None:
+        if beta_format != "pcm16":
+            raise MuseProtocolError("Muse Voice requires pcm16 input audio")
+        return 24_000
+    if ga_format is None:
+        return 24_000
+    if isinstance(ga_format, str):
+        if ga_format != "pcm16":
+            raise MuseProtocolError("Muse Voice requires audio/pcm input audio")
+        return 24_000
+    format_mapping: Final = _mapping(ga_format, "session.audio.input.format")
+    if format_mapping.get("type") != "audio/pcm":
+        raise MuseProtocolError("Muse Voice requires audio/pcm input audio")
+    channels: Final = format_mapping.get("channels", 1)
+    if isinstance(channels, bool) or channels != 1:
+        raise MuseProtocolError("Muse Voice requires mono input audio")
+    rate: Final = format_mapping.get("rate", 24_000)
+    if isinstance(rate, bool) or not isinstance(rate, int) or rate not in SUPPORTED_SAMPLE_RATES:
+        raise MuseProtocolError("Muse Voice supports PCM16 at 16000 Hz or 24000 Hz")
+    return 16_000 if rate == 16_000 else 24_000
+
+
+def _parse_mode(session: Mapping[str, JsonValue], audio_input: Mapping[str, JsonValue]) -> MuseMode:
+    turn_detection_present: Final = "turn_detection" in session or "turn_detection" in audio_input
+    turn_detection: Final = session.get("turn_detection", audio_input.get("turn_detection"))
+    if turn_detection_present and turn_detection is None:
+        return "PUSH_TO_TALK"
+    if turn_detection is None:
+        return "ENDPOINTING"
+    turn_detection_mapping: Final = _mapping(turn_detection, "turn_detection")
+    if turn_detection_mapping.get("type") not in (None, "server_vad"):
+        raise MuseProtocolError("Muse Voice supports server_vad turn detection or null")
+    return "ENDPOINTING"
+
+
+def parse_session_update(payload: str, expected_model: str) -> MuseSessionConfig:
+    message: Final = _json_object(payload)
+    if message.get("type") not in ("session.update", "transcription_session.update"):
+        raise MuseProtocolError("expected session.update")
+    session: Final = _mapping(message.get("session"), "session")
+    if not session:
+        raise MuseProtocolError("session.update requires a session object")
+    if session.get("type") not in (None, "transcription", "realtime"):
+        raise MuseProtocolError("Muse Voice supports transcription sessions only")
+    audio: Final = _mapping(session.get("audio"), "session.audio")
+    audio_input: Final = _mapping(audio.get("input"), "session.audio.input")
+    beta_transcription: Final = session.get("input_audio_transcription")
+    ga_transcription: Final = audio_input.get("transcription")
+    if beta_transcription is not None and ga_transcription is not None:
+        raise MuseProtocolError("input transcription must use either beta or GA layout")
+    transcription: Final = _mapping(
+        beta_transcription if beta_transcription is not None else ga_transcription,
+        "input audio transcription",
+    )
+    unsupported: Final = tuple(sorted(key for key in transcription if key not in _SUPPORTED_TRANSCRIPTION_KEYS))
+    if unsupported:
+        verbose_logger.warning("Meta realtime: dropping unsupported transcription settings %s", unsupported)
+    requested_model: Final = _string(transcription.get("model"), "transcription model")
+    normalized_model: Final = _normalize_model(expected_model)
+    if normalized_model != MUSE_MODEL:
+        raise MuseProtocolError("unsupported Meta realtime model")
+    if requested_model is not None and _normalize_model(requested_model) != normalized_model:
+        raise MuseProtocolError("realtime session model cannot be changed")
+    language: Final = _string(transcription.get("language"), "language")
+    return MuseSessionConfig(
+        model=normalized_model,
+        mode=_parse_mode(session, audio_input),
+        sample_rate=_parse_sample_rate(session),
+        language_bias=() if language is None else (normalize_language(language),),
+    )
+
+
+def session_created_event(config: MuseSessionConfig, session_id: str) -> OpenAIRealtimeTranscriptionSessionCreated:
+    event: Final[OpenAIRealtimeTranscriptionSessionCreated] = {
+        "type": "session.created",
+        "event_id": _event_id(),
+        "session": config.openai_session(session_id),
+    }
+    return event
+
+
+def error_event(message: str) -> OpenAIRealtimeErrorEvent:
+    event: Final[OpenAIRealtimeErrorEvent] = {
+        "type": "error",
+        "error": {"type": "server_error", "message": message},
+    }
+    return event
+
+
+def _speech_event(
+    event_type: Literal["input_audio_buffer.speech_started", "input_audio_buffer.speech_stopped"], item_id: str
+) -> OpenAIRealtimeInputAudioBufferSpeechEvent:
+    event: Final[OpenAIRealtimeInputAudioBufferSpeechEvent] = {
+        "type": event_type,
+        "event_id": _event_id(),
+        "item_id": item_id,
+    }
+    return event
+
+
+def _delta_event(item_id: str, delta: str) -> OpenAIRealtimeInputAudioTranscriptionDelta:
+    event: Final[OpenAIRealtimeInputAudioTranscriptionDelta] = {
+        "type": "conversation.item.input_audio_transcription.delta",
+        "event_id": _event_id(),
+        "item_id": item_id,
+        "content_index": 0,
+        "delta": delta,
+    }
+    return event
+
+
+def _completed_event(
+    item_id: str, transcript: str, usage: RealtimeInputAudioTranscriptionUsage | None
+) -> OpenAIRealtimeInputAudioTranscriptionCompleted:
+    event: Final[OpenAIRealtimeInputAudioTranscriptionCompleted] = {
+        "type": "conversation.item.input_audio_transcription.completed",
+        "event_id": _event_id(),
+        "item_id": item_id,
+        "content_index": 0,
+        "transcript": transcript,
+    }
+    if usage is None:
+        return event
+    billed: Final[OpenAIRealtimeInputAudioTranscriptionCompleted] = {**event, "usage": usage}
+    return billed
+
+
+def _required_turn_id(message: Mapping[str, JsonValue], event: str) -> str:
+    value: Final = message.get("turnId")
+    if isinstance(value, bool) or not isinstance(value, (str, int)):
+        raise MuseProtocolError(f"{event} event has invalid turnId")
+    turn_id: Final = str(value).strip()
+    if not turn_id:
+        raise MuseProtocolError(f"{event} event has invalid turnId")
+    return turn_id
+
+
+def _new_suffix(previous: str, current: str) -> str:
+    return current[len(previous) :] if current.startswith(previous) else ""
+
+
+@dataclass(slots=True)
+class _TurnState:
+    item_id: str
+    started: bool = False
+    start_emitted: bool = False
+    latest_partial: str | None = None
+    emitted_partial: str = ""
+    final_text: str | None = None
+    completed_emitted: bool = False
+    stopped: bool = False
+    stopped_emitted: bool = False
+
+    def finish(self, transcript: str) -> None:
+        self.final_text = transcript
+        self.stopped = True
+
+    def drain(
+        self, take_usage: Callable[[], RealtimeInputAudioTranscriptionUsage | None]
+    ) -> Iterator[OpenAIRealtimeEvents]:
+        has_content: Final = self.latest_partial is not None or self.final_text is not None
+        if (self.started or has_content) and not self.start_emitted:
+            self.start_emitted = True
+            yield _speech_event("input_audio_buffer.speech_started", self.item_id)
+        if self.latest_partial is not None and self.final_text is None:
+            delta: Final = _new_suffix(self.emitted_partial, self.latest_partial)
+            if delta:
+                self.emitted_partial = self.latest_partial
+                yield _delta_event(self.item_id, delta)
+        if self.stopped and not self.stopped_emitted:
+            self.stopped_emitted = True
+            yield _speech_event("input_audio_buffer.speech_stopped", self.item_id)
+        if self.final_text is not None and self.stopped_emitted and not self.completed_emitted:
+            self.completed_emitted = True
+            yield _completed_event(self.item_id, self.final_text, take_usage())
+
+
+class MuseEventTransformer:
+    def __init__(self, *, turn_limit: int = 128) -> None:
+        self._turns: dict[str, _TurnState] = {}  # mutable-ok: bounded, insertion-ordered per-turn emit state
+        self._turn_limit: Final = turn_limit
+        self._active_turn_id: str | None = None
+        self._mode: MuseMode = "ENDPOINTING"
+        self._last_audio_processed_ms: float = 0.0
+        self._unbilled_seconds: float = 0.0
+
+    def configure(self, config: MuseSessionConfig) -> None:
+        self._mode = config.mode
+
+    def transform(self, message: Mapping[str, JsonValue]) -> tuple[OpenAIRealtimeEvents, ...]:
+        event_type: Final = message.get("type")
+        if event_type == "error":
+            return (error_event(_PROVIDER_ERROR_MESSAGE),)
+        if event_type == "audioProgress":
+            self._update_audio_progress(message)
+            return ()
+        turn: Final = self._apply_turn_event(event_type, message)
+        if turn is None:
+            return ()
+        return tuple(turn.drain(self.take_unbilled_usage))
+
+    def take_unbilled_usage(self) -> RealtimeInputAudioTranscriptionUsage | None:
+        seconds: Final = self._unbilled_seconds
+        if seconds <= 0:
+            return None
+        self._unbilled_seconds = 0.0
+        usage: Final[RealtimeInputAudioTranscriptionDurationUsage] = {"type": "duration", "seconds": seconds}
+        return usage
+
+    def _apply_turn_event(self, event_type: JsonValue | None, message: Mapping[str, JsonValue]) -> _TurnState | None:
+        match event_type:
+            case "speechStart":
+                return self._speech_start(message)
+            case "transcript":
+                return self._transcript(message)
+            case "speechEnd":
+                return self._speech_end(message)
+            case "speechComplete":
+                return self._speech_complete(message)
+            case _:
+                return None
+
+    def _turn(self, turn_id: str) -> _TurnState:
+        existing: Final = self._turns.get(turn_id)
+        if existing is not None:
+            return existing
+        created: Final = _TurnState(item_id=turn_id)
+        self._turns[turn_id] = created
+        if len(self._turns) > self._turn_limit:
+            del self._turns[next(iter(self._turns))]
+        return created
+
+    def _speech_start(self, message: Mapping[str, JsonValue]) -> _TurnState:
+        turn: Final = self._turn(_required_turn_id(message, "speechStart"))
+        if turn.stopped:
+            return turn
+        turn.started = True
+        self._active_turn_id = turn.item_id
+        return turn
+
+    def _transcript(self, message: Mapping[str, JsonValue]) -> _TurnState | None:
+        transcript: Final = message.get("transcript")
+        if not isinstance(transcript, str):
+            raise MuseProtocolError("transcript event has invalid transcript")
+        if not transcript and message.get("turnId") is None and self._active_turn_id is None:
+            return None
+        turn: Final = self._turn(self._transcript_turn_id(message))
+        if message.get("final") is True:
+            self._finish(turn, transcript)
+        elif turn.final_text is None:
+            turn.latest_partial = transcript
+        return turn
+
+    def _speech_end(self, message: Mapping[str, JsonValue]) -> _TurnState:
+        turn: Final = self._turn(_required_turn_id(message, "speechEnd"))
+        turn.stopped = True
+        return turn
+
+    def _speech_complete(self, message: Mapping[str, JsonValue]) -> _TurnState:
+        transcript: Final = message.get("transcript")
+        if not isinstance(transcript, str):
+            raise MuseProtocolError("speechComplete event has invalid transcript")
+        turn: Final = self._turn(_required_turn_id(message, "speechComplete"))
+        self._finish(turn, transcript)
+        return turn
+
+    def _finish(self, turn: _TurnState, transcript: str) -> None:
+        turn.finish(transcript)
+        self._release_active(turn)
+
+    def _release_active(self, turn: _TurnState) -> None:
+        if self._active_turn_id == turn.item_id:
+            self._active_turn_id = None
+
+    def _update_audio_progress(self, message: Mapping[str, JsonValue]) -> None:
+        processed_ms: Final = message.get("audioProcessedMs")
+        if (
+            isinstance(processed_ms, bool)
+            or not isinstance(processed_ms, (int, float))
+            or not math.isfinite(processed_ms)
+            or processed_ms < 0
+        ):
+            raise MuseProtocolError("audioProgress event has invalid audioProcessedMs")
+        if processed_ms <= self._last_audio_processed_ms:
+            return
+        self._unbilled_seconds += (float(processed_ms) - self._last_audio_processed_ms) / 1000
+        self._last_audio_processed_ms = float(processed_ms)
+
+    def _transcript_turn_id(self, message: Mapping[str, JsonValue]) -> str:
+        if message.get("turnId") is not None:
+            return _required_turn_id(message, "transcript")
+        if self._active_turn_id is not None:
+            return self._active_turn_id
+        if self._mode != "PUSH_TO_TALK":
+            raise MuseProtocolError("transcript event is missing turnId outside an active turn")
+        turn_id: Final = f"item_{uuid.uuid4().hex}"
+        self._active_turn_id = turn_id
+        return turn_id
+
+
+class MetaRealtimeConfig(BaseRealtimeConfig):
+    def __init__(
+        self,
+        *,
+        monotonic: Callable[[], float] = time.monotonic,
+        sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
+    ) -> None:
+        self._monotonic: Final = monotonic
+        self._sleep: Final = sleep
+        self._transformer: Final = MuseEventTransformer()
+        self._access_token: str | None = None
+        self._config: MuseSessionConfig | None = None
+        self._pending_audio: bytes = b""
+        self._end_stream_sent: bool = False
+        self._pacing_origin: float | None = None
+        self._sent_duration: float = 0.0
+
+    def validate_environment(
+        self,
+        headers: dict[str, str],  # mutable-ok: BaseRealtimeConfig contract
+        model: str,
+        api_key: str | None = None,
+    ) -> dict[str, str]:  # mutable-ok: BaseRealtimeConfig contract
+        token: Final = api_key or get_secret_str("META_API_KEY")
+        if token is None:
+            raise ValueError("api_key is required for Meta API calls")
+        self._access_token = normalize_access_token(token)
+        return headers
+
+    def get_complete_url(self, api_base: str | None, model: str, api_key: str | None = None) -> str:
+        if _normalize_model(model) != MUSE_MODEL:
+            raise ValueError(f"Unsupported Meta realtime model: {model}")
+        return build_muse_realtime_url(api_base)
+
+    def is_setup_message(self, msg_obj: Mapping[str, object]) -> bool:
+        return "authorization" in msg_obj
+
+    def transform_session_created_event(
+        self,
+        model: str,
+        logging_session_id: str,
+        session_configuration_request: str | None = None,
+    ) -> OpenAIRealtimeTranscriptionSessionCreated:
+        return session_created_event(_DEFAULT_SESSION_CONFIG, logging_session_id)
+
+    def transform_realtime_request(
+        self,
+        message: str,
+        model: str,
+        session_configuration_request: str | None = None,
+    ) -> tuple[str | bytes, ...]:
+        request: Final = _json_object(message)
+        event_type: Final = request.get("type")
+        if event_type in ("session.update", "transcription_session.update"):
+            return self._configure(message, model)
+        if event_type == "input_audio_buffer.append":
+            return self._append_audio(request)
+        if event_type == "input_audio_buffer.commit":
+            return self._flush_audio(end_stream=self._require_config().mode == "PUSH_TO_TALK")
+        if event_type == "input_audio_buffer.end":
+            return self._flush_audio(end_stream=True)
+        if event_type == "input_audio_buffer.clear":
+            self._pending_audio = b""
+            return ()
+        verbose_logger.debug("Meta realtime: dropping unsupported client event %s", event_type)
+        return ()
+
+    async def pace_backend_send(self, message: bytes) -> None:
+        now: Final = self._monotonic()
+        origin: Final = self._pacing_origin
+        effective_origin: Final = (
+            now - self._sent_duration if origin is None or now > origin + self._sent_duration else origin
+        )
+        delay: Final = effective_origin + self._sent_duration - now
+        if delay > 0:
+            await self._sleep(delay)
+        self._pacing_origin = effective_origin
+        self._sent_duration += len(message) / self._require_config().bytes_per_second
+
+    def unbilled_usage_on_session_close(self, model: str) -> RealtimeInputAudioTranscriptionUsage | None:
+        return self._transformer.take_unbilled_usage()
+
+    def transform_realtime_response(
+        self,
+        message: str | bytes,
+        model: str,
+        logging_obj: LiteLLMLoggingObj,
+        realtime_response_transform_input: RealtimeResponseTransformInput,
+    ) -> RealtimeResponseTypedDict:
+        payload: Final = message.decode("utf-8") if isinstance(message, bytes) else message
+        result: Final[RealtimeResponseTypedDict] = {
+            "response": list(self._backend_events(payload)),  # mutable-ok: RealtimeResponseTypedDict.response is a list
+            "current_output_item_id": realtime_response_transform_input.get("current_output_item_id"),
+            "current_response_id": realtime_response_transform_input.get("current_response_id"),
+            "current_delta_chunks": realtime_response_transform_input.get("current_delta_chunks"),
+            "current_conversation_id": realtime_response_transform_input.get("current_conversation_id"),
+            "current_item_chunks": realtime_response_transform_input.get("current_item_chunks"),
+            "current_delta_type": realtime_response_transform_input.get("current_delta_type"),
+            "session_configuration_request": realtime_response_transform_input.get("session_configuration_request"),
+        }
+        return result
+
+    def _backend_events(self, payload: str) -> tuple[OpenAIRealtimeEvents, ...]:
+        frame: Final = _json_object(payload)
+        session_id: Final = frame.get("sessionId")
+        if session_id is None:
+            return self._transformer.transform(frame)
+        if not isinstance(session_id, str) or not session_id.strip():
+            raise MuseProtocolError("provider returned an invalid handshake response")
+        return (session_created_event(self._require_config(), session_id.strip()),)
+
+    def _configure(self, message: str, model: str) -> tuple[str, ...]:
+        if self._config is not None:
+            verbose_logger.debug("Meta realtime: ignoring session.update after the Muse handshake was sent")
+            return ()
+        access_token: Final = self._access_token
+        if access_token is None:
+            raise MuseProtocolError("Meta API key was not validated before the session was configured")
+        config: Final = parse_session_update(message, model)
+        self._config = config
+        self._transformer.configure(config)
+        return (json.dumps(config.handshake(access_token), separators=(",", ":")),)
+
+    def _append_audio(self, request: Mapping[str, JsonValue]) -> tuple[bytes, ...]:
+        config: Final = self._require_config()
+        encoded: Final = request.get("audio")
+        if not isinstance(encoded, str):
+            raise MuseProtocolError("Audio must be a base64 string")
+        if len(encoded) > config.max_encoded_append_bytes:
+            raise MuseProtocolError("Audio append exceeds the four-second backlog limit")
+        try:
+            audio: Final = base64.b64decode(encoded, validate=True)
+        except (binascii.Error, ValueError):
+            raise MuseProtocolError("Audio must be valid base64") from None
+        if len(audio) % 2:
+            raise MuseProtocolError("PCM16 audio must contain complete samples")
+        buffered: Final = self._pending_audio + audio
+        packet_end: Final = len(buffered) - len(buffered) % config.packet_bytes
+        self._pending_audio = buffered[packet_end:]
+        return tuple(
+            buffered[start : start + config.packet_bytes] for start in range(0, packet_end, config.packet_bytes)
+        )
+
+    def _flush_audio(self, *, end_stream: bool) -> tuple[str | bytes, ...]:
+        remainder: Final = self._pending_audio
+        self._pending_audio = b""
+        frames: Final[tuple[bytes, ...]] = (remainder,) if remainder else ()
+        if not end_stream or self._end_stream_sent:
+            return frames
+        self._end_stream_sent = True
+        return (*frames, _END_STREAM)
+
+    def _require_config(self) -> MuseSessionConfig:
+        if self._config is None:
+            raise MuseProtocolError("session.update must configure the Muse session before audio is sent")
+        return self._config

litellm/llms/openai_like/providers.json

@@ -173,7 +173,7 @@
     "api_key_env": "META_API_KEY",
     "api_base_env": "META_API_BASE",
     "base_class": "openai_gpt",
-    "supported_endpoints": ["/v1/chat/completions", "/v1/responses", "/v1/messages"]
+    "supported_endpoints": ["/v1/chat/completions", "/v1/responses", "/v1/messages", "/v1/realtime"]
   },
   "cognition": {
     "base_url": "https://api.cognition.ai/v1",

litellm/model_prices_and_context_window_backup.json

@@ -34719,6 +34719,22 @@
         "supports_web_search": true,
         "supports_xhigh_reasoning_effort": true
     },
+    "meta/muse-voice-transcribe-1.0": {
+        "input_cost_per_second": 0.00005,
+        "litellm_provider": "meta",
+        "mode": "audio_transcription",
+        "source": "https://dev.meta.ai/docs/speech-to-text",
+        "supported_endpoints": [
+            "/v1/realtime"
+        ],
+        "supported_modalities": [
+            "audio"
+        ],
+        "supported_output_modalities": [
+            "text"
+        ],
+        "supports_audio_input": true
+    },
     "meta_llama/Llama-3.3-70B-Instruct": {
         "litellm_provider": "meta_llama",
         "max_input_tokens": 128000,
@@ -59108,9 +59124,9 @@
         "litellm_provider": "wandb",
         "mode": "chat",
         "supports_reasoning": true,
-        "input_cost_per_token": 0.00000131,
-        "output_cost_per_token": 0.00000396,
-        "cache_read_input_token_cost": 0.000000044,
+        "input_cost_per_token": 1.31e-06,
+        "output_cost_per_token": 3.96e-06,
+        "cache_read_input_token_cost": 4.4e-08,
         "supports_prompt_caching": true,
         "source": "https://wandb.ai/site/pricing/tokens/"
     },
@@ -59118,9 +59134,9 @@
         "litellm_provider": "wandb",
         "mode": "chat",
         "supports_reasoning": true,
-        "input_cost_per_token": 0.0000001,
-        "output_cost_per_token": 0.00000015,
-        "cache_read_input_token_cost": 0.00000005,
+        "input_cost_per_token": 1e-07,
+        "output_cost_per_token": 1.5e-07,
+        "cache_read_input_token_cost": 5e-08,
         "supports_prompt_caching": true,
         "source": "https://wandb.ai/site/pricing/tokens/"
     },

litellm/types/llms/meta.py

@@ -0,0 +1,21 @@
+from typing import Literal, TypeAlias
+
+from typing_extensions import NotRequired, ReadOnly, TypedDict
+
+MuseMode: TypeAlias = Literal["PUSH_TO_TALK", "ENDPOINTING"]
+MuseAudioEncoding: TypeAlias = Literal["PCM_16KHZ", "PCM_24KHZ"]
+MuseSampleRate: TypeAlias = Literal[16000, 24000]
+
+
+class MuseAuthorization(TypedDict):
+    accessToken: ReadOnly[str]
+
+
+class MuseHandshake(TypedDict):
+    authorization: ReadOnly[MuseAuthorization]
+    audioEncoding: ReadOnly[MuseAudioEncoding]
+    model: ReadOnly[str]
+    mode: ReadOnly[MuseMode]
+    partialMode: ReadOnly[Literal["CUMULATIVE"]]
+    emitAudioProgress: ReadOnly[bool]
+    languageBias: NotRequired[ReadOnly[tuple[str, ...]]]

litellm/types/llms/openai.py

@@ -2190,6 +2190,53 @@ class OpenAIRealtimeInputAudioBufferSpeechEvent(TypedDict):
     item_id: ReadOnly[str]
 
 
+class OpenAIRealtimeErrorDetail(TypedDict):
+    type: ReadOnly[str]
+    message: ReadOnly[str]
+
+
+class OpenAIRealtimeErrorEvent(TypedDict):
+    type: ReadOnly[Literal["error"]]
+    error: ReadOnly[OpenAIRealtimeErrorDetail]
+
+
+class OpenAIRealtimeTranscriptionAudioFormat(TypedDict):
+    type: ReadOnly[Literal["audio/pcm"]]
+    rate: ReadOnly[int]
+
+
+class OpenAIRealtimeTranscriptionSettings(TypedDict):
+    model: ReadOnly[str]
+    language: NotRequired[ReadOnly[str]]
+
+
+class OpenAIRealtimeServerVadTurnDetection(TypedDict):
+    type: ReadOnly[Literal["server_vad"]]
+
+
+class OpenAIRealtimeTranscriptionAudioInput(TypedDict):
+    format: ReadOnly[OpenAIRealtimeTranscriptionAudioFormat]
+    transcription: ReadOnly[OpenAIRealtimeTranscriptionSettings]
+    turn_detection: ReadOnly[OpenAIRealtimeServerVadTurnDetection | None]
+
+
+class OpenAIRealtimeTranscriptionAudio(TypedDict):
+    input: ReadOnly[OpenAIRealtimeTranscriptionAudioInput]
+
+
+class OpenAIRealtimeTranscriptionSession(TypedDict):
+    id: ReadOnly[str]
+    object: ReadOnly[Literal["realtime.transcription_session"]]
+    type: ReadOnly[Literal["transcription"]]
+    audio: ReadOnly[OpenAIRealtimeTranscriptionAudio]
+
+
+class OpenAIRealtimeTranscriptionSessionCreated(TypedDict):
+    type: ReadOnly[Literal["session.created"]]
+    event_id: ReadOnly[str]
+    session: ReadOnly[OpenAIRealtimeTranscriptionSession]
+
+
 class OpenAIRealtimeInputAudioTranscriptionDelta(TypedDict):
     type: ReadOnly[Literal["conversation.item.input_audio_transcription.delta"]]
     event_id: ReadOnly[str]
@@ -2204,6 +2251,7 @@ class OpenAIRealtimeInputAudioTranscriptionCompleted(TypedDict):
     item_id: ReadOnly[str]
     content_index: ReadOnly[int]
     transcript: ReadOnly[str]
+    usage: NotRequired[ReadOnly[Mapping[str, object]]]
 
 
 class OpenAIRealtimeUsageTokenDetails(TypedDict):
@@ -2260,6 +2308,8 @@ OpenAIRealtimeEvents = (
     | OpenAIRealtimeInputAudioBufferSpeechEvent
     | OpenAIRealtimeInputAudioTranscriptionDelta
     | OpenAIRealtimeInputAudioTranscriptionCompleted
+    | OpenAIRealtimeTranscriptionSessionCreated
+    | OpenAIRealtimeErrorEvent
 )
 
 OpenAIRealtimeStreamList = list[OpenAIRealtimeEvents]

litellm/types/realtime.py

@@ -169,9 +169,19 @@ class RealtimeInputAudioTranscriptionUsageInputTokenDetails(TypedDict):
     audio_tokens: ReadOnly[int]
 
 
-class RealtimeInputAudioTranscriptionUsage(TypedDict):
+class RealtimeInputAudioTranscriptionTokenUsage(TypedDict):
     type: ReadOnly[Literal["tokens"]]
     input_tokens: ReadOnly[int]
     output_tokens: ReadOnly[int]
     total_tokens: ReadOnly[int]
     input_token_details: ReadOnly[RealtimeInputAudioTranscriptionUsageInputTokenDetails]
+
+
+class RealtimeInputAudioTranscriptionDurationUsage(TypedDict):
+    type: ReadOnly[Literal["duration"]]
+    seconds: ReadOnly[float]
+
+
+RealtimeInputAudioTranscriptionUsage = (
+    RealtimeInputAudioTranscriptionTokenUsage | RealtimeInputAudioTranscriptionDurationUsage
+)

litellm/utils.py

@@ -9285,6 +9285,10 @@ class ProviderConfigManager:
             from litellm.llms.gemini.realtime.transformation import GeminiRealtimeConfig
 
             return GeminiRealtimeConfig()
+        if LlmProviders.META == provider:
+            from litellm.llms.meta.realtime.transformation import MetaRealtimeConfig
+
+            return MetaRealtimeConfig()
         return None
 
     @staticmethod

model_prices_and_context_window.json

@@ -34719,6 +34719,22 @@
         "supports_web_search": true,
         "supports_xhigh_reasoning_effort": true
     },
+    "meta/muse-voice-transcribe-1.0": {
+        "input_cost_per_second": 0.00005,
+        "litellm_provider": "meta",
+        "mode": "audio_transcription",
+        "source": "https://dev.meta.ai/docs/speech-to-text",
+        "supported_endpoints": [
+            "/v1/realtime"
+        ],
+        "supported_modalities": [
+            "audio"
+        ],
+        "supported_output_modalities": [
+            "text"
+        ],
+        "supports_audio_input": true
+    },
     "meta_llama/Llama-3.3-70B-Instruct": {
         "litellm_provider": "meta_llama",
         "max_input_tokens": 128000,
@@ -59108,9 +59124,9 @@
         "litellm_provider": "wandb",
         "mode": "chat",
         "supports_reasoning": true,
-        "input_cost_per_token": 0.00000131,
-        "output_cost_per_token": 0.00000396,
-        "cache_read_input_token_cost": 0.000000044,
+        "input_cost_per_token": 1.31e-06,
+        "output_cost_per_token": 3.96e-06,
+        "cache_read_input_token_cost": 4.4e-08,
         "supports_prompt_caching": true,
         "source": "https://wandb.ai/site/pricing/tokens/"
     },
@@ -59118,9 +59134,9 @@
         "litellm_provider": "wandb",
         "mode": "chat",
         "supports_reasoning": true,
-        "input_cost_per_token": 0.0000001,
-        "output_cost_per_token": 0.00000015,
-        "cache_read_input_token_cost": 0.00000005,
+        "input_cost_per_token": 1e-07,
+        "output_cost_per_token": 1.5e-07,
+        "cache_read_input_token_cost": 5e-08,
         "supports_prompt_caching": true,
         "source": "https://wandb.ai/site/pricing/tokens/"
     },

tests/test_litellm/litellm_core_utils/test_realtime_streaming.py

@@ -10,8 +10,6 @@ from websockets.exceptions import ConnectionClosed
 from websockets.frames import Close
 
 import litellm
-
-
 from litellm.integrations.custom_guardrail import CustomGuardrail
 from litellm.litellm_core_utils.realtime_streaming import (
     REALTIME_SESSION_SUCCESS_LOGGED_KEY,
@@ -20,10 +18,6 @@ from litellm.litellm_core_utils.realtime_streaming import (
 )
 from litellm.llms.xai.realtime.transformation import XAIRealtimeNormalizer
 from litellm.types.guardrails import GuardrailEventHooks
-from litellm.types.llms.openai import (
-    OpenAIRealtimeStreamResponseBaseObject,
-    OpenAIRealtimeStreamSessionEvents,
-)
 
 
 def _make_transcript_event(text: str, item_id: str = "item_x") -> bytes:
@@ -161,6 +155,7 @@ async def test_backend_to_client_send_text_receives_str_not_bytes():
     logging_obj = MagicMock()
     logging_obj.async_success_handler = AsyncMock()
     logging_obj.success_handler = MagicMock()
+    logging_obj.dispatch_success_handlers = AsyncMock()
     streaming = RealTimeStreaming(client_ws, backend_ws, logging_obj)
 
     await streaming.backend_to_client_send_messages()
@@ -812,7 +807,6 @@ async def test_transcription_captured_in_backend_to_client():
     Test that conversation.item.input_audio_transcription.completed events
     from the backend are captured as user input during the WebSocket session.
     """
-    import litellm
 
     client_ws = MagicMock()
     client_ws.send_text = AsyncMock()
@@ -838,6 +832,7 @@ async def test_transcription_captured_in_backend_to_client():
     logging_obj.model_call_details = {"messages": "default-message-value"}
     logging_obj.async_success_handler = AsyncMock()
     logging_obj.success_handler = MagicMock()
+    logging_obj.dispatch_success_handlers = AsyncMock()
     streaming = RealTimeStreaming(client_ws, backend_ws, logging_obj)
     await streaming.backend_to_client_send_messages()
 
@@ -883,6 +878,7 @@ async def test_transcription_session_captures_usage_and_skips_response_create():
     logging_obj.model_call_details = {}
     logging_obj.async_success_handler = AsyncMock()
     logging_obj.success_handler = MagicMock()
+    logging_obj.dispatch_success_handlers = AsyncMock()
 
     streaming = RealTimeStreaming(client_ws, backend_ws, logging_obj)
     await streaming.backend_to_client_send_messages()
@@ -1100,7 +1096,6 @@ def test_capture_transcription_usage_deduplicates_when_already_stored():
     When the event is already in messages (logged via store_message), it must not
     be appended a second time by _capture_transcription_usage.
     """
-    import litellm
 
     streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock())
     # Add the event type to the default logged list so _should_store_message returns True.
@@ -1409,7 +1404,6 @@ async def test_realtime_guardrail_blocks_prompt_injection(monkeypatch: pytest.Mo
     )
 
 
-
 @pytest.mark.asyncio
 async def test_realtime_guardrail_allows_clean_transcript(monkeypatch: pytest.MonkeyPatch):
     """
@@ -1466,7 +1460,6 @@ async def test_realtime_guardrail_allows_clean_transcript(monkeypatch: pytest.Mo
     assert len(response_creates) == 1, f"Clean transcript should trigger response.create, got: {sent_to_backend}"
 
 
-
 @pytest.mark.asyncio
 async def test_realtime_text_input_guardrail_blocks_and_returns_error(monkeypatch: pytest.MonkeyPatch):
     """
@@ -1560,7 +1553,6 @@ async def test_realtime_text_input_guardrail_blocks_and_returns_error(monkeypatc
     assert len(original_items) == 0, f"Blocked item should not be forwarded to backend, got: {original_items}"
 
 
-
 @pytest.mark.asyncio
 async def test_realtime_function_call_output_guardrail_blocks_and_returns_error(monkeypatch: pytest.MonkeyPatch):
     """
@@ -1649,7 +1641,6 @@ async def test_realtime_function_call_output_guardrail_blocks_and_returns_error(
     assert "test@example.com" not in sanitized_item["output"]
 
 
-
 @pytest.mark.asyncio
 async def test_realtime_function_call_output_guardrail_allows_clean_output(monkeypatch: pytest.MonkeyPatch):
     """
@@ -1714,7 +1705,6 @@ async def test_realtime_function_call_output_guardrail_allows_clean_output(monke
     assert len(forwarded) == 1, f"Clean function_call_output should be forwarded, got: {forwarded}"
 
 
-
 @pytest.mark.asyncio
 async def test_realtime_text_input_guardrail_uses_pre_call_mode(monkeypatch: pytest.MonkeyPatch):
     """
@@ -1750,7 +1740,6 @@ async def test_realtime_text_input_guardrail_uses_pre_call_mode(monkeypatch: pyt
     )
 
 
-
 @pytest.mark.asyncio
 async def test_realtime_session_created_injects_session_update_for_audio_guardrail(monkeypatch: pytest.MonkeyPatch):
     """
@@ -1807,7 +1796,6 @@ async def test_realtime_session_created_injects_session_update_for_audio_guardra
     )
 
 
-
 @pytest.mark.asyncio
 async def test_realtime_session_created_does_not_inject_session_update_for_pre_call_only(
     monkeypatch: pytest.MonkeyPatch,
@@ -1852,7 +1840,6 @@ async def test_realtime_session_created_does_not_inject_session_update_for_pre_c
     assert len(session_updates) == 0, f"pre_call-only guardrail must not inject session.update, got: {sent_to_backend}"
 
 
-
 @pytest.mark.asyncio
 async def test_pre_call_and_post_call_guardrails_do_not_disable_server_vad(monkeypatch: pytest.MonkeyPatch):
     """Model Armor-style pre_call + post_call must not gate audio VAD."""
@@ -1868,17 +1855,17 @@ async def test_pre_call_and_post_call_guardrails_do_not_disable_server_vad(monke
         litellm,
         "callbacks",
         [
-                ModelArmorStyleGuardrail(
-                    guardrail_name="model_armor_all_pre_call",
-                    event_hook=GuardrailEventHooks.pre_call,
-                    default_on=False,
-                ),
-                ModelArmorStyleGuardrail(
-                    guardrail_name="model_armor_all_post_call",
-                    event_hook=GuardrailEventHooks.post_call,
-                    default_on=False,
-                ),
-            ],
+            ModelArmorStyleGuardrail(
+                guardrail_name="model_armor_all_pre_call",
+                event_hook=GuardrailEventHooks.pre_call,
+                default_on=False,
+            ),
+            ModelArmorStyleGuardrail(
+                guardrail_name="model_armor_all_post_call",
+                event_hook=GuardrailEventHooks.post_call,
+                default_on=False,
+            ),
+        ],
     )
 
     client_ws = MagicMock()
@@ -1902,7 +1889,6 @@ async def test_pre_call_and_post_call_guardrails_do_not_disable_server_vad(monke
     assert streaming._has_audio_transcription_guardrails() is False
 
 
-
 @pytest.mark.asyncio
 async def test_end_session_after_n_fails_closes_connection(monkeypatch: pytest.MonkeyPatch):
     """
@@ -1949,7 +1935,6 @@ async def test_end_session_after_n_fails_closes_connection(monkeypatch: pytest.M
     assert streaming._violation_count == 2
 
 
-
 @pytest.mark.asyncio
 async def test_on_violation_end_session_closes_on_first_fail(monkeypatch: pytest.MonkeyPatch):
     """
@@ -1995,7 +1980,6 @@ async def test_on_violation_end_session_closes_on_first_fail(monkeypatch: pytest
     assert streaming._violation_count == 1
 
 
-
 @pytest.mark.asyncio
 async def test_provider_path_suppresses_duplicate_session_created_after_synthetic():
     client_ws = MagicMock()
@@ -2956,7 +2940,9 @@ async def test_log_messages_routes_async_logging_through_bounded_worker():
 
         mock_worker.ensure_initialized_and_enqueue.assert_called_once()
         enqueued = mock_worker.ensure_initialized_and_enqueue.call_args
-        assert (enqueued.args or tuple(enqueued.kwargs.values()))[0] is logging_obj.dispatch_success_handlers.return_value
+        assert (enqueued.args or tuple(enqueued.kwargs.values()))[
+            0
+        ] is logging_obj.dispatch_success_handlers.return_value
         logging_obj.dispatch_success_handlers.assert_called_once_with(streaming.messages, prefer_async_handlers=True)
         logging_obj.success_handler.assert_not_called()
         # the bare create_task path must no longer be used for success logging
@@ -3041,6 +3027,7 @@ async def test_session_close_flushes_unbilled_transcription_usage():
     logging_obj: Final = MagicMock()
     logging_obj.async_success_handler = AsyncMock()
     logging_obj.success_handler = MagicMock()
+    logging_obj.dispatch_success_handlers = AsyncMock()
 
     usage: Final[RealtimeInputAudioTranscriptionUsage] = {
         "type": "tokens",
@@ -3116,6 +3103,7 @@ async def test_session_close_flush_noop_without_unbilled_usage():
     logging_obj: Final = MagicMock()
     logging_obj.async_success_handler = AsyncMock()
     logging_obj.success_handler = MagicMock()
+    logging_obj.dispatch_success_handlers = AsyncMock()
 
     provider_config: Final = MagicMock()
     provider_config.unbilled_usage_on_session_close = MagicMock(return_value=None)
@@ -3136,7 +3124,6 @@ async def test_session_close_flush_noop_without_unbilled_usage():
     )
 
 
-
 _UPSTREAM_REFUSAL: Final = "Publisher model `publishers/google/models/gemini-live-2.5-flash` was not found"
 
 
@@ -3204,9 +3191,7 @@ def _backend_ws_closing_with(*frames: bytes | Exception) -> MagicMock:
 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
-    )
+    streaming: Final = RealTimeStreaming(client_ws, backend_ws, logging, model="gpt-realtime", logging_worker=worker)
     return _RelaySession(streaming=streaming, logging=logging, worker=worker)
 
 
@@ -3412,3 +3397,136 @@ async def test_refused_session_does_not_stamp_the_reservation_ownership_marker()
 
     assert session.logging.logged_failures == (upstream_close,)
     assert REALTIME_SESSION_SUCCESS_LOGGED_KEY not in session.logging.model_call_details
+
+
+@pytest.mark.asyncio
+async def test_transformed_transcription_completion_never_sends_response_create():
+    from typing import Final
+
+    completed_event: Final = {
+        "type": "conversation.item.input_audio_transcription.completed",
+        "event_id": "event_1",
+        "item_id": "turn_1",
+        "content_index": 0,
+        "transcript": "private transcript",
+        "usage": {"type": "duration", "seconds": 0.5},
+    }
+    provider_config: Final = MagicMock()
+    provider_config.requires_session_configuration.return_value = True
+    provider_config.transform_realtime_response.return_value = {
+        "response": completed_event,
+        "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.transform_realtime_request.return_value = (json.dumps({"type": "response.create"}),)
+    provider_config.is_setup_message.return_value = False
+    provider_config.is_content_message.return_value = False
+    client_ws: Final = MagicMock()
+    client_ws.send_text = AsyncMock()
+    backend_ws: Final = MagicMock()
+    backend_ws.send = AsyncMock()
+
+    streaming: Final = RealTimeStreaming(
+        client_ws,
+        backend_ws,
+        MagicMock(),
+        provider_config=provider_config,
+        model="muse-voice-transcribe-1.0",
+        force_transcription_model="muse-voice-transcribe-1.0",
+    )
+
+    await streaming._handle_provider_config_message("{}")
+
+    assert json.loads(client_ws.send_text.await_args.args[0]) == completed_event
+    backend_ws.send.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_transcription_session_still_runs_transcription_guardrail(monkeypatch: pytest.MonkeyPatch):
+    class BlockingGuardrail(CustomGuardrail):
+        async def apply_guardrail(self, inputs, request_data, input_type, logging_obj=None):
+            raise ValueError("blocked transcript")
+
+    guardrail: Final = BlockingGuardrail(
+        guardrail_name="transcription-blocker",
+        event_hook=GuardrailEventHooks.realtime_input_transcription,
+        default_on=True,
+    )
+    monkeypatch.setattr(litellm, "callbacks", [guardrail])
+
+    completed_event: Final = {
+        "type": "conversation.item.input_audio_transcription.completed",
+        "event_id": "event_1",
+        "item_id": "turn_1",
+        "content_index": 0,
+        "transcript": "blocked transcript",
+        "usage": {"type": "duration", "seconds": 0.5},
+    }
+    provider_config: Final = MagicMock()
+    provider_config.requires_session_configuration.return_value = True
+    provider_config.transform_realtime_response.return_value = {
+        "response": completed_event,
+        "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.transform_realtime_request.return_value = ()
+    provider_config.is_setup_message.return_value = False
+    provider_config.is_content_message.return_value = False
+    client_ws: Final = MagicMock()
+    client_ws.send_text = AsyncMock()
+    backend_ws: Final = MagicMock()
+    backend_ws.send = AsyncMock()
+
+    streaming: Final = RealTimeStreaming(
+        client_ws,
+        backend_ws,
+        MagicMock(),
+        provider_config=provider_config,
+        model="muse-voice-transcribe-1.0",
+        force_transcription_model="muse-voice-transcribe-1.0",
+    )
+
+    await streaming._handle_provider_config_message("{}")
+
+    sent_to_client: Final = [json.loads(call.args[0]) for call in client_ws.send_text.await_args_list]
+    assert completed_event in sent_to_client
+    error_events: Final = [event for event in sent_to_client if event.get("type") == "error"]
+    assert len(error_events) == 1
+    assert error_events[0]["error"]["type"] == "guardrail_violation"
+    backend_ws.send.assert_not_awaited()
+    assert streaming._violation_count == 1
+
+
+@pytest.mark.asyncio
+async def test_provider_bytes_are_sent_raw_after_pacing():
+    from typing import Final
+
+    backend_ws: Final = MagicMock()
+    backend_ws.send = AsyncMock()
+    provider_config: Final = MagicMock()
+    provider_config.requires_session_configuration.return_value = True
+    provider_config.transform_realtime_request.return_value = (b"\x00\x01", '{"type":"endStream"}')
+    provider_config.pace_backend_send = AsyncMock()
+    provider_config.is_setup_message.return_value = False
+    streaming: Final = RealTimeStreaming(
+        MagicMock(),
+        backend_ws,
+        MagicMock(),
+        provider_config=provider_config,
+        model="muse-voice-transcribe-1.0",
+    )
+
+    assert await streaming._send_to_backend(json.dumps({"type": "input_audio_buffer.commit"})) is True
+
+    assert [call.args[0] for call in backend_ws.send.await_args_list] == [b"\x00\x01", '{"type":"endStream"}']
+    provider_config.pace_backend_send.assert_awaited_once_with(b"\x00\x01")

tests/test_litellm/llms/meta/realtime/test_meta_realtime_transformation.py

@@ -0,0 +1,683 @@
+import base64
+import itertools
+import json
+from typing import Final
+from unittest.mock import MagicMock
+
+import pytest
+
+from litellm.llms.meta.realtime.transformation import (
+    DEFAULT_MUSE_REALTIME_URL,
+    MUSE_MODEL,
+    MetaRealtimeConfig,
+    MuseEventTransformer,
+    MuseProtocolError,
+    MuseSessionConfig,
+    build_muse_realtime_url,
+    normalize_access_token,
+    normalize_language,
+    parse_session_update,
+    session_created_event,
+)
+from litellm.types.llms.meta import MuseMode
+from litellm.types.realtime import RealtimeResponseTransformInput
+
+EMPTY_TRANSFORM_INPUT: Final[RealtimeResponseTransformInput] = {
+    "session_configuration_request": None,
+    "current_output_item_id": None,
+    "current_response_id": None,
+    "current_delta_chunks": None,
+    "current_item_chunks": None,
+    "current_conversation_id": None,
+    "current_delta_type": None,
+}
+
+
+def _event(event_type: str, **fields: object) -> str:
+    return json.dumps({"type": event_type, **fields})
+
+
+def _ga_session_update(rate: int = 24_000, turn_detection: object = "server_vad") -> str:
+    return _event(
+        "session.update",
+        session={
+            "type": "transcription",
+            "audio": {
+                "input": {
+                    "format": {"type": "audio/pcm", "rate": rate},
+                    "turn_detection": None if turn_detection is None else {"type": turn_detection},
+                    "transcription": {"model": f"meta/{MUSE_MODEL}"},
+                }
+            },
+        },
+    )
+
+
+def _configured(rate: int = 24_000, turn_detection: object = "server_vad", **kwargs: object) -> MetaRealtimeConfig:
+    config = MetaRealtimeConfig(**kwargs)
+    config.validate_environment({}, MUSE_MODEL, api_key="secret-token")
+    config.transform_realtime_request(_ga_session_update(rate, turn_detection), MUSE_MODEL)
+    return config
+
+
+def _backend_events(config: MetaRealtimeConfig, payload: str) -> list[dict[str, object]]:
+    response = config.transform_realtime_response(payload, MUSE_MODEL, MagicMock(), EMPTY_TRANSFORM_INPUT)["response"]
+    assert isinstance(response, list)
+    return response
+
+
+def test_beta_session_translates_language_and_drops_non_openai_hints():
+    config = parse_session_update(
+        _event(
+            "session.update",
+            session={
+                "type": "transcription",
+                "input_audio_format": "pcm16",
+                "turn_detection": {"type": "server_vad"},
+                "input_audio_transcription": {
+                    "model": "meta/muse-voice-transcribe-1.0",
+                    "language": "en-US",
+                    "prompt": "must not become a keyword",
+                },
+            },
+        ),
+        "meta/muse-voice-transcribe-1.0",
+    )
+
+    assert config.sample_rate == 24_000
+    assert config.packet_bytes == 3_840
+    assert config.mode == "ENDPOINTING"
+    assert config.language_bias == ("English",)
+    assert config.handshake("Bearer token") == {
+        "mode": "ENDPOINTING",
+        "authorization": {"accessToken": "Bearer token"},
+        "audioEncoding": "PCM_24KHZ",
+        "model": MUSE_MODEL,
+        "partialMode": "CUMULATIVE",
+        "emitAudioProgress": True,
+        "languageBias": ("English",),
+    }
+    assert "must not become a keyword" not in json.dumps(config.handshake("Bearer token"))
+
+
+def test_ga_session_accepts_16khz_mono_push_to_talk():
+    config = parse_session_update(
+        _event(
+            "session.update",
+            session={
+                "type": "transcription",
+                "audio": {
+                    "input": {
+                        "format": {"type": "audio/pcm", "rate": 16000, "channels": 1},
+                        "turn_detection": None,
+                        "transcription": {"model": MUSE_MODEL, "language": "zh-Hans"},
+                    }
+                },
+            },
+        ),
+        MUSE_MODEL,
+    )
+
+    assert config.sample_rate == 16_000
+    assert config.packet_bytes == 2_560
+    assert config.mode == "PUSH_TO_TALK"
+    assert config.language_bias == ("Mandarin Chinese",)
+    assert config.handshake("Bearer token")["audioEncoding"] == "PCM_16KHZ"
+    assert "languageBias" not in MuseSessionConfig(MUSE_MODEL, "ENDPOINTING", 24_000, ()).handshake("Bearer token")
+
+
+@pytest.mark.parametrize(
+    ("source", "expected"),
+    [
+        ("EN_us", "English"),
+        ("mandarin chinese", "Mandarin Chinese"),
+        ("fil-PH", "Tagalog"),
+        ("iw-IL", "Hebrew"),
+        ("pt-BR", "Portuguese"),
+    ],
+)
+def test_language_normalization_uses_official_muse_names(source: str, expected: str):
+    assert normalize_language(source) == expected
+
+
+@pytest.mark.parametrize(
+    ("session", "message"),
+    [
+        ({"input_audio_format": "g711_ulaw"}, "requires pcm16"),
+        ({"audio": {"input": {"format": {"type": "audio/pcm", "rate": 8000}}}}, "16000 Hz or 24000 Hz"),
+        (
+            {"audio": {"input": {"format": {"type": "audio/pcm", "rate": 24000, "channels": 2}}}},
+            "requires mono",
+        ),
+        (
+            {"input_audio_format": "pcm16", "audio": {"input": {"format": {"type": "audio/pcm"}}}},
+            "either beta or GA layout",
+        ),
+        ({"input_audio_transcription": {"model": "other-model"}}, "cannot be changed"),
+        ({"input_audio_transcription": {"language": "xx"}}, "unsupported Muse Voice language"),
+        ({"turn_detection": {"type": "semantic_vad"}}, "server_vad turn detection or null"),
+        ({"type": "realtime", "audio": {"input": {"turn_detection": {"type": "semantic_vad"}}}}, "server_vad"),
+    ],
+)
+def test_session_rejects_unsupported_audio_model_and_hints(session: dict[str, object], message: str):
+    with pytest.raises(MuseProtocolError, match=message):
+        parse_session_update(_event("session.update", session={"type": "transcription", **session}), MUSE_MODEL)
+
+
+def test_session_created_event_exposes_openai_transcription_shape():
+    config = parse_session_update(
+        _event(
+            "session.update",
+            session={
+                "audio": {
+                    "input": {
+                        "format": {"type": "audio/pcm", "rate": 24000},
+                        "transcription": {"model": MUSE_MODEL, "language": "ja"},
+                    }
+                },
+            },
+        ),
+        MUSE_MODEL,
+    )
+
+    created = session_created_event(config, "provider-session")
+
+    assert created["type"] == "session.created"
+    assert created["session"]["id"] == "provider-session"
+    assert created["session"]["type"] == "transcription"
+    assert created["session"]["audio"]["input"]["turn_detection"] == {"type": "server_vad"}
+    assert created["session"]["audio"]["input"]["transcription"] == {"model": MUSE_MODEL, "language": "Japanese"}
+
+
+def test_turnless_empty_silence_transcript_is_ignored():
+    transformer = MuseEventTransformer()
+
+    assert transformer.transform(json.loads(_event("transcript", transcript="", final=True))) == ()
+
+
+def test_transcript_without_speech_start_synthesizes_start_before_delta():
+    transformer = MuseEventTransformer()
+
+    events = transformer.transform(json.loads(_event("transcript", turnId="turn-1", transcript="hello", final=False)))
+
+    assert [event["type"] for event in events] == [
+        "input_audio_buffer.speech_started",
+        "conversation.item.input_audio_transcription.delta",
+    ]
+
+
+def test_cumulative_partials_emit_only_extensions_and_final_is_authoritative():
+    transformer = MuseEventTransformer()
+
+    def send(payload: str) -> tuple[dict[str, object], ...]:
+        return transformer.transform(json.loads(payload))
+
+    started = send(_event("speechStart", turnId="turn-1"))
+    first = send(_event("transcript", turnId="turn-1", transcript="hello", final=False))
+    extension = send(_event("transcript", turnId="turn-1", transcript="hello world", final=False))
+    rewrite = send(_event("transcript", turnId="turn-1", transcript="hullo world", final=False))
+    completed = send(_event("speechComplete", turnId="turn-1", transcript="hullo world"))
+
+    assert [event["type"] for event in started] == ["input_audio_buffer.speech_started"]
+    assert first[0]["delta"] == "hello"
+    assert extension[0]["delta"] == " world"
+    assert rewrite == ()
+    assert completed[0]["type"] == "input_audio_buffer.speech_stopped"
+    assert completed[1]["type"] == "conversation.item.input_audio_transcription.completed"
+    assert completed[1]["item_id"] == "turn-1"
+    assert completed[1]["transcript"] == "hullo world"
+    assert send(_event("speechEnd", turnId="turn-1")) == ()
+
+
+def test_speech_end_then_speech_complete_emits_stopped_then_completed():
+    transformer = MuseEventTransformer()
+
+    transformer.transform(json.loads(_event("speechStart", turnId="turn-1")))
+    stopped = transformer.transform(json.loads(_event("speechEnd", turnId="turn-1")))
+    completed = transformer.transform(json.loads(_event("speechComplete", turnId="turn-1", transcript="done")))
+
+    assert [event["type"] for event in stopped] == ["input_audio_buffer.speech_stopped"]
+    assert [event["type"] for event in completed] == ["conversation.item.input_audio_transcription.completed"]
+    assert completed[0]["transcript"] == "done"
+
+
+def test_turnless_partial_between_speech_end_and_speech_complete_stays_on_that_turn():
+    transformer = MuseEventTransformer()
+    transformer.transform(json.loads(_event("speechStart", turnId="turn-1")))
+    transformer.transform(json.loads(_event("transcript", transcript="what is", final=False)))
+    transformer.transform(json.loads(_event("speechEnd", turnId="turn-1")))
+
+    post_processed = transformer.transform(
+        json.loads(_event("transcript", transcript="what is the weather", final=False))
+    )
+    completed = transformer.transform(
+        json.loads(_event("speechComplete", turnId="turn-1", transcript="What is the weather?"))
+    )
+
+    assert _typed(post_processed) == [("conversation.item.input_audio_transcription.delta", "turn-1")]
+    assert post_processed[0]["delta"] == " the weather"
+    assert _typed(completed) == [("conversation.item.input_audio_transcription.completed", "turn-1")]
+    assert completed[0]["transcript"] == "What is the weather?"
+
+
+def _typed(events: tuple[dict[str, object], ...]) -> list[tuple[object, object]]:
+    return [(event["type"], event["item_id"]) for event in events]
+
+
+def test_overlapping_turns_emit_independently_and_correlate_by_item_id():
+    transformer = MuseEventTransformer()
+
+    def send(payload: str) -> list[tuple[object, object]]:
+        return _typed(transformer.transform(json.loads(payload)))
+
+    assert send(_event("speechStart", turnId="turn-a")) == [("input_audio_buffer.speech_started", "turn-a")]
+    assert send(_event("speechStart", turnId="turn-b")) == [("input_audio_buffer.speech_started", "turn-b")]
+    assert send(_event("transcript", turnId="turn-b", transcript="second", final=False)) == [
+        ("conversation.item.input_audio_transcription.delta", "turn-b")
+    ]
+    assert send(_event("speechComplete", turnId="turn-a", transcript="first")) == [
+        ("input_audio_buffer.speech_stopped", "turn-a"),
+        ("conversation.item.input_audio_transcription.completed", "turn-a"),
+    ]
+    assert send(_event("speechEnd", turnId="turn-a")) == []
+    assert send(_event("speechEnd", turnId="turn-b")) == [("input_audio_buffer.speech_stopped", "turn-b")]
+    assert send(_event("speechComplete", turnId="turn-b", transcript="second final")) == [
+        ("conversation.item.input_audio_transcription.completed", "turn-b")
+    ]
+
+
+def test_empty_vad_turn_is_closed_and_does_not_block_the_next_turn():
+    transformer = MuseEventTransformer()
+
+    def send(payload: str) -> list[tuple[object, object]]:
+        return _typed(transformer.transform(json.loads(payload)))
+
+    assert send(_event("speechStart", turnId="noise")) == [("input_audio_buffer.speech_started", "noise")]
+    assert send(_event("speechEnd", turnId="noise")) == [("input_audio_buffer.speech_stopped", "noise")]
+    assert send(_event("speechStart", turnId="speech")) == [("input_audio_buffer.speech_started", "speech")]
+    assert send(_event("transcript", turnId="speech", transcript="hello", final=False)) == [
+        ("conversation.item.input_audio_transcription.delta", "speech")
+    ]
+    assert send(_event("speechEnd", turnId="speech")) == [("input_audio_buffer.speech_stopped", "speech")]
+    assert send(_event("speechComplete", turnId="speech", transcript="hello world")) == [
+        ("conversation.item.input_audio_transcription.completed", "speech")
+    ]
+
+
+@pytest.mark.parametrize("transcript", ["", "late words"])
+def test_late_speech_complete_after_an_empty_speech_end_completes_that_item(transcript: str):
+    transformer = MuseEventTransformer()
+    transformer.transform(json.loads(_event("speechStart", turnId="turn-1")))
+    transformer.transform(json.loads(_event("speechEnd", turnId="turn-1")))
+    transformer.transform(json.loads(_event("speechStart", turnId="turn-2")))
+
+    (completed,) = transformer.transform(json.loads(_event("speechComplete", turnId="turn-1", transcript=transcript)))
+
+    assert completed["type"] == "conversation.item.input_audio_transcription.completed"
+    assert completed["item_id"] == "turn-1"
+    assert completed["transcript"] == transcript
+
+
+def test_push_to_talk_speech_complete_closes_the_turn_without_speech_end():
+    transformer = MuseEventTransformer()
+    transformer.configure(MuseSessionConfig(MUSE_MODEL, "PUSH_TO_TALK", 24_000, ()))
+
+    transformer.transform(json.loads(_event("speechStart", turnId="turn-1")))
+    transformer.transform(json.loads(_event("transcript", turnId="turn-1", transcript="hel", final=False)))
+    events = transformer.transform(json.loads(_event("speechComplete", turnId="turn-1", transcript="hello")))
+
+    assert [event["type"] for event in events] == [
+        "input_audio_buffer.speech_stopped",
+        "conversation.item.input_audio_transcription.completed",
+    ]
+    assert events[1]["transcript"] == "hello"
+
+
+_TERMINAL_SIGNALS: Final = {
+    "speechEnd": _event("speechEnd", turnId="turn-1"),
+    "speechComplete": _event("speechComplete", turnId="turn-1", transcript="final words"),
+    "final": _event("transcript", turnId="turn-1", transcript="final words", final=True),
+}
+_TERMINAL_ORDERINGS: Final = tuple(
+    ordering for size in (1, 2, 3) for ordering in itertools.permutations(_TERMINAL_SIGNALS, size)
+)
+
+
+@pytest.mark.parametrize("mode", ["ENDPOINTING", "PUSH_TO_TALK"])
+@pytest.mark.parametrize("ordering", _TERMINAL_ORDERINGS, ids="-".join)
+def test_every_terminal_signal_order_closes_the_turn_exactly_once(mode: MuseMode, ordering: tuple[str, ...]):
+    transformer = MuseEventTransformer()
+    transformer.configure(MuseSessionConfig(MUSE_MODEL, mode, 24_000, ()))
+    transformer.transform(json.loads(_event("speechStart", turnId="turn-1")))
+    transformer.transform(json.loads(_event("transcript", turnId="turn-1", transcript="fin", final=False)))
+
+    emitted = [
+        event["type"] for signal in ordering for event in transformer.transform(json.loads(_TERMINAL_SIGNALS[signal]))
+    ]
+    replayed = [
+        event["type"] for signal in ordering for event in transformer.transform(json.loads(_TERMINAL_SIGNALS[signal]))
+    ]
+
+    has_text = bool(set(ordering) & {"speechComplete", "final"})
+    assert emitted == [
+        "input_audio_buffer.speech_stopped",
+        *(["conversation.item.input_audio_transcription.completed"] if has_text else []),
+    ]
+    assert replayed == []
+
+
+def test_push_to_talk_final_transcript_completes_without_speech_end():
+    transformer = MuseEventTransformer()
+    transformer.configure(MuseSessionConfig(MUSE_MODEL, "PUSH_TO_TALK", 24_000, ()))
+
+    events = transformer.transform(json.loads(_event("transcript", transcript="hello there", final=True)))
+
+    assert [event["type"] for event in events] == [
+        "input_audio_buffer.speech_started",
+        "input_audio_buffer.speech_stopped",
+        "conversation.item.input_audio_transcription.completed",
+    ]
+    assert events[2]["transcript"] == "hello there"
+    assert str(events[0]["item_id"]).startswith("item_")
+
+
+def test_positive_audio_progress_deltas_attach_to_next_completion_and_speaker_is_ignored():
+    transformer = MuseEventTransformer()
+
+    def send(payload: str) -> tuple[dict[str, object], ...]:
+        return transformer.transform(json.loads(payload))
+
+    send(_event("audioProgress", audioProcessedMs=1000))
+    send(_event("audioProgress", audioProcessedMs=750))
+    send(_event("audioProgress", audioProcessedMs=1600))
+    assert send(_event("speaker", turnId=42, label=" Speaker 2 ")) == ()
+    completed = send(_event("speechComplete", turnId=42, transcript="hello"))
+
+    assert "speaker" not in completed[-1]
+    assert completed[-1]["usage"] == {"type": "duration", "seconds": 1.6}
+    assert transformer.take_unbilled_usage() is None
+    assert send(_event("speechEnd", turnId=42)) == ()
+
+
+def test_trailing_audio_progress_is_returned_once():
+    transformer = MuseEventTransformer()
+
+    transformer.transform(json.loads(_event("audioProgress", audioProcessedMs=250)))
+
+    assert transformer.take_unbilled_usage() == {"type": "duration", "seconds": 0.25}
+    assert transformer.take_unbilled_usage() is None
+
+
+def test_finished_turn_ignores_late_duplicates():
+    transformer = MuseEventTransformer()
+
+    released = transformer.transform(json.loads(_event("speechComplete", turnId="turn-1", transcript="done")))
+
+    assert [event["type"] for event in released] == [
+        "input_audio_buffer.speech_started",
+        "input_audio_buffer.speech_stopped",
+        "conversation.item.input_audio_transcription.completed",
+    ]
+    assert transformer.transform(json.loads(_event("speechComplete", turnId="turn-1", transcript="duplicate"))) == ()
+    assert transformer.transform(json.loads(_event("speechEnd", turnId="turn-1"))) == ()
+    assert transformer.transform(json.loads(_event("speechStart", turnId="turn-1"))) == ()
+    assert (
+        transformer.transform(json.loads(_event("transcript", turnId="turn-1", transcript="late", final=False))) == ()
+    )
+
+
+def test_late_duplicate_speech_start_does_not_capture_the_next_turnless_transcript():
+    transformer = MuseEventTransformer()
+    transformer.configure(MuseSessionConfig(MUSE_MODEL, "PUSH_TO_TALK", 24_000, ()))
+    transformer.transform(json.loads(_event("speechStart", turnId="turn-1")))
+    transformer.transform(json.loads(_event("speechComplete", turnId="turn-1", transcript="first")))
+
+    assert transformer.transform(json.loads(_event("speechStart", turnId="turn-1"))) == ()
+    events = transformer.transform(json.loads(_event("transcript", transcript="second", final=True)))
+
+    assert [event["type"] for event in events] == [
+        "input_audio_buffer.speech_started",
+        "input_audio_buffer.speech_stopped",
+        "conversation.item.input_audio_transcription.completed",
+    ]
+    assert events[2]["transcript"] == "second"
+    assert events[2]["item_id"] != "turn-1"
+
+
+def test_turn_memory_is_bounded_by_turn_limit():
+    transformer = MuseEventTransformer(turn_limit=2)
+
+    transformer.transform(json.loads(_event("speechComplete", turnId="turn-1", transcript="one")))
+    transformer.transform(json.loads(_event("speechComplete", turnId="turn-2", transcript="two")))
+    assert transformer.transform(json.loads(_event("speechEnd", turnId="turn-1"))) == ()
+    transformer.transform(json.loads(_event("speechComplete", turnId="turn-3", transcript="three")))
+
+    forgotten = transformer.transform(json.loads(_event("speechEnd", turnId="turn-1")))
+
+    assert [event["type"] for event in forgotten] == ["input_audio_buffer.speech_stopped"]
+
+
+def test_provider_error_is_sanitized_and_encodable():
+    token = "private-token"
+    provider_body = f"authorization failed for Bearer {token}"
+    transformed = MuseEventTransformer().transform(
+        json.loads(_event("error", code="AUTH", message=provider_body, request={"accessToken": token}))
+    )
+
+    encoded = json.dumps(transformed[0])
+    assert json.loads(encoded)["error"] == {
+        "type": "server_error",
+        "message": "Meta Muse realtime transcription failed",
+    }
+    assert token not in encoded
+    assert provider_body not in encoded
+
+
+@pytest.mark.parametrize(
+    ("raw", "expected"),
+    [("token", "Bearer token"), (" Bearer token ", "Bearer token"), ("bearer   token", "Bearer token")],
+)
+def test_access_token_normalization_adds_single_bearer_prefix(raw: str, expected: str):
+    assert normalize_access_token(raw) == expected
+
+
+@pytest.mark.parametrize("raw", ["", " ", "Bearer", " bearer "])
+def test_access_token_normalization_rejects_empty_tokens(raw: str):
+    with pytest.raises(ValueError, match=r"token|key is required"):
+        normalize_access_token(raw)
+
+
+@pytest.mark.parametrize(
+    ("api_base", "expected"),
+    [
+        (None, DEFAULT_MUSE_REALTIME_URL),
+        ("https://example.test/custom/path?ignored=yes", "wss://example.test/v1/asr/realtime"),
+        ("wss://example.test:8443/other", "wss://example.test:8443/v1/asr/realtime"),
+    ],
+)
+def test_realtime_url_pins_muse_path(api_base: str | None, expected: str):
+    assert build_muse_realtime_url(api_base) == expected
+    assert MetaRealtimeConfig().get_complete_url(api_base, f"meta/{MUSE_MODEL}") == expected
+
+
+@pytest.mark.parametrize(
+    "api_base",
+    [
+        "http://example.test",
+        "ws://example.test",
+        "wss://user:pass@example.test",
+        "wss://example.test/path#fragment",
+        "not-a-url",
+    ],
+)
+def test_realtime_url_rejects_insecure_or_ambiguous_bases(api_base: str):
+    with pytest.raises(ValueError, match="absolute wss:// or https://"):
+        build_muse_realtime_url(api_base)
+
+
+def test_unsupported_model_is_rejected_before_connecting():
+    with pytest.raises(ValueError, match="Unsupported Meta realtime model: meta/other-model"):
+        MetaRealtimeConfig().get_complete_url(None, "meta/other-model")
+
+
+def test_missing_api_key_is_rejected(monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.delenv("META_API_KEY", raising=False)
+
+    with pytest.raises(ValueError, match="api_key is required for Meta API calls"):
+        MetaRealtimeConfig().validate_environment({}, MUSE_MODEL)
+
+
+def test_bearer_token_travels_only_in_the_json_handshake():
+    config = MetaRealtimeConfig()
+    headers = {"x-existing": "kept"}
+
+    assert config.validate_environment(headers, MUSE_MODEL, api_key="secret-token") == {"x-existing": "kept"}
+    (handshake,) = config.transform_realtime_request(_ga_session_update(), MUSE_MODEL)
+
+    assert isinstance(handshake, str)
+    assert json.loads(handshake)["authorization"] == {"accessToken": "Bearer secret-token"}
+    assert config.is_setup_message(json.loads(handshake)) is True
+    assert config.is_setup_message({"type": "input_audio_buffer.append"}) is False
+    assert config.transform_realtime_request(_ga_session_update(), MUSE_MODEL) == ()
+
+
+def test_synthetic_session_created_uses_default_transcription_shape():
+    created = MetaRealtimeConfig().transform_session_created_event(f"meta/{MUSE_MODEL}", "trace-1")
+
+    assert created["type"] == "session.created"
+    assert created["session"]["id"] == "trace-1"
+    assert created["session"]["audio"]["input"]["format"] == {"type": "audio/pcm", "rate": 24000}
+    assert created["session"]["audio"]["input"]["transcription"] == {"model": MUSE_MODEL}
+
+
+def test_audio_before_session_update_is_rejected():
+    config = MetaRealtimeConfig()
+    config.validate_environment({}, MUSE_MODEL, api_key="secret-token")
+
+    with pytest.raises(MuseProtocolError, match=r"session\.update must configure"):
+        config.transform_realtime_request(_event("input_audio_buffer.append", audio="AAAA"), MUSE_MODEL)
+
+
+@pytest.mark.parametrize(("rate", "packet_bytes"), [(16_000, 2_560), (24_000, 3_840)])
+def test_pcm_is_packetized_into_raw_binary_frames(rate: int, packet_bytes: int):
+    config = _configured(rate=rate)
+    pcm = b"\xff\xfe\x00\x80" * (packet_bytes // 2) + b"\x01\x02\x03\x04"
+
+    frames = config.transform_realtime_request(
+        _event("input_audio_buffer.append", audio=base64.b64encode(pcm).decode()), MUSE_MODEL
+    )
+    remainder = config.transform_realtime_request(_event("input_audio_buffer.commit"), MUSE_MODEL)
+
+    assert frames == (pcm[:packet_bytes], pcm[packet_bytes : packet_bytes * 2])
+    assert remainder == (pcm[packet_bytes * 2 :],)
+
+
+@pytest.mark.parametrize(
+    ("audio", "message"),
+    [
+        ("not base64!", "valid base64"),
+        (base64.b64encode(b"\x00").decode(), "complete samples"),
+        (12, "base64 string"),
+        ("A" * (4 * ((24_000 * 2 * 4 + 2) // 3) + 4), "four-second backlog"),
+    ],
+)
+def test_invalid_audio_appends_are_rejected(audio: object, message: str):
+    config = _configured()
+
+    with pytest.raises(MuseProtocolError, match=message):
+        config.transform_realtime_request(_event("input_audio_buffer.append", audio=audio), MUSE_MODEL)
+
+
+@pytest.mark.asyncio
+async def test_backend_sends_are_paced_to_real_time():
+    sleeps: list[float] = []
+
+    async def record_sleep(delay: float) -> None:
+        sleeps.append(delay)
+
+    config = _configured(monotonic=lambda: 10.0, sleep=record_sleep)
+    packet = b"\x01\x02" * 1_920
+
+    await config.pace_backend_send(packet)
+    await config.pace_backend_send(packet)
+    await config.pace_backend_send(packet)
+
+    assert sleeps == pytest.approx([0.08, 0.16])
+
+
+def test_endpointing_commit_flushes_without_end_stream_but_end_sends_it_once():
+    config = _configured(turn_detection="server_vad")
+
+    assert config.transform_realtime_request(_event("input_audio_buffer.commit"), MUSE_MODEL) == ()
+    assert config.transform_realtime_request(_event("input_audio_buffer.end"), MUSE_MODEL) == ('{"type":"endStream"}',)
+    assert config.transform_realtime_request(_event("input_audio_buffer.end"), MUSE_MODEL) == ()
+
+
+def test_push_to_talk_commit_ends_the_stream_once():
+    config = _configured(turn_detection=None)
+    config.transform_realtime_request(
+        _event("input_audio_buffer.append", audio=base64.b64encode(b"\x01\x02").decode()), MUSE_MODEL
+    )
+
+    assert config.transform_realtime_request(_event("input_audio_buffer.commit"), MUSE_MODEL) == (
+        b"\x01\x02",
+        '{"type":"endStream"}',
+    )
+    assert config.transform_realtime_request(_event("input_audio_buffer.end"), MUSE_MODEL) == ()
+
+
+def test_clear_drops_buffered_remainder_and_unknown_events_are_ignored():
+    config = _configured()
+    config.transform_realtime_request(
+        _event("input_audio_buffer.append", audio=base64.b64encode(b"\x01\x02").decode()), MUSE_MODEL
+    )
+
+    assert config.transform_realtime_request(_event("input_audio_buffer.clear"), MUSE_MODEL) == ()
+    assert config.transform_realtime_request(_event("response.create"), MUSE_MODEL) == ()
+    assert config.transform_realtime_request(_event("input_audio_buffer.commit"), MUSE_MODEL) == ()
+
+
+def test_provider_ack_becomes_session_created_with_provider_id():
+    config = _configured(rate=16_000, turn_detection=None)
+
+    (created,) = _backend_events(config, json.dumps({"sessionId": " provider-session "}))
+
+    assert created["type"] == "session.created"
+    assert created["session"]["id"] == "provider-session"
+    assert created["session"]["audio"]["input"]["format"]["rate"] == 16000
+    assert created["session"]["audio"]["input"]["turn_detection"] is None
+
+
+def test_provider_turn_events_and_close_usage_flow_through_config():
+    config = _configured()
+
+    assert _backend_events(config, json.dumps({"type": "audioProgress", "audioProcessedMs": 1349})) == []
+    assert _backend_events(config, _event("speechStart", turnId="t1"))[0]["type"] == "input_audio_buffer.speech_started"
+    assert _backend_events(config, _event("speechEnd", turnId="t1"))[0]["type"] == "input_audio_buffer.speech_stopped"
+    completed = _backend_events(config, _event("speechComplete", turnId="t1", transcript="what is the weather"))
+
+    assert [event["type"] for event in completed] == ["conversation.item.input_audio_transcription.completed"]
+    assert completed[0]["usage"] == {"type": "duration", "seconds": 1.349}
+    assert config.unbilled_usage_on_session_close(MUSE_MODEL) is None
+
+    assert _backend_events(config, json.dumps({"type": "audioProgress", "audioProcessedMs": 2349})) == []
+    assert config.unbilled_usage_on_session_close(MUSE_MODEL) == {"type": "duration", "seconds": 1.0}
+
+
+def test_provider_error_frame_becomes_openai_error_without_leaking_token():
+    config = _configured()
+
+    (error,) = _backend_events(config, _event("error", message="bad token secret-token"))
+
+    assert error == {
+        "type": "error",
+        "error": {"type": "server_error", "message": "Meta Muse realtime transcription failed"},
+    }
+    assert "secret-token" not in json.dumps(error)
+
+
+def test_invalid_provider_ack_is_rejected():
+    config = _configured()
+
+    with pytest.raises(MuseProtocolError, match="invalid handshake response"):
+        _backend_events(config, json.dumps({"sessionId": ""}))

tests/test_litellm/realtime_api/test_main.py

@@ -4,7 +4,6 @@ from types import TracebackType
 from typing import Final
 from unittest.mock import MagicMock, patch
 
-
 import pytest
 
 import litellm
@@ -152,6 +151,33 @@ async def test_vertex_credential_resolution_bounds_a_thread_offloaded_refresh():
     assert time.monotonic() - start < 5
 
 
+@pytest.mark.asyncio
+async def test_meta_realtime_dispatches_to_base_handler_with_meta_config(monkeypatch: pytest.MonkeyPatch):
+    from litellm.llms.meta.realtime.transformation import MetaRealtimeConfig
+
+    captured: dict[str, object] = {}
+
+    def mock_get_llm_provider(model, api_base, api_key):
+        return model.removeprefix("meta/"), "meta", None, api_base
+
+    async def mock_async_realtime(**kwargs):
+        captured.update(kwargs)
+
+    monkeypatch.setattr(realtime_main, "get_llm_provider", mock_get_llm_provider)
+    monkeypatch.setattr(realtime_main.base_llm_http_handler, "async_realtime", mock_async_realtime)
+
+    await realtime_main._arealtime.__wrapped__(
+        model="meta/muse-voice-transcribe-1.0",
+        websocket=MagicMock(),
+        litellm_logging_obj=FakeLogging(),
+        query_params={"model": "meta/muse-voice-transcribe-1.0", "intent": "transcription"},
+    )
+
+    assert isinstance(captured["provider_config"], MetaRealtimeConfig)
+    assert captured["model"] == "muse-voice-transcribe-1.0"
+    assert captured["query_params"] == {"model": "muse-voice-transcribe-1.0", "intent": "transcription"}
+
+
 @pytest.mark.asyncio
 async def test_arealtime_vertex_branch_resolves_credentials_under_a_bound(monkeypatch):
     """The wiring half of the regression: the vertex branch of _arealtime must