fix(realtime): preserve absent models and reject unsupported guardrails

This commit is contained in:
Emerson Gomes 2026-10-06 10:27:56 -05:00
parent c7950e6f41
commit 5d8bdd02bb
No known key found for this signature in database
GPG key ID: D3DF28AB5D1B5E17
4 changed files with 83 additions and 6 deletions

View file

@ -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:

View file

@ -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()

View file

@ -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)])

View file

@ -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"))