From 5d8bdd02bbe6c3bbd3841513a1c8ae20d6b7e3a6 Mon Sep 17 00:00:00 2001 From: Emerson Gomes Date: Tue, 6 Oct 2026 10:27:56 -0500 Subject: [PATCH] fix(realtime): preserve absent models and reject unsupported guardrails --- .../litellm_core_utils/realtime_streaming.py | 13 +++++- litellm/proxy/common_utils/realtime_utils.py | 3 +- .../test_realtime_streaming.py | 44 +++++++++++++++++++ .../proxy/common_utils/test_realtime_cache.py | 29 ++++++++++-- 4 files changed, 83 insertions(+), 6 deletions(-) diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index 8963ebd5c75..1156ebe1023 100644 --- a/litellm/litellm_core_utils/realtime_streaming.py +++ b/litellm/litellm_core_utils/realtime_streaming.py @@ -124,7 +124,8 @@ def _translation_audio_bytes_per_second(audio_format: str | RealtimeAudioFormats case "g711_ulaw" | "g711_alaw" | AudioPCMU() | AudioPCMA(): return 8000.0 case _: - return None + pass + return None class RealtimeEventNormalizer(Protocol): @@ -1816,6 +1817,16 @@ class RealTimeStreaming: return ClientLoopExit.CLIENT_DISCONNECTED async def bidirectional_forward(self) -> None: + if self._is_translation_session and self._has_realtime_guardrails(): + await self.websocket.send_text( + realtime_error_event( + "Translation sessions cannot enforce configured realtime guardrails", error_type="guardrail_error" + ) + ) + await self.websocket.close( + code=1008, reason="Translation sessions cannot enforce configured realtime guardrails" + ) + return forward_task: Final = asyncio.create_task(self.backend_to_client_send_messages()) client_task: Final = asyncio.create_task(self.client_ack_messages()) try: diff --git a/litellm/proxy/common_utils/realtime_utils.py b/litellm/proxy/common_utils/realtime_utils.py index ff039754555..6e38c4bec7d 100644 --- a/litellm/proxy/common_utils/realtime_utils.py +++ b/litellm/proxy/common_utils/realtime_utils.py @@ -1,3 +1,4 @@ +import json from functools import lru_cache from litellm.constants import _REALTIME_BODY_CACHE_SIZE @@ -9,4 +10,4 @@ def _realtime_request_body(model: str | None) -> bytes: Generate the realtime websocket request body. Cached with LRU semantics to avoid repeated string formatting work while keeping memory usage bounded. """ - return f'{{"model": "{model or ""}"}}'.encode() + return json.dumps({"model": model}).encode() diff --git a/tests/unit/litellm_core_utils/test_realtime_streaming.py b/tests/unit/litellm_core_utils/test_realtime_streaming.py index 1820726a579..21e66965d6c 100644 --- a/tests/unit/litellm_core_utils/test_realtime_streaming.py +++ b/tests/unit/litellm_core_utils/test_realtime_streaming.py @@ -3104,6 +3104,50 @@ async def test_translation_failed_audio_send_is_not_billed() -> None: assert streaming.messages == [] +@pytest.mark.asyncio +@pytest.mark.parametrize("default_on,requested", ((False, False), (True, False), (False, True))) +@pytest.mark.parametrize( + "event_hook", + (GuardrailEventHooks.realtime_input_transcription, GuardrailEventHooks.pre_call, GuardrailEventHooks.post_call), +) +async def test_translation_rejects_applicable_guardrails_before_forwarding_audio( + monkeypatch: pytest.MonkeyPatch, default_on: bool, requested: bool, event_hook: GuardrailEventHooks +) -> None: + guardrail: Final = CustomGuardrail(guardrail_name="speech-policy", event_hook=event_hook, default_on=default_on) + monkeypatch.setattr(litellm, "callbacks", [guardrail]) + input_event: Final = json.dumps( + {"type": "session.input_audio_buffer.append", "audio": base64.b64encode(bytes(4800)).decode()} + ) + output_event: Final = {"type": "session.output_audio.delta", "delta": base64.b64encode(bytes(4800)).decode()} + client: Final = _ga_client_ws() + client.receive_text = AsyncMock(side_effect=[input_event, ConnectionClosed(None, None)]) + client.send_text = AsyncMock() + client.close = AsyncMock() + backend: Final = MagicMock() + backend.send = AsyncMock() + backend.recv = AsyncMock(side_effect=[json.dumps(output_event), ConnectionClosed(None, None)]) + streaming: Final = RealTimeStreaming( + websocket=client, + backend_ws=backend, + logging_obj=MagicMock(), + translation_session=True, + request_data={"litellm_metadata": {"guardrails": ["speech-policy"] if requested else []}}, + ) + + await streaming.bidirectional_forward() + + messages: Final = tuple(json.loads(call.args[0]) for call in client.send_text.await_args_list) + if default_on or requested: + assert messages[0]["error"]["type"] == "guardrail_error" + client.close.assert_awaited_once_with( + code=1008, reason="Translation sessions cannot enforce configured realtime guardrails" + ) + backend.send.assert_not_awaited() + else: + assert output_event in messages + backend.send.assert_awaited_once_with(input_event) + + @pytest.mark.asyncio @pytest.mark.parametrize("retain_close", (False, True)) @pytest.mark.parametrize("reported_input,expected_input", [(None, 2.0), (0.0, 0.0), (0.25, 0.25)]) diff --git a/tests/unit/proxy/common_utils/test_realtime_cache.py b/tests/unit/proxy/common_utils/test_realtime_cache.py index 8316ed1d29a..5b4ea5eb352 100644 --- a/tests/unit/proxy/common_utils/test_realtime_cache.py +++ b/tests/unit/proxy/common_utils/test_realtime_cache.py @@ -1,6 +1,8 @@ -from typing import Any, cast +from collections.abc import Mapping +from typing import Any, Final, cast import pytest +from pydantic import TypeAdapter from litellm.proxy.common_utils.realtime_utils import _realtime_request_body from litellm.proxy.proxy_server import _realtime_query_params_template @@ -40,13 +42,32 @@ def test_realtime_request_body_caches_each_model_separately(): assert gpt4o_body_first is not gpt4o_mini_body +@pytest.mark.parametrize("is_translation", (False, True)) +def test_realtime_request_body_preserves_default_model_authorization(is_translation: bool) -> None: + from litellm.proxy.agent_endpoints.auth.managed_authorization import managed_inference_request + from litellm.proxy.common_utils.http_parsing_utils import resolve_realtime_route_model + + body: Final = TypeAdapter(Mapping[str, object]).validate_json(_realtime_request_body(None)) + route: Final = "/v1/realtime/translations" if is_translation else "/v1/realtime" + intent: Final = None if is_translation else "transcription" + admitted: Final = managed_inference_request(route, body, {}, None, intent=intent) + + assert admitted["model"] == resolve_realtime_route_model(None, intent, is_translation) + assert isinstance(admitted["model"], str) and admitted["model"] + + +@pytest.mark.parametrize("model", (None, "requested", 'model"with\\quotes', "model\nwith-newline")) +def test_realtime_request_body_preserves_model_names(model: str | None) -> None: + body: Final = TypeAdapter(Mapping[str, object]).validate_json(_realtime_request_body(model)) + + assert body["model"] == model + + def test_realtime_query_params_template_caches_each_pair_separately(): params_with_intent_first = _realtime_query_params_template("gpt-4o", "intent-a") params_with_intent_second = _realtime_query_params_template("gpt-4o", "intent-a") params_without_intent = _realtime_query_params_template("gpt-4o", None) - params_transcription_without_model = _realtime_query_params_template( - None, "transcription" - ) + params_transcription_without_model = _realtime_query_params_template(None, "transcription") assert params_with_intent_first is params_with_intent_second assert params_with_intent_first == (("model", "gpt-4o"), ("intent", "intent-a"))