scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-17 04:34:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 173

fix(realtime)!: keep turn_detection.type when a transcript guardrail disables auto-response (#44939)

Browse source 
* fix(realtime): keep turn_detection.type when a transcript guardrail disables auto-response

* test(realtime): type the GA session.update helper and mark its locals Final

* test(realtime): expect typed turn_detection on the voice guardrail session updates

* test(realtime): accept either gate path for the client's first voice session.update

* fix(gemini): drop VAD tuning when the guardrail turns auto-response off

* fix(gemini): check create_response without a redundant isinstance

* fix(vertex_ai): keep the client's VAD tuning on Vertex when auto-response is off

* fix(realtime): keep the transcript-guardrail gate typeless on xAI so it does not auto-answer

* fix(realtime): replace any client turn_detection with the untyped gate on xAI

xAI answers on its own under a typed server_vad even with create_response off,
so a client's typed turn_detection must not reach xAI while the transcript
guardrail gates the turn.

* fix(realtime): keep a client's early turn_detection and hide the guardrail cancel's error

A session.update sent before session.created reached OpenAI ahead of the proxy's own gate, so the gate replaced the client's turn_detection. The proxy now re-sends the client's gated turn_detection after the gate when the two differ.

The guardrail's response.cancel now carries a litellm_guardrail_ event_id, and the error OpenAI returns for it (response_cancel_not_active when nothing is answering) no longer reaches the client.

* fix(realtime): send the guardrail gate before a client's early session.update and let Muse and Chirp fall back to server_vad under drop_params

* fix(realtime): hide only the cancel errors that echo the guardrail's own response.cancel ids

Track the event ids of the guardrail cancels this session sent and drop a backend error only when its
error.event_id is one of them, so a client's own event id with the same prefix still gets its error.
Cover the scripted response.cancel reply in the integration upstream, update the stale list-mode test
to the typed gate, and mark the xAI cancel error cell as an expected failure, since xAI replies with
its own event id.

* test(integration support): deliver the disconnect a real server sends after the client goes away

The scripted upstream now keeps reading frames after a reply hits a closed socket, so the unit test's
fake ASGI receive has to end with the websocket.disconnect message a real server delivers.

* fix(realtime): delete the blocked transcript item before the guardrail's refusal turn

OpenAI GA answered the blocked question in some refusal turns because the
blocked user item stayed in the conversation. The raw OpenAI-protocol path
now sends conversation.item.delete for it right after the cancel. Provider
config backends have no such event, so that path no longer passes an item id.

Also bound the guardrail integration file by the shared graceful-stop
timeout and make the xAI cancel-error xfail require an AssertionError.

* fix(realtime): keep the blocked item on xAI, which cannot delete an audio input item

* test(realtime): assert the xAI voice block sends no conversation.item.delete upstream

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-10-16 15:24:40 +00:00 • committed by GitHub
parent f81e657903
commit 859529ccbc
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
19 changed files with 1252 additions and 178 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

228
litellm/litellm_core_utils/realtime_streaming.py
View file

@ -1,16 +1,18 @@
import asyncio
import json
import traceback
from collections.abc import Coroutine, Mapping, Sequence
from collections.abc import Callable, Coroutine, Mapping, Sequence
from dataclasses import dataclass
from enum import Enum, auto
from typing import TYPE_CHECKING, Any, Final, NoReturn, Protocol, TypedDict, cast
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, NoReturn, Protocol, TypeAlias, TypedDict, cast
from pydantic import TypeAdapter
from typing_extensions import ReadOnly
import litellm
from litellm._logging import redact_internal_details_from_client_message, verbose_logger
from litellm._uuid import uuid
from litellm.constants import (
REALTIME_METERED_USAGE_KEY,
REALTIME_SESSION_FAILURE_LOGGED_KEY,
@ -116,10 +118,87 @@ def _decode_json_object(payload: str) -> Mapping[str, object]:
return json.loads(payload)
_GUARDRAIL_TURN_DETECTION_GATE: Final[Mapping[str, object]] = MappingProxyType(
{"type": "server_vad", "create_response": False}
)
_EMPTY_SESSION: Final[Mapping[str, object]] = MappingProxyType({})
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object])
_GUARDRAIL_EVENT_ID_PREFIX: Final = "litellm_guardrail_"
def _json_object(value: object) -> Mapping[str, object] | None:
return _JSON_OBJECT.validate_python(value) if isinstance(value, Mapping) else None
def _mapping_field(container: Mapping[str, object], key: str) -> Mapping[str, object] | None:
return _json_object(container.get(key))
def _guardrail_event_id() -> str:
return f"{_GUARDRAIL_EVENT_ID_PREFIX}{uuid.uuid4().hex}"
def _error_event_id(event: Mapping[str, object]) -> str | None:
error: Final = _mapping_field(event, "error")
caused_by: Final = None if error is None else error.get("event_id")
return caused_by if isinstance(caused_by, str) else None
_TurnDetectionGate: TypeAlias = Callable[[object], Mapping[str, object]]
def _turn_detection_with_auto_response_off(turn_detection: object) -> Mapping[str, object]:
client_turn_detection: Final = _json_object(turn_detection)
if client_turn_detection is None:
return dict(_GUARDRAIL_TURN_DETECTION_GATE)
return {**client_turn_detection, "create_response": False}
def _nested_turn_detection_part(session: Mapping[str, object], gate: _TurnDetectionGate) -> Mapping[str, object]:
audio: Final = _mapping_field(session, "audio")
if audio is None:
return {}
audio_input: Final = _mapping_field(audio, "input")
if audio_input is None or "turn_detection" not in audio_input:
return {}
gated_input: Final = {
**audio_input,
"turn_detection": gate(audio_input["turn_detection"]),
}
return {"audio": {**audio, "input": gated_input}}
def _with_create_response_disabled(
session: Mapping[str, object], gate: _TurnDetectionGate
) -> Mapping[str, object] | None:
"""``session`` with every turn_detection the client sent unable to auto-respond, or None when it sent none."""
flat_part: Final = {"turn_detection": gate(session["turn_detection"])} if "turn_detection" in session else {}
nested_part: Final = _nested_turn_detection_part(session, gate)
if not flat_part and not nested_part:
return None
return {**session, **flat_part, **nested_part}
def _gate_session_update(
session: object, gate: _TurnDetectionGate, inject_gate_when_absent: bool
) -> Mapping[str, object] | None:
"""The session to forward under a transcript guardrail, or None when the frame needs no change."""
client_session: Final = _json_object(session)
if client_session is None:
return None
gated: Final = _with_create_response_disabled(client_session, gate)
if gated is not None or not inject_gate_when_absent:
return gated
return {**client_session, "turn_detection": gate(None)}
class RealtimeEventNormalizer(Protocol):
def should_drop(self, event: object) -> bool: ...
def normalize(self, event: dict) -> dict: ...
def patch_outgoing_session(self, session: dict) -> dict: ...
def gate_turn_detection(self, turn_detection: object, /) -> Mapping[str, object]: ...
@property
def deletes_conversation_items(self) -> bool: ...
class RealtimeUsageMeter(Protocol):
@ -197,6 +276,8 @@ class RealTimeStreaming:
# Track whether we have already sent the guardrail turn-detection update
# that disables provider auto-response for transcription guardrails.
self._guardrail_turn_detection_update_sent: bool = False
self._session_gate_sent: bool = False
self._guardrail_cancel_event_ids: frozenset[str] = frozenset()
# Deferred Gemini Live setup: Pipecat may stream audio before session.update.
# Buffer client audio until the backend acknowledges setup (setupComplete).
self._backend_setup_complete: bool = provider_config is None or provider_config.requires_session_configuration()
@ -213,6 +294,12 @@ class RealTimeStreaming:
self._is_transcription_session: bool = force_transcription_model is not None
# Optional per-provider GA event normalizer (e.g. XAIRealtimeNormalizer).
self._event_normalizer = event_normalizer
self._gate_turn_detection: Final[_TurnDetectionGate] = (
_turn_detection_with_auto_response_off if event_normalizer is None else event_normalizer.gate_turn_detection
)
self._deletes_conversation_items: Final = (
event_normalizer is None or event_normalizer.deletes_conversation_items
)
self._usage_meter = usage_meter
@property
@ -734,20 +821,37 @@ class RealTimeStreaming:
except (json.JSONDecodeError, TypeError):
return
def _make_turn_detection_update(self, turn_detection: Mapping[str, object]) -> dict[str, object]:
session: Final[dict[str, object]] = (
{"turn_detection": turn_detection}
if self._backend_uses_beta_protocol
else {"type": "realtime", "audio": {"input": {"turn_detection": turn_detection}}}
)
return {"type": "session.update", "session": session}
def _make_disable_auto_response_message(self) -> str:
"""Return a session.update that disables VAD auto-response."""
turn_detection: Final[dict[str, str | bool]] = {
"type": "server_vad",
"create_response": False,
}
if self._backend_uses_beta_protocol:
session: dict[str, object] = {"turn_detection": turn_detection}
else:
session = {
"type": "realtime",
"audio": {"input": {"turn_detection": turn_detection}},
}
return json.dumps({"type": "session.update", "session": session})
return json.dumps(self._make_turn_detection_update(self._gate_turn_detection(None)))
async def _send_guardrail_cancel(self) -> None:
event_id: Final = _guardrail_event_id()
self._guardrail_cancel_event_ids = self._guardrail_cancel_event_ids | {event_id}
await self._send_to_backend(json.dumps({"type": "response.cancel", "event_id": event_id}))
async def _withdraw_blocked_turn(self, item_id: str | None) -> None:
await self._send_guardrail_cancel()
if item_id is not None and self._deletes_conversation_items:
await self._send_to_backend(json.dumps({"type": "conversation.item.delete", "item_id": item_id}))
def _is_guardrail_cancel_error(self, event: Mapping[str, object]) -> bool:
return _error_event_id(event) in self._guardrail_cancel_event_ids
async def _send_session_gate_once(self) -> None:
"""Gate the backend's default turn_detection ahead of the first client session.update, which overrides it."""
if self._session_gate_sent or not self._should_disable_vad_auto_response():
return
self._session_gate_sent = True
await self._send_to_backend(self._make_disable_auto_response_message())
async def _maybe_send_guardrail_turn_detection_update(self) -> None:
"""Disable provider auto-response once when transcription guardrails are enabled."""
@ -921,7 +1025,7 @@ class RealTimeStreaming:
if pre_block_backend_message is not None:
await self._send_to_backend(pre_block_backend_message)
if not self._is_transcription_session:
await self._send_to_backend(json.dumps({"type": "response.cancel"}))
await self._withdraw_blocked_turn(item_id)
# Send the policy violation hint (shows as small gray status text in UI).
await self.websocket.send_text(
json.dumps(
@ -1051,10 +1155,7 @@ class RealTimeStreaming:
self.store_message(event_str)
self._capture_transcription_usage(event)
await self._send_event_to_client(event, event_str)
blocked = await self.run_realtime_guardrails(
cast(str, transcript),
item_id=cast(str | None, event.get("item_id")),
)
blocked = await self.run_realtime_guardrails(cast(str, transcript))
if not blocked and self._should_disable_vad_auto_response():
await self._send_to_backend(json.dumps({"type": "response.create"}))
continue
@ -1086,7 +1187,7 @@ class RealTimeStreaming:
if event_type == "session.created" and self._should_disable_vad_auto_response():
self.store_message(event_obj)
await self.websocket.send_text(self._event_to_client_json(event_obj))
await self._send_to_backend(self._make_disable_auto_response_message())
await self._send_session_gate_once()
return True
if event_type == "conversation.item.input_audio_transcription.completed":
@ -1140,6 +1241,10 @@ class RealTimeStreaming:
if self._usage_meter is not None:
self._usage_meter.observe_backend_event(event)
if event.get("type") == "error" and self._is_guardrail_cancel_error(event):
verbose_logger.debug("Realtime: not forwarding an error caused by a guardrail event: %s", event)
continue
if await self._handle_raw_backend_message(event, raw_response):
continue
@ -1472,19 +1577,21 @@ class RealTimeStreaming:
and self.session_configuration_request is None
and not self._guardrail_turn_detection_update_sent
and self._should_disable_vad_auto_response()
):
session: Mapping[str, object] | None = msg_obj.setdefault("session", {})
if isinstance(session, dict):
existing_td = session.get("turn_detection")
if not isinstance(existing_td, dict):
existing_td = {}
existing_td["create_response"] = False
session["turn_detection"] = existing_td
message = json.dumps(msg_obj)
guardrail_turn_detection_injected = True
verbose_logger.debug(
"Injected turn_detection into first session.update for audio transcription guardrails"
and (
gated_first := _gate_session_update(
client_event.get("session", _EMPTY_SESSION),
self._gate_turn_detection,
inject_gate_when_absent=True,
)
)
is not None
):
msg_obj["session"] = gated_first
message = json.dumps(msg_obj)
guardrail_turn_detection_injected = True
verbose_logger.debug(
"Injected turn_detection into first session.update for audio transcription guardrails"
)
## GUARDRAIL: Force ``create_response`` to False in any
# client-provided ``turn_detection`` so a later
@ -1492,52 +1599,25 @@ class RealTimeStreaming:
# and bypass the transcription guardrail after the
# initial disable. Covers both the flat beta key and the
# nested GA ``audio.input.turn_detection`` shape, since
# the GA remap below also accepts either form. Skipped
# when the injection block above already ran for this
# message, to avoid redundant double-serialization.
# the GA remap below also accepts either form. A later
# update with no turn_detection is forwarded as is: OpenAI
# keeps the earlier disable for an unset field, and xAI
# falls back to its manual default, which never auto-responds.
if (
msg_type == "session.update"
and not guardrail_turn_detection_injected
and self._should_disable_vad_auto_response()
and (
gated_later := _gate_session_update(
client_event.get("session"),
self._gate_turn_detection,
inject_gate_when_absent=False,
)
)
is not None
):
session = client_event.get("session")
if isinstance(session, dict):
td_overridden = False
flat_td = session.get("turn_detection")
flat_td_present = flat_td is not None
if flat_td_present:
if not isinstance(flat_td, dict):
flat_td = {}
if flat_td.get("create_response") is not False:
flat_td["create_response"] = False
session["turn_detection"] = flat_td
td_overridden = True
nested_td_present = False
audio = session.get("audio")
if isinstance(audio, dict):
audio_input = audio.get("input")
if isinstance(audio_input, dict):
nested_td = audio_input.get("turn_detection")
if nested_td is not None:
nested_td_present = True
if not isinstance(nested_td, dict):
nested_td = {}
if nested_td.get("create_response") is not False:
nested_td["create_response"] = False
audio_input["turn_detection"] = nested_td
td_overridden = True
# Symmetric with the first-update injection block:
# if the client omitted turn_detection entirely on
# a subsequent session.update, still inject the
# ``create_response: False`` override so the
# transcription guardrail cannot be re-enabled by
# any downstream merge that drops the original
# disable.
if not flat_td_present and not nested_td_present:
session["turn_detection"] = {"create_response": False}
td_overridden = True
if td_overridden:
message = json.dumps(msg_obj)
msg_obj["session"] = gated_later
message = json.dumps(msg_obj)
# GA compatibility: remap beta-style session fields only when
# the upstream is in GA mode. Beta upstreams expect the flat
@ -1590,6 +1670,8 @@ class RealTimeStreaming:
# would permanently disable the injection if ``_send_to_backend``
# raised — neither this loop nor
# ``_maybe_send_guardrail_turn_detection_update`` would retry.
if msg_type == "session.update" and self.provider_config is None:
await self._send_session_gate_once()
sent = await self._send_to_backend(message)
if guardrail_turn_detection_injected and sent:
self._guardrail_turn_detection_update_sent = True

10
litellm/llms/base_llm/realtime/transcription_protocol.py
View file

@ -102,6 +102,16 @@ def json_integer(
return value
def uses_server_vad(
update: TranscriptionSessionUpdate, backend: str, drop_params: bool, error: ProtocolErrorType
) -> bool:
if update.turn_detection_disabled:
return False
if update.turn_detection_type in (None, "server_vad") or drop_params:
return True
raise error(f"{backend} supports server_vad turn detection or null. Set drop_params to use server_vad instead")
def new_event_id() -> str:
return f"event_{uuid.uuid4().hex}"

11
litellm/llms/gemini/realtime/transformation.py
View file

@ -161,6 +161,10 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
"""Google AI Studio Gemini 3.5+ accepts ``id`` on functionResponses; Vertex AI rejects it."""
return True
def _rejects_vad_tuning_with_detection_off(self) -> bool:
"""Google AI Studio closes the socket (1007) on padding or silence tuning next to ``disabled: true``."""
return True
@staticmethod
def _usage_detail_alias(details: Mapping[str, int | None] | None, defaults: dict[str, int]) -> dict[str, int]:
if not isinstance(details, dict):
@ -279,10 +283,15 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
OpenAI ``semantic_vad`` has no Gemini Live equivalent — return an empty
dict so callers omit ``realtimeInputConfig`` (mapping it with
``disabled: true`` breaks native-audio sessions).
``disabled: true`` breaks native-audio sessions). An explicit
``create_response: false`` maps to detection off alone where the backend
rejects tuning next to ``disabled: true``.
"""
if isinstance(value, dict) and value.get("type") == "semantic_vad" and "create_response" not in value:
return AutomaticActivityDetection()
auto_response_off: Final = "create_response" in value and value["create_response"] is False
if auto_response_off and self._rejects_vad_tuning_with_detection_off():
return AutomaticActivityDetection(disabled=True)
automatic_activity_dection: Final = AutomaticActivityDetection()
if "create_response" in value and isinstance(value["create_response"], bool):

17
litellm/llms/meta/realtime/transformation.py
View file

@ -26,6 +26,7 @@ from litellm.llms.base_llm.realtime.transcription_protocol import (
speech_event,
transcription_session,
transcription_session_created_event,
uses_server_vad,
)
from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig
from litellm.secret_managers.main import get_secret_str
@ -230,15 +231,7 @@ def _parse_sample_rate(update: TranscriptionSessionUpdate) -> MuseSampleRate:
return 16_000 if rate == 16_000 else 24_000
def _parse_mode(update: TranscriptionSessionUpdate) -> MuseMode:
if update.turn_detection_disabled:
return "PUSH_TO_TALK"
if update.turn_detection_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:
def parse_session_update(payload: str, expected_model: str, *, drop_params: bool = False) -> MuseSessionConfig:
update: Final = parse_transcription_session_update(payload, MuseProtocolError)
if update.session_type not in (None, "transcription", "realtime"):
raise MuseProtocolError("Muse Voice supports transcription sessions only")
@ -253,7 +246,7 @@ def parse_session_update(payload: str, expected_model: str) -> MuseSessionConfig
raise MuseProtocolError("realtime session model cannot be changed")
return MuseSessionConfig(
model=normalized_model,
mode=_parse_mode(update),
mode="ENDPOINTING" if uses_server_vad(update, "Muse Voice", drop_params, MuseProtocolError) else "PUSH_TO_TALK",
sample_rate=_parse_sample_rate(update),
language_bias=() if update.language is None else (normalize_language(update.language),),
)
@ -441,9 +434,11 @@ class MetaRealtimeConfig(BaseRealtimeConfig):
*,
monotonic: Callable[[], float] = time.monotonic,
sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
drop_params: bool = False,
) -> None:
self._monotonic: Final = monotonic
self._sleep: Final = sleep
self._drop_params: Final = drop_params
self._transformer: Final = MuseEventTransformer()
self._access_token: str | None = None
self._config: MuseSessionConfig | None = None
@ -553,7 +548,7 @@ class MetaRealtimeConfig(BaseRealtimeConfig):
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)
config: Final = parse_session_update(message, model, drop_params=self._drop_params)
self._config = config
self._transformer.configure(config)
return (json.dumps(config.handshake(access_token), separators=(",", ":")),)

18
litellm/llms/vertex_ai/audio_transcription/realtime_transformation.py
View file

@ -13,7 +13,6 @@ from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
from litellm.llms.base_llm.realtime.transcription_protocol import (
RealtimeTranscriptionProtocolError,
TranscriptionAudioFormat,
TranscriptionSessionUpdate,
completed_event,
decode_pcm16_append,
delta_event,
@ -23,6 +22,7 @@ from litellm.llms.base_llm.realtime.transcription_protocol import (
speech_event,
transcription_session,
transcription_session_created_event,
uses_server_vad,
)
from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig, RealtimeBackend
from litellm.llms.vertex_ai.audio_transcription.transformation import (
@ -125,7 +125,7 @@ def default_session_config(model: str) -> ChirpSessionConfig:
)
def parse_chirp_session_update(payload: str, expected_model: str) -> ChirpSessionConfig:
def parse_chirp_session_update(payload: str, expected_model: str, *, drop_params: bool = False) -> ChirpSessionConfig:
update: Final = parse_transcription_session_update(payload, ChirpProtocolError)
if update.session_type not in (None, "transcription", "realtime"):
raise ChirpProtocolError("Speech-to-Text streaming supports transcription sessions only")
@ -141,7 +141,7 @@ def parse_chirp_session_update(payload: str, expected_model: str) -> ChirpSessio
model=model,
language=None if update.language is None else normalize_transcription_language_to_bcp47(update.language),
sample_rate=_parse_sample_rate(update.audio_format),
server_vad=_parse_server_vad(update),
server_vad=uses_server_vad(update, "Speech-to-Text streaming", drop_params, ChirpProtocolError),
)
@ -161,14 +161,6 @@ def _parse_sample_rate(audio_format: TranscriptionAudioFormat | None) -> int:
return rate
def _parse_server_vad(update: TranscriptionSessionUpdate) -> bool:
if update.turn_detection_disabled:
return False
if update.turn_detection_type not in (None, "server_vad"):
raise ChirpProtocolError("Speech-to-Text streaming supports server_vad turn detection or null")
return True
def session_created_event(config: ChirpSessionConfig, session_id: str) -> OpenAIRealtimeTranscriptionSessionCreated:
return transcription_session_created_event(config.openai_session(session_id))
@ -331,11 +323,13 @@ class VertexChirpRealtimeConfig(BaseRealtimeConfig):
project: str,
location: str | None,
backend_factory: Callable[[SpeechStreamingTarget], RealtimeBackend] = _default_backend_factory,
drop_params: bool = False,
) -> None:
self._resolve_access_token: Final = resolve_access_token
self._project: Final = validate_vertex_transcription_project_id(project)
self._location: Final = validate_vertex_transcription_location(location, DEFAULT_SPEECH_TO_TEXT_LOCATION)
self._backend_factory: Final = backend_factory
self._drop_params: Final = drop_params
self._transformer: Final = ChirpEventTransformer()
self._config: ChirpSessionConfig | None = None
self._session_id: str | None = None
@ -423,7 +417,7 @@ class VertexChirpRealtimeConfig(BaseRealtimeConfig):
if self._config is not None:
verbose_logger.debug("Speech-to-Text streaming: ignoring session.update after the stream was configured")
return ()
config: Final = parse_chirp_session_update(message, model)
config: Final = parse_chirp_session_update(message, model, drop_params=self._drop_params)
self._config = config
self._transformer.configure(config, self._session_id or f"sess_{uuid.uuid4().hex}")
return (config.configure_command(),)

8
litellm/llms/vertex_ai/realtime/transformation.py
View file

@ -47,6 +47,9 @@ class VertexAIRealtimeConfig(GeminiRealtimeConfig):
def _include_function_response_id(self) -> bool:
return False
def _rejects_vad_tuning_with_detection_off(self) -> bool:
return False
# ------------------------------------------------------------------
# URL
# ------------------------------------------------------------------
@ -248,9 +251,12 @@ def vertex_realtime_config(
resolve_access_token: Callable[[], Awaitable[str]],
project: str,
location: str | None,
drop_params: bool,
) -> VertexAIRealtimeConfig | VertexChirpRealtimeConfig:
if is_vertex_speech_to_text_model(model):
return VertexChirpRealtimeConfig(resolve_access_token=resolve_access_token, project=project, location=location)
return VertexChirpRealtimeConfig(
resolve_access_token=resolve_access_token, project=project, location=location, drop_params=drop_params
)
return VertexAIRealtimeConfig(
access_token=access_token,
project=project,

13
litellm/llms/xai/realtime/transformation.py
View file

@ -16,6 +16,7 @@ construction time (see ``handler.py``) so all normalization is isolated here
and ``RealTimeStreaming`` stays provider-agnostic.
"""
from collections.abc import Mapping
from typing import Final
@ -77,6 +78,18 @@ class XAIRealtimeNormalizer:
event = self._normalize_response_usage_event(event, event_type)
return event
def gate_turn_detection(self, _turn_detection: object, /) -> Mapping[str, object]:
"""xAI auto-answers under ``server_vad`` even with ``create_response`` off, so the gate is always untyped."""
return {"create_response": False}
@property
def deletes_conversation_items(self) -> bool:
"""xAI answers ``conversation.item.delete`` with ``Item not found`` until it commits the audio item.
The guardrail blocks on an interim transcript, before that commit.
"""
return False
def patch_outgoing_session(self, session: dict) -> dict:
"""Patch a client ``session.update`` payload before forwarding to xAI.

4
litellm/realtime_api/main.py
View file

@ -500,11 +500,13 @@ async def _arealtime(
)
return
drop_params: Final = litellm.drop_params is True or normalize_drop_params(kwargs.get("drop_params")) is True
provider_config: BaseRealtimeConfig | None = None
if _custom_llm_provider in LlmProviders._member_map_.values():
provider_config = ProviderConfigManager.get_provider_realtime_config(
model=model,
provider=LlmProviders(_custom_llm_provider),
drop_params=drop_params,
)
if provider_config is not None:
await base_llm_http_handler.async_realtime(
@ -585,7 +587,6 @@ async def _arealtime(
aws_bedrock_runtime_endpoint: Final = kwargs.get("aws_bedrock_runtime_endpoint")
aws_external_id: Final = kwargs.get("aws_external_id")
aws_session_tags: Final = kwargs.get("aws_session_tags")
drop_params: Final = litellm.drop_params is True or normalize_drop_params(kwargs.get("drop_params")) is True
await bedrock_realtime.async_realtime(
model=model,
@ -671,6 +672,7 @@ async def _arealtime(
resolve_access_token=resolve_vertex_access_token,
project=resolved_project,
location=vertex_location,
drop_params=drop_params,
)
await base_llm_http_handler.async_realtime(

3
litellm/utils.py
View file

@ -9975,6 +9975,7 @@ class ProviderConfigManager:
def get_provider_realtime_config(
model: str,
provider: LlmProviders,
drop_params: bool = False,
) -> BaseRealtimeConfig | None:
if LlmProviders.GEMINI == provider:
from litellm.llms.gemini.realtime.transformation import GeminiRealtimeConfig
@ -9983,7 +9984,7 @@ class ProviderConfigManager:
if LlmProviders.META == provider:
from litellm.llms.meta.realtime.transformation import MetaRealtimeConfig
return MetaRealtimeConfig()
return MetaRealtimeConfig(drop_params=drop_params)
return None
@staticmethod

198
tests/integration/_support/upstream.py
View file

@ -9,9 +9,10 @@ import struct
import uuid
import zlib
from collections import deque
from collections.abc import AsyncIterator, Mapping
from collections.abc import AsyncIterator, Callable, Iterator, Mapping
from dataclasses import dataclass, field
from queue import SimpleQueue
from types import MappingProxyType
from typing import Final, cast
import httpx
@ -42,7 +43,7 @@ from starlette.datastructures import UploadFile as StarletteUploadFile
from starlette.requests import Request
from starlette.responses import JSONResponse, Response, StreamingResponse
from starlette.routing import Route, WebSocketRoute
from starlette.websockets import WebSocket, WebSocketDisconnect
from starlette.websockets import WebSocket, WebSocketDisconnect, WebSocketState
JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
INTERNAL_FIELDS: Final = frozenset(
@ -81,6 +82,25 @@ class Observation:
api_key: str = ""
@dataclass(frozen=True, slots=True)
class _RealtimeConnection:
websocket: WebSocket
response: RealtimeResponse
session: dict[str, JsonValue]
scenario_id: str
authorization: str
api_key: str
events: Iterator[dict[str, JsonValue]]
typed_turn_detection: deque[JsonValue]
def auto_answers(self) -> bool:
return (
self.response.dialect == "xai"
and len(self.typed_turn_detection) > 0
and _is_typed_server_vad(self.typed_turn_detection[-1])
)
class InteractionState(BaseModel):
"""What the scripted Interactions API answers for one interaction id until a DELETE drops it."""
@ -92,6 +112,12 @@ class InteractionState(BaseModel):
delay_seconds: float = 0
class _ReceivedFrame(BaseModel):
model_config = ConfigDict(frozen=True)
type: str
text: str | None = None
class _ScenarioRegistration(BaseModel):
scenario_id: str
response: StoredResponse
@ -415,23 +441,47 @@ class Provider:
) -> None:
await websocket.accept()
session: Final = _realtime_session(response, scenario_id, websocket.query_params.get("model", ""))
frames: Final = _received_texts(websocket)
held: Final = await anext(frames, None) if response.created_after_first_frame else None
for _ in range(response.created_repeats):
await websocket.send_json({"type": response.created_event, "event_id": _event_id(), "session": session})
event_index: Final = iter(response.events)
async for message in websocket.iter_json():
payload: Final = JSON_OBJECT.validate_python(message)
self.observations.put(Observation(websocket.url.path, authorization, payload, "WEBSOCKET_FRAME", api_key))
if payload.get("type") == "session.update":
await websocket.send_json(
_realtime_update_reply(payload.get("session"), session, response.session_type)
connection: Final = _RealtimeConnection(
websocket, response, session, scenario_id, authorization, api_key, iter(response.events), deque(maxlen=1)
)
if held is not None:
await self._realtime_frame(connection, held)
async for message in frames:
await self._realtime_frame(connection, message)
async def _realtime_frame(self, connection: _RealtimeConnection, message: str) -> None:
websocket: Final = connection.websocket
response: Final = connection.response
payload: Final = JSON_OBJECT.validate_json(message)
self.observations.put(
Observation(websocket.url.path, connection.authorization, payload, "WEBSOCKET_FRAME", connection.api_key)
)
frame_type: Final = payload.get("type")
if frame_type == "session.update":
reply: Final = _realtime_update_reply(
payload.get("session"), connection.session, response.session_type, response.dialect
)
if reply["type"] == "session.updated":
connection.typed_turn_detection.extend(
filter(_declares_a_type, _declared_turn_detection(payload.get("session")))
)
continue
if payload.get("type") not in _REALTIME_TRIGGERS:
continue
event: Final = next(event_index, None)
if event is None:
continue
await websocket.send_json(_rendered_realtime_event(event, scenario_id))
await _reply_while_open(websocket, reply)
return
if frame_type == "response.cancel":
await _reply_while_open(websocket, _cancel_not_active(payload.get("event_id"), response.dialect))
return
if frame_type not in _REALTIME_TRIGGERS:
return
event: Final = next(connection.events, None)
if event is None:
return
await _reply_while_open(websocket, _rendered_realtime_event(event, connection.scenario_id))
if frame_type == "input_audio_buffer.commit" and connection.auto_answers():
await _reply_while_open(websocket, _rendered_realtime_event(dict(_AUTO_ANSWER), connection.scenario_id))
async def muse_realtime(self, websocket: WebSocket) -> None:
await websocket.accept()
@ -454,8 +504,23 @@ class Provider:
await websocket.send_json(_rendered_realtime_event(pending.popleft(), scenario_id))
async def gemini_live(self, websocket: WebSocket) -> None:
authorization: Final = websocket.headers.get("authorization", "")
scenario_id: Final = websocket.headers.get("x-goog-user-project", "")
await self._gemini_live_session(
websocket,
websocket.headers.get("x-goog-user-project", ""),
websocket.headers.get("authorization", ""),
_accepts_every_setup,
)
async def gemini_ai_studio_live(self, websocket: WebSocket) -> None:
await self._gemini_live_session(websocket, websocket.query_params.get("key", ""), "", _ai_studio_setup_refusal)
async def _gemini_live_session(
self,
websocket: WebSocket,
scenario_id: str,
authorization: str,
setup_refusal: Callable[[Mapping[str, JsonValue]], str | None],
) -> None:
self.observations.put(
Observation(
websocket.url.path,
@ -477,6 +542,9 @@ class Provider:
Observation(websocket.url.path, authorization, payload, "WEBSOCKET_FRAME", scenario_id)
)
if "setup" in payload:
if (refusal := setup_refusal(payload)) is not None:
await websocket.close(code=_INVALID_ARGUMENT_CLOSE, reason=refusal[:_CLOSE_REASON_BYTES])
return
await websocket.send_json({"setupComplete": {}})
continue
if not _GEMINI_LIVE_TRIGGERS.intersection(payload):
@ -587,6 +655,7 @@ class Provider:
WebSocketRoute("/openai/realtime", self.realtime),
WebSocketRoute("/v1/asr/realtime", self.muse_realtime),
WebSocketRoute(GEMINI_LIVE_PATH, self.gemini_live),
WebSocketRoute(GEMINI_AI_STUDIO_LIVE_PATH, self.gemini_ai_studio_live),
]
)
@ -612,6 +681,19 @@ def _interaction_body(interaction_id: str, state: InteractionState) -> dict[str,
_REALTIME_TRIGGERS: Final = frozenset({"response.create", "input_audio_buffer.commit"})
_GEMINI_LIVE_TRIGGERS: Final = frozenset({"clientContent", "realtimeInput"})
GEMINI_LIVE_PATH: Final = "/ws/google.cloud.aiplatform.v1.LlmBidiService/BidiGenerateContent"
GEMINI_AI_STUDIO_LIVE_PATH: Final = "/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent"
_INVALID_ARGUMENT_CLOSE: Final = 1007
_CLOSE_REASON_BYTES: Final = 123
_AI_STUDIO_TUNING_FIELDS: Final = MappingProxyType(
{"prefixPaddingMs": "prefix_padding_ms", "silenceDurationMs": "silence_duration_ms"}
)
_AUTO_ANSWER: Final[Mapping[str, JsonValue]] = MappingProxyType(
{
"type": "response.done",
"event_id": "evt_$UNIQUE_ID",
"response": {"id": "resp_auto_$UNIQUE_ID", "object": "realtime.response", "status": "completed", "output": []},
}
)
_TRANSCRIPTION_UPDATE_REFUSED: Final = "Passing a realtime session update to a transcription session is not allowed."
_REALTIME_UPDATE_REFUSED: Final = "Passing a transcription session update to a realtime session is not allowed."
_NESTED_TURN_DETECTION_TYPE: Final = "session.audio.input.turn_detection.type"
@ -651,11 +733,11 @@ def _missing_turn_detection_type(session: Mapping[str, JsonValue]) -> str | None
def _realtime_update_reply(
session: JsonValue, created: Mapping[str, JsonValue], session_type: str | None
session: JsonValue, created: Mapping[str, JsonValue], session_type: str | None, dialect: str
) -> dict[str, JsonValue]:
if not isinstance(session, dict):
return _realtime_error("invalid_value", "Invalid value for 'session': expected an object.", "session")
missing: Final = _missing_turn_detection_type(session)
missing: Final = None if dialect == "xai" else _missing_turn_detection_type(session)
if missing is not None:
return _realtime_error("missing_required_parameter", f"Missing required parameter: '{missing}'.", missing)
declared: Final = session.get("type")
@ -666,6 +748,82 @@ def _realtime_update_reply(
return {"type": "session.updated", "event_id": _event_id(), "session": {**created, **session}}
async def _received_texts(websocket: WebSocket) -> AsyncIterator[str]:
while (frame := _ReceivedFrame.model_validate(await websocket.receive())).type == "websocket.receive":
if frame.text is not None:
yield frame.text
async def _reply_while_open(websocket: WebSocket, frame: Mapping[str, JsonValue]) -> None:
if websocket.application_state is not WebSocketState.CONNECTED:
return
try:
await websocket.send_json(frame)
except WebSocketDisconnect:
return
def _cancel_not_active(cancel_event_id: JsonValue, dialect: str) -> dict[str, JsonValue]:
message: Final = "Cancellation failed: no active response found"
if dialect == "xai":
own_id: Final = _event_id()
return {
"type": "error",
"event_id": own_id,
"error": {
"message": message,
"type": "invalid_request_error",
"code": "invalid_request_error",
"event_id": own_id,
},
}
return {
"type": "error",
"event_id": _event_id(),
"error": {
"type": "invalid_request_error",
"code": "response_cancel_not_active",
"message": message,
"param": None,
"event_id": cancel_event_id,
},
}
def _declared_turn_detection(session: JsonValue) -> tuple[JsonValue, ...]:
if not isinstance(session, dict):
return ()
audio: Final = session.get("audio")
audio_input: Final = audio.get("input") if isinstance(audio, dict) else None
if isinstance(audio_input, dict) and "turn_detection" in audio_input:
return (audio_input["turn_detection"],)
return (session["turn_detection"],) if "turn_detection" in session else ()
def _declares_a_type(turn_detection: JsonValue) -> bool:
return isinstance(turn_detection, dict) and "type" in turn_detection
def _is_typed_server_vad(turn_detection: JsonValue) -> bool:
return isinstance(turn_detection, dict) and turn_detection.get("type") == "server_vad"
def _accepts_every_setup(_payload: Mapping[str, JsonValue]) -> str | None:
return None
def _ai_studio_setup_refusal(payload: Mapping[str, JsonValue]) -> str | None:
setup: Final = payload.get("setup")
config: Final = setup.get("realtimeInputConfig") if isinstance(setup, dict) else None
detection: Final = config.get("automaticActivityDetection") if isinstance(config, dict) else None
if not isinstance(detection, dict) or detection.get("disabled") is not True:
return None
tuning: Final = tuple(name for key, name in _AI_STUDIO_TUNING_FIELDS.items() if key in detection)
if not tuning:
return None
return f"automatic_activity_detection.disabled is true, but the following fields were also set: {', '.join(tuning)}"
def _rendered_realtime_event(event: Mapping[str, JsonValue], scenario_id: str) -> dict[str, JsonValue]:
return JSON_OBJECT.validate_json(
json.dumps(event, separators=(",", ":"))

2
tests/integration/cost_calculation/cost_tracking_case.py
View file

@ -235,6 +235,8 @@ class RealtimeResponse(BaseModel):
session_type: str | None = None
created_event: Literal["session.created", "transcription_session.created"] = "session.created"
created_repeats: int = 1
created_after_first_frame: bool = False
dialect: Literal["openai", "xai"] = "openai"
StoredResponse: TypeAlias = Annotated[

473
tests/integration/providers/test_realtime_transcription_guardrail_session_update.py
View file

@ -8,10 +8,10 @@ while a client frame alone never flips a voice session into one. Every row runs
which answers the injected and rewritten updates the way the vendors do.
A blocked transcript on a transcription session reaches the client as a ``guardrail_violation`` error and
nothing else: the ``response.cancel``, the voiced block prompt and the ``response.create`` a voice session gets
never go to a backend that cannot speak, while ``on_violation: end_session`` and ``end_session_after_n_fails``
still close the session. A guardrail that raises anything but a block closes the session the way it does on a
voice session.
nothing else: the ``response.cancel``, the blocked item's deletion, the voiced block prompt and the
``response.create`` a voice session gets never go to a backend that cannot speak, while
``on_violation: end_session`` and ``end_session_after_n_fails`` still close the session. A guardrail that raises
anything but a block closes the session the way it does on a voice session.
"""
from __future__ import annotations
@ -51,6 +51,7 @@ from tests.integration._support.database import read_rows
from tests.integration._support.process import (
OwnedProxy,
UpstreamSlot,
graceful_stop_seconds,
owned_proxy_process,
owned_upstream,
stop_root_process,
@ -60,7 +61,7 @@ from tests.integration.cost_calculation.cost_tracking_case import RealtimeRespon
RecordProperty = Callable[[str, object], None]
pytestmark: Final = pytest.mark.timeout(180)
pytestmark: Final = pytest.mark.timeout(2 * graceful_stop_seconds() + 120)
WORKERS: Final = 2
BURST: Final = 20
@ -83,7 +84,7 @@ CREATED_TYPES: Final = frozenset({"session.created", "transcription_session.crea
TRANSCRIPT_COMPLETED: Final = "conversation.item.input_audio_transcription.completed"
RESPONSE_DONE: Final = "response.done"
GUARDRAIL_VIOLATION: Final = ("guardrail_violation", "content_policy_violation")
MISSING_TURN_DETECTION_TYPE: Final = ("invalid_request_error", "missing_required_parameter")
SESSION_TYPE_MISMATCH: Final = ("invalid_request_error", "invalid_parameter")
INVALID_SESSION_VALUE: Final = ("invalid_request_error", "invalid_value")
FIVE_KB: Final = "x" * 5120
UNAUTHENTICATED_STATUS: Final = 403
@ -104,10 +105,9 @@ BETA_TRANSCRIPTION_UPDATE: Final[dict[str, JsonValue]] = {
"input_audio_transcription": {"model": TRANSCRIBE_MODEL},
}
VOICE_UPDATE: Final[dict[str, JsonValue]] = {"type": "realtime", "instructions": "answer briefly"}
VOICE_UPDATE_FORWARDED: Final[dict[str, JsonValue]] = {
"type": "realtime",
"instructions": "answer briefly",
"audio": {"input": {"turn_detection": {"create_response": False}}},
VOICE_UPDATE_GATED: Final[dict[str, JsonValue]] = {
**VOICE_UPDATE,
"audio": {"input": {"turn_detection": {"type": "server_vad", "create_response": False}}},
}
RE_ENABLE_UPDATE: Final[dict[str, JsonValue]] = {
"type": "realtime",
@ -123,14 +123,32 @@ CLIENT_DECLARED_TRANSCRIPTION: Final[dict[str, JsonValue]] = {
}
CLIENT_DECLARED_TRANSCRIPTION_FORWARDED: Final[dict[str, JsonValue]] = {
"type": "transcription",
"audio": {"input": {"turn_detection": {"create_response": False}}},
"audio": {"input": {"turn_detection": {"type": "server_vad", "create_response": False}}},
}
TURN_DETECTION_TYPE_MISSING: Final = "Missing required parameter: 'session.audio.input.turn_detection.type'."
REALTIME_UPDATE_REFUSED: Final = "Passing a transcription session update to a realtime session is not allowed."
GA_INJECTED_UPDATE: Final[dict[str, JsonValue]] = {
"type": "realtime",
"audio": {"input": {"turn_detection": {"type": "server_vad", "create_response": False}}},
}
BETA_INJECTED_UPDATE: Final[dict[str, JsonValue]] = {"turn_detection": {"type": "server_vad", "create_response": False}}
XAI_VOICE_MODEL: Final = "grok-voice"
GEMINI_LIVE_MODEL: Final = "gemini-3.8-live"
VERTEX_LIVE_LOCATION: Final = "us-central1"
UNTYPED_GATE: Final[dict[str, JsonValue]] = {"create_response": False}
SEMANTIC_VAD: Final[dict[str, JsonValue]] = {"type": "semantic_vad", "eagerness": "low"}
TUNED_SERVER_VAD: Final[dict[str, JsonValue]] = {
"type": "server_vad",
"prefix_padding_ms": 300,
"silence_duration_ms": 500,
}
GEMINI_TUNED_DETECTION: Final[dict[str, JsonValue]] = {
"disabled": True,
"prefixPaddingMs": 300,
"silenceDurationMs": 500,
}
CLIENT_CANCEL_ID: Final = "client_cancel_1"
CANCEL_NOT_ACTIVE: Final = ("invalid_request_error", "response_cancel_not_active")
AUTO_ANSWER_PREFIX: Final = "resp_auto_"
COMMIT: Final[dict[str, JsonValue]] = {"type": "input_audio_buffer.commit"}
APPEND: Final[dict[str, JsonValue]] = {"type": "input_audio_buffer.append", "audio": PCM_200MS_24KHZ}
PUSH_TO_TALK: Final = "PUSH_TO_TALK"
@ -172,7 +190,13 @@ GUARDRAIL_END_CLOSE: Final = 1000
PROXY_FAILURE_CLOSE: Final = 1011
PROXY_FAILURE_REASON: Final = "proxy failed while relaying the upstream websocket"
TOOL_OUTPUT_BLOCKED: Final = json.dumps({"error": "Tool output blocked by content policy"})
VOICE_BLOCK_FRAMES: Final = ("response.cancel", "conversation.item.create", "response.create")
VOICE_BLOCK_FRAMES: Final = (
"response.cancel",
"conversation.item.delete",
"conversation.item.create",
"response.create",
)
XAI_VOICE_BLOCK_FRAMES: Final = ("response.cancel", "conversation.item.create", "response.create")
CREATED_SECONDS: Final = 60.0
STEP_SECONDS: Final = 15.0
SDK_SECONDS: Final = 45.0
@ -318,6 +342,30 @@ def _voice_scenario(*transcripts: JsonValue) -> RealtimeResponse:
return RealtimeResponse(content_type="application/x-realtime", events=tuple(_voice_turns(transcripts)))
def _xai_voice_scenario(transcript: str) -> RealtimeResponse:
return RealtimeResponse(
content_type="application/x-realtime", events=tuple(_voice_turns((transcript,))), dialect="xai"
)
def _voice_scenario_created_after_the_first_frame(transcript: str) -> RealtimeResponse:
return RealtimeResponse(
content_type="application/x-realtime", events=tuple(_voice_turns((transcript,))), created_after_first_frame=True
)
def _live_scenario() -> RealtimeResponse:
return RealtimeResponse(content_type="application/x-realtime", events=())
def _realtime_update(turn_detection: JsonValue) -> dict[str, JsonValue]:
return {"type": "realtime", "audio": {"input": {"turn_detection": turn_detection}}}
def _without_auto_response(turn_detection: dict[str, JsonValue]) -> dict[str, JsonValue]:
return {**turn_detection, "create_response": False}
def _muse_scenario(transcript: str) -> RealtimeResponse:
return RealtimeResponse(
content_type="application/x-realtime",
@ -358,8 +406,41 @@ def _with_transcription_model(update: dict[str, JsonValue], model: str) -> dict[
return {**update, "audio": {"input": {**audio_input, "transcription": {"model": model}}}}
def _muse_deployment(scenario: Scenario, scenario_id: str, api_base: str) -> str:
return scenario.model(model=f"meta/{MUSE_MODEL}", api_key=scenario_id, api_base=api_base)
def _muse_deployment(scenario: Scenario, scenario_id: str, api_base: str, *, drop_params: bool) -> str:
return scenario.model(model=f"meta/{MUSE_MODEL}", api_key=scenario_id, api_base=api_base, drop_params=drop_params)
def _gemini_studio_deployment(scenario: Scenario, scenario_id: str, upstream_url: str) -> str:
return scenario.model(
model=f"gemini/{GEMINI_LIVE_MODEL}",
api_key=scenario_id,
api_base=upstream_url.rstrip("/"),
model_info={"mode": "realtime"},
)
def _vertex_credentials(upstream_url: str) -> str:
return json.dumps(
{
"type": "external_account",
"audience": "synthetic-vertex-audience",
"subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
"token_url": f"{upstream_url}/_oauth/token",
"credential_source": {"url": f"{upstream_url}/health"},
}
)
def _vertex_live_deployment(scenario: Scenario, project: str, upstream_url: str) -> str:
upstream: Final = upstream_url.rstrip("/")
return scenario.model(
model=f"vertex_ai/{GEMINI_LIVE_MODEL}",
vertex_project=project,
vertex_credentials=_vertex_credentials(upstream),
vertex_location=VERTEX_LIVE_LOCATION,
api_base=upstream,
model_info={"mode": "realtime"},
)
def _named_deployment_id(creator: Gateway, scenario: Scenario, name: str, model: str, scenario_id: str) -> str:
@ -549,6 +630,55 @@ def _transcribe(
return asyncio.run(_session(ws_base, path, query, key, frames, until=until, headers=headers))
async def _received_after_sending(
socket: ClientConnection, frames: tuple[dict[str, JsonValue], ...], until: str
) -> AsyncIterator[dict[str, JsonValue]]:
try:
for frame in frames:
await socket.send(json.dumps(frame))
async for event in _until(socket, until, STEP_SECONDS):
yield event
except ConnectionClosed as closed:
yield {"type": "closed", "code": _close_code(closed)}
def _is_proxy_answer(event: dict[str, JsonValue]) -> bool:
if event.get("type") != RESPONSE_DONE:
return False
return not string_value(object_value(event["response"])["id"]).startswith(AUTO_ANSWER_PREFIX)
async def _until_the_proxy_answer(
socket: ClientConnection, frames: tuple[dict[str, JsonValue], ...]
) -> AsyncIterator[dict[str, JsonValue]]:
try:
yield JSON_OBJECT.validate_json(await asyncio.wait_for(socket.recv(), CREATED_SECONDS))
for frame in frames:
await socket.send(json.dumps(frame))
deadline: Final = asyncio.get_running_loop().time() + STEP_SECONDS
while (event := await _next_event(socket, deadline)) is not None:
yield event
if _is_proxy_answer(event):
return
yield {"type": "timeout"}
except ConnectionClosed as closed:
yield {"type": "closed", "code": _close_code(closed)}
async def _talked_until_the_proxy_answer(
ws_base: str, query: str, key: str, frames: tuple[dict[str, JsonValue], ...]
) -> Session:
headers: Final = {"Authorization": f"Bearer {key}"}
async with websockets.connect(f"{ws_base}/v1/realtime?{query}", additional_headers=headers) as socket:
return Session(tuple([event async for event in _until_the_proxy_answer(socket, frames)]), None)
async def _talked_at_connect(ws_base: str, query: str, key: str, frames: tuple[dict[str, JsonValue], ...]) -> Session:
headers: Final = {"Authorization": f"Bearer {key}"}
async with websockets.connect(f"{ws_base}/v1/realtime?{query}", additional_headers=headers) as socket:
return Session(tuple([event async for event in _received_after_sending(socket, frames, RESPONSE_DONE)]), None)
def _talk(
ws_base: str,
query: str,
@ -617,10 +747,43 @@ def _session_updates(observed: tuple[dict[str, JsonValue], ...]) -> tuple[JsonVa
return tuple(frame.get("session") for frame in _sent(observed) if frame.get("type") == "session.update")
def _assert_voice_updates(observed: tuple[dict[str, JsonValue], ...], *later: dict[str, JsonValue]) -> None:
updates: Final = _session_updates(observed)
assert updates[0] == GA_INJECTED_UPDATE, updates
assert updates[1] == VOICE_UPDATE_GATED, updates
assert updates[2:] == later, updates
def _query(upgrade: dict[str, JsonValue]) -> JsonValue:
return upgrade["query"]
def _declared_turn_detection(update: JsonValue) -> JsonValue:
session: Final = object_value(update)
audio_input: Final = object_value(object_value(session.get("audio", {})).get("input", {}))
return audio_input["turn_detection"] if "turn_detection" in audio_input else session.get("turn_detection")
def _upstream_turn_detections(observed: tuple[dict[str, JsonValue], ...]) -> tuple[JsonValue, ...]:
return tuple(_declared_turn_detection(update) for update in _session_updates(observed))
def _response_ids(session: Session) -> tuple[str, ...]:
done: Final = tuple(object_value(event["response"]) for event in session.events if event["type"] == RESPONSE_DONE)
return tuple(string_value(response["id"]) for response in done)
def _error_event_ids(session: Session) -> tuple[JsonValue, ...]:
errors: Final = tuple(object_value(event["error"]) for event in session.events if event["type"] == "error")
return tuple(error.get("event_id") for error in errors)
def _setup_detection(observed: tuple[dict[str, JsonValue], ...]) -> JsonValue:
setups: Final = tuple(object_value(frame["setup"]) for frame in _sent(observed) if "setup" in frame)
assert len(setups) == 1, _sent(observed)
return object_value(setups[0].get("realtimeInputConfig", {})).get("automaticActivityDetection")
def _spend_rows(key: str, count: int) -> list[dict[str, JsonValue]]:
return eventually(
lambda: read_rows(SPEND_SQL, (sha256(key.encode()).hexdigest(),)),
@ -704,6 +867,7 @@ def _write_config(directory: Path, ca_bundle: Path) -> Path:
"litellm_settings": {
"ssl_verify": str(ca_bundle),
"enable_redis_auth_cache": True,
"gemini_live_defer_setup": True,
"cache": True,
"cache_params": {"type": "redis", "host": "os.environ/REDIS_HOST", "port": "os.environ/REDIS_PORT"},
},
@ -918,19 +1082,18 @@ def test_voice_session_keeps_the_injected_update_and_the_clean_transcript_flows(
assert session.types == (
"session.created",
"session.updated",
"error",
"session.updated",
TRANSCRIPT_COMPLETED,
RESPONSE_DONE,
), session
assert session.errors == (MISSING_TURN_DETECTION_TYPE,), session
assert session.error_messages == (TURN_DETECTION_TYPE_MISSING,), session
assert session.errors == (), session
assert _sent_types(observed) == (
"session.update",
"session.update",
"input_audio_buffer.commit",
"response.create",
), _sent(observed)
assert _session_updates(observed) == (GA_INJECTED_UPDATE, VOICE_UPDATE_FORWARDED), _session_updates(observed)
_assert_voice_updates(observed)
rows: Final = _spend_rows(key, 1)
assert rows[0]["call_type"] == "_arealtime", rows
@ -945,17 +1108,15 @@ def test_voice_session_blocked_transcript_is_refused_through_the_backend(guardra
)
observed: Final = _observed(guardrail_proxy.gateway.upstream_url, handle.scenario_id)
assert session.transcripts == (BLOCKED_TRANSCRIPT,), session
assert session.errors == (MISSING_TURN_DETECTION_TYPE, GUARDRAIL_VIOLATION), session
assert session.errors == (GUARDRAIL_VIOLATION,), session
assert session.types[-1] == RESPONSE_DONE, session
assert _sent_types(observed) == (
"session.update",
"session.update",
"input_audio_buffer.commit",
"response.cancel",
"conversation.item.create",
"response.create",
*VOICE_BLOCK_FRAMES,
), _sent(observed)
assert _session_updates(observed) == (GA_INJECTED_UPDATE, VOICE_UPDATE_FORWARDED), _session_updates(observed)
_assert_voice_updates(observed)
def test_voice_session_cannot_re_enable_auto_response_with_a_later_update(guardrail_proxy: OwnedProxy) -> None:
@ -967,12 +1128,8 @@ def test_voice_session_cannot_re_enable_auto_response_with_a_later_update(guardr
session: Final = _talk(_ws_base(_owned_url(guardrail_proxy)), f"model={model}", key, frames)
observed: Final = _observed(guardrail_proxy.gateway.upstream_url, handle.scenario_id)
assert session.types[-1] == RESPONSE_DONE, session
assert session.errors == (MISSING_TURN_DETECTION_TYPE,), session
assert _session_updates(observed) == (
GA_INJECTED_UPDATE,
VOICE_UPDATE_FORWARDED,
RE_ENABLE_FORWARDED,
), _session_updates(observed)
assert session.errors == (), session
_assert_voice_updates(observed, RE_ENABLE_FORWARDED)
def test_client_declared_transcription_type_does_not_bypass_the_voice_guardrail(guardrail_proxy: OwnedProxy) -> None:
@ -984,16 +1141,14 @@ def test_client_declared_transcription_type_does_not_bypass_the_voice_guardrail(
session: Final = _talk(_ws_base(_owned_url(guardrail_proxy)), f"model={model}", key, frames, seconds=20)
observed: Final = _observed(guardrail_proxy.gateway.upstream_url, handle.scenario_id)
assert session.transcripts == (BLOCKED_TRANSCRIPT,), session
assert session.errors == (MISSING_TURN_DETECTION_TYPE, GUARDRAIL_VIOLATION), session
assert session.error_messages[0] == TURN_DETECTION_TYPE_MISSING, session
assert session.errors == (SESSION_TYPE_MISMATCH, GUARDRAIL_VIOLATION), session
assert session.error_messages[0] == REALTIME_UPDATE_REFUSED, session
assert session.types[-1] == RESPONSE_DONE, session
assert _sent_types(observed) == (
"session.update",
"session.update",
"input_audio_buffer.commit",
"response.cancel",
"conversation.item.create",
"response.create",
*VOICE_BLOCK_FRAMES,
), _sent(observed)
assert _session_updates(observed) == (
GA_INJECTED_UPDATE,
@ -1014,6 +1169,202 @@ def test_backend_session_created_typed_transcription_skips_the_injection_on_the_
assert [_query(upgrade) for upgrade in _upgrades(observed)] == [[["model", TRANSCRIBE_MODEL]]]
def test_voice_session_typed_turn_detection_keeps_its_type_and_tuning(guardrail_proxy: OwnedProxy) -> None:
with guardrail_proxy.gateway.scenario() as scenario:
handle: Final = _scripted(scenario, _voice_scenario(CLEAN_TRANSCRIPT))
key: Final = scenario.key()
model: Final = _openai_deployment(scenario, handle.scenario_id, model=VOICE_MODEL)
frames: Final = _voice_frames(_realtime_update(SEMANTIC_VAD))
session: Final = _talk(_ws_base(_owned_url(guardrail_proxy)), f"model={model}", key, frames)
observed: Final = _observed(guardrail_proxy.gateway.upstream_url, handle.scenario_id)
assert session.errors == (), session
assert session.types[-1] == RESPONSE_DONE, session
assert _session_updates(observed) == (
GA_INJECTED_UPDATE,
_realtime_update(_without_auto_response(SEMANTIC_VAD)),
), _session_updates(observed)
@dataclass(frozen=True, slots=True)
class LaterUpdateCase:
headers: Mapping[str, str]
first: dict[str, JsonValue]
later: dict[str, JsonValue]
forwarded: tuple[dict[str, JsonValue], ...]
LATER_UPDATE_CASES: Final = (
pytest.param(
LaterUpdateCase(
{},
VOICE_UPDATE,
{"type": "realtime", "instructions": "later"},
(GA_INJECTED_UPDATE, VOICE_UPDATE_GATED),
),
id="ga",
),
pytest.param(
LaterUpdateCase(
BETA_HEADERS,
{"turn_detection": {"type": "server_vad"}},
{"instructions": "later"},
(BETA_INJECTED_UPDATE, BETA_INJECTED_UPDATE),
),
id="beta",
),
)
@pytest.mark.parametrize("case", LATER_UPDATE_CASES)
def test_later_update_without_turn_detection_is_forwarded_verbatim(
guardrail_proxy: OwnedProxy, case: LaterUpdateCase
) -> None:
with guardrail_proxy.gateway.scenario() as scenario:
handle: Final = _scripted(scenario, _voice_scenario(CLEAN_TRANSCRIPT))
key: Final = scenario.key()
model: Final = _openai_deployment(scenario, handle.scenario_id, model=VOICE_MODEL)
session: Final = _drive(
_ws_base(_owned_url(guardrail_proxy)),
f"model={model}",
key,
(_step(_update_frame(case.first), _update_frame(case.later), COMMIT, until=RESPONSE_DONE),),
headers=case.headers,
)
observed: Final = _observed(guardrail_proxy.gateway.upstream_url, handle.scenario_id)
assert session.errors == (), session
assert session.types[-1] == RESPONSE_DONE, session
assert _session_updates(observed) == (*case.forwarded, case.later), _session_updates(observed)
def test_session_update_sent_before_session_created_follows_the_gate(guardrail_proxy: OwnedProxy) -> None:
with guardrail_proxy.gateway.scenario() as scenario:
handle: Final = _scripted(scenario, _voice_scenario_created_after_the_first_frame(CLEAN_TRANSCRIPT))
key: Final = scenario.key()
model: Final = _openai_deployment(scenario, handle.scenario_id, model=VOICE_MODEL)
frames: Final = _voice_frames(_realtime_update(TUNED_SERVER_VAD))
session: Final = asyncio.run(
_talked_at_connect(_ws_base(_owned_url(guardrail_proxy)), f"model={model}", key, frames)
)
observed: Final = _observed(guardrail_proxy.gateway.upstream_url, handle.scenario_id)
assert session.types == (
"session.created",
"session.updated",
"session.updated",
TRANSCRIPT_COMPLETED,
RESPONSE_DONE,
), session
assert session.errors == (), session
assert _session_updates(observed) == (
GA_INJECTED_UPDATE,
_realtime_update(_without_auto_response(TUNED_SERVER_VAD)),
), _session_updates(observed)
def test_client_response_cancel_error_still_reaches_the_client(guardrail_proxy: OwnedProxy) -> None:
with guardrail_proxy.gateway.scenario() as scenario:
handle: Final = _scripted(scenario, _voice_scenario(CLEAN_TRANSCRIPT))
key: Final = scenario.key()
model: Final = _openai_deployment(scenario, handle.scenario_id, model=VOICE_MODEL)
cancel: Final[dict[str, JsonValue]] = {"type": "response.cancel", "event_id": CLIENT_CANCEL_ID}
session: Final = _drive(
_ws_base(_owned_url(guardrail_proxy)), f"model={model}", key, (_step(cancel, until="error"),)
)
assert session.errors == (CANCEL_NOT_ACTIVE,), session
assert _error_event_ids(session) == (CLIENT_CANCEL_ID,), session
def _xai_blocked_session(
guardrail_proxy: OwnedProxy, scenario: Scenario
) -> tuple[Session, tuple[dict[str, JsonValue], ...]]:
handle: Final = _scripted(scenario, _xai_voice_scenario(BLOCKED_TRANSCRIPT))
key: Final = scenario.key()
model: Final = scenario.model(model=f"xai/{XAI_VOICE_MODEL}", api_key=handle.scenario_id)
frames: Final = _voice_frames(_realtime_update(TUNED_SERVER_VAD))
session: Final = asyncio.run(
_talked_until_the_proxy_answer(_ws_base(_owned_url(guardrail_proxy)), f"model={model}", key, frames)
)
return session, _observed(guardrail_proxy.gateway.upstream_url, handle.scenario_id)
def test_xai_voice_session_gets_untyped_gates_and_no_unguarded_answer(guardrail_proxy: OwnedProxy) -> None:
with guardrail_proxy.gateway.scenario() as scenario:
session, observed = _xai_blocked_session(guardrail_proxy, scenario)
turn_detections: Final = _upstream_turn_detections(observed)
assert turn_detections == (UNTYPED_GATE, UNTYPED_GATE), turn_detections
assert session.transcripts == (BLOCKED_TRANSCRIPT,), session
assert GUARDRAIL_VIOLATION in session.errors, session
assert not any(response.startswith(AUTO_ANSWER_PREFIX) for response in _response_ids(session)), session
def test_xai_voice_session_block_sends_xai_no_item_delete(guardrail_proxy: OwnedProxy) -> None:
with guardrail_proxy.gateway.scenario() as scenario:
session, observed = _xai_blocked_session(guardrail_proxy, scenario)
assert session.transcripts == (BLOCKED_TRANSCRIPT,), session
assert _sent_types(observed) == (
"session.update",
"session.update",
"input_audio_buffer.commit",
*XAI_VOICE_BLOCK_FRAMES,
), _sent(observed)
@pytest.mark.xfail(
strict=True,
raises=AssertionError,
reason=(
"xAI answers the guardrail's response.cancel with an error carrying its own event_id, so the proxy cannot "
"tell it apart from an error for a client event and forwards it, as the merge base does"
),
)
def test_xai_guardrail_cancel_error_does_not_reach_the_client(guardrail_proxy: OwnedProxy) -> None:
with guardrail_proxy.gateway.scenario() as scenario:
session, _ = _xai_blocked_session(guardrail_proxy, scenario)
assert session.errors == (GUARDRAIL_VIOLATION,), session
TUNED_VOICE_UPDATES: Final = (
pytest.param(_realtime_update(TUNED_SERVER_VAD), id="nested"),
pytest.param({"turn_detection": TUNED_SERVER_VAD}, id="flat"),
)
@pytest.mark.parametrize("update", TUNED_VOICE_UPDATES)
def test_gemini_ai_studio_voice_session_with_vad_tuning_connects_under_the_guardrail(
guardrail_proxy: OwnedProxy, update: dict[str, JsonValue]
) -> None:
with guardrail_proxy.gateway.scenario() as scenario:
handle: Final = _scripted(scenario, _live_scenario())
key: Final = scenario.key()
upstream_url: Final = guardrail_proxy.gateway.upstream_url
model: Final = _gemini_studio_deployment(scenario, handle.scenario_id, upstream_url)
session: Final = _drive(
_ws_base(_owned_url(guardrail_proxy)),
f"model={model}",
key,
(_step(_update_frame(update), until="session.updated"),),
)
observed: Final = _observed(upstream_url, handle.scenario_id)
assert session.types == ("session.created", "session.updated"), session
assert _setup_detection(observed) == {"disabled": True}, _sent(observed)
def test_vertex_live_voice_session_keeps_vad_tuning_under_the_guardrail(guardrail_proxy: OwnedProxy) -> None:
with guardrail_proxy.gateway.scenario() as scenario:
handle: Final = _scripted(scenario, _live_scenario())
key: Final = scenario.key()
upstream_url: Final = guardrail_proxy.gateway.upstream_url
model: Final = _vertex_live_deployment(scenario, handle.scenario_id, upstream_url)
session: Final = _drive(
_ws_base(_owned_url(guardrail_proxy)),
f"model={model}",
key,
(_step(_update_frame(_realtime_update(TUNED_SERVER_VAD)), until="session.updated"),),
)
observed: Final = _observed(upstream_url, handle.scenario_id)
assert session.types == ("session.created", "session.updated"), session
assert _setup_detection(observed) == GEMINI_TUNED_DETECTION, _sent(observed)
MUSE_PUSH_TO_TALK_FRAMES: Final[tuple[dict[str, JsonValue], ...]] = (
{"binary_bytes": MUSE_PACKET_BYTES},
{"binary_bytes": MUSE_PACKET_BYTES},
@ -1041,10 +1392,11 @@ def _muse_session(
turn_detection: JsonValue,
*,
until: str | None = None,
drop_params: bool = False,
) -> tuple[Session, tuple[dict[str, JsonValue], ...]]:
handle: Final = _scripted(scenario, _muse_scenario(transcript), control_url=tls_upstream)
key: Final = scenario.key()
model: Final = _muse_deployment(scenario, handle.scenario_id, tls_upstream)
model: Final = _muse_deployment(scenario, handle.scenario_id, tls_upstream, drop_params=drop_params)
verdict: Final = TRANSCRIPT_COMPLETED if BLOCKED_WORD not in transcript else "error"
session: Final = _talk(
_ws_base(_owned_url(guardrail_proxy)),
@ -1120,6 +1472,40 @@ def test_muse_server_vad_transcription_session_keeps_endpointing(
), _muse_audio_frames(observed)
def test_muse_semantic_vad_under_drop_params_falls_back_to_endpointing(
guardrail_proxy: OwnedProxy, tls_upstream: str
) -> None:
with guardrail_proxy.gateway.scenario() as scenario:
session, observed = _muse_session(
guardrail_proxy,
scenario,
tls_upstream,
CLEAN_TRANSCRIPT,
f"&{TRANSCRIPTION_QUERY}",
{"type": "semantic_vad"},
drop_params=True,
)
assert session.transcripts == (CLEAN_TRANSCRIPT,), session
assert session.errors == (), session
assert _muse_handshake(observed)["mode"] == ENDPOINTING, _muse_handshake(observed)
def test_muse_semantic_vad_without_drop_params_is_refused(guardrail_proxy: OwnedProxy, tls_upstream: str) -> None:
with guardrail_proxy.gateway.scenario() as scenario:
session, observed = _muse_session(
guardrail_proxy,
scenario,
tls_upstream,
CLEAN_TRANSCRIPT,
"",
{"type": "semantic_vad"},
until=TRANSCRIPT_COMPLETED,
)
assert session.types[-1] == "closed", session
assert session.transcripts == (), session
assert _upgrades(observed) == (), observed
@pytest.mark.xfail(
strict=True,
reason=(
@ -1229,8 +1615,8 @@ def test_opt_in_guardrail_gates_a_voice_session_on_an_opted_out_key(guardrail_pr
)
observed: Final = _observed(guardrail_proxy.gateway.upstream_url, handle.scenario_id)
assert session.transcripts == (BLOCKED_TRANSCRIPT,), session
assert session.errors == (MISSING_TURN_DETECTION_TYPE, GUARDRAIL_VIOLATION), session
assert _session_updates(observed) == (GA_INJECTED_UPDATE, VOICE_UPDATE_FORWARDED), _session_updates(observed)
assert session.errors == (GUARDRAIL_VIOLATION,), session
_assert_voice_updates(observed)
def test_opt_in_guardrail_leaves_a_transcription_session_alone_on_an_opted_out_key(guardrail_proxy: OwnedProxy) -> None:
@ -1717,12 +2103,11 @@ def test_voice_session_guardrail_raising_runtime_error_closes_with_the_same_prox
assert session.types == (
"session.created",
"session.updated",
"error",
"session.updated",
TRANSCRIPT_COMPLETED,
"error",
"closed",
), session
assert session.errors[0] == MISSING_TURN_DETECTION_TYPE, session
_assert_closed_by_a_proxy_failure(session)
assert _sent_types(observed) == ("session.update", "session.update", "input_audio_buffer.commit"), _sent(
observed
@ -1740,8 +2125,8 @@ def test_voice_session_guardrail_raising_value_error_is_voiced_through_the_backe
(_step(VOICE_UPDATE_FRAME, COMMIT, until=RESPONSE_DONE),),
)
assert session.transcripts == (blocked,), session
assert session.errors == (MISSING_TURN_DETECTION_TYPE, GUARDRAIL_VIOLATION), session
assert session.error_messages[1] == f"{VALUE_ERROR_WORD} is not allowed", session
assert session.errors == (GUARDRAIL_VIOLATION,), session
assert session.error_messages == (f"{VALUE_ERROR_WORD} is not allowed",), session
assert session.types[-1] == RESPONSE_DONE, session
assert _sent_types(observed) == (
"session.update",
@ -1764,7 +2149,7 @@ def test_voice_session_second_violation_under_end_session_after_n_fails_closes_a
(_step(VOICE_UPDATE_FRAME, COMMIT, until=RESPONSE_DONE), _step(COMMIT, until="closed")),
)
assert session.transcripts == (blocked, blocked), session
assert session.errors == (MISSING_TURN_DETECTION_TYPE, GUARDRAIL_VIOLATION, GUARDRAIL_VIOLATION), session
assert session.errors == (GUARDRAIL_VIOLATION, GUARDRAIL_VIOLATION), session
assert session.types[-1] == "closed", session
assert session.close_code == GUARDRAIL_END_CLOSE, session
assert _sent_types(observed) == (
@ -1948,7 +2333,7 @@ def test_non_string_transcript_field_closes_a_voice_session_with_the_same_proxy_
assert session.types == (
"session.created",
"session.updated",
"error",
"session.updated",
TRANSCRIPT_COMPLETED,
"error",
"closed",

14
tests/integration/providers/test_realtime_unguarded_voice_turns.py
View file

@ -114,12 +114,10 @@ TRANSCRIPT_FIELDS: Final = (
pytest.param({"transcript": {"text": "a"}}, id="object"),
)
VOICE_UPDATE: Final[dict[str, JsonValue]] = {"type": "realtime", "instructions": "answer briefly"}
VOICE_UPDATE_FORWARDED: Final[dict[str, JsonValue]] = {
"type": "realtime",
"instructions": "answer briefly",
"audio": {"input": {"turn_detection": {"create_response": False}}},
VOICE_UPDATE_GATED: Final[dict[str, JsonValue]] = {
**VOICE_UPDATE,
"audio": {"input": {"turn_detection": {"type": "server_vad", "create_response": False}}},
}
TURN_DETECTION_TYPE_MISSING: Final = "Missing required parameter: 'session.audio.input.turn_detection.type'."
INJECTED_UPDATE: Final[dict[str, JsonValue]] = {
"type": "realtime",
"audio": {"input": {"turn_detection": {"type": "server_vad", "create_response": False}}},
@ -982,15 +980,15 @@ def test_list_mode_transcript_guardrail_still_turns_auto_response_off_and_answer
assert session.types == (
"session.created",
"session.updated",
"error",
"session.updated",
TRANSCRIPT_COMPLETED,
RESPONSE_DONE,
), session
assert session.transcript_fields == ({"transcript": TRANSCRIPT},), session
assert session.error_messages == (TURN_DETECTION_TYPE_MISSING,), session
assert session.error_messages == (), session
assert _sent(observed) == (
_update_frame(INJECTED_UPDATE),
_update_frame(VOICE_UPDATE_FORWARDED),
_update_frame(VOICE_UPDATE_GATED),
COMMIT,
PROXY_RESPONSE_CREATE,
), _sent(observed)

6
tests/unit/integration_support/test_upstream.py
View file

@ -21,7 +21,11 @@ async def test_realtime_disconnect_after_session_creation_keeps_unexpected_error
response: Final = RealtimeResponse(content_type="application/x-realtime", events=({"type": "response.done"},))
provider.scenario_store.put("owned-scenario", response)
incoming: Final[deque[Message]] = deque(
({"type": "websocket.connect"}, {"type": "websocket.receive", "text": '{"type":"response.create"}'})
(
{"type": "websocket.connect"},
{"type": "websocket.receive", "text": '{"type":"response.create"}'},
{"type": "websocket.disconnect", "code": 1006},
)
)
outgoing: Final[deque[Message]] = deque()

320
tests/unit/litellm_core_utils/test_realtime_streaming.py
View file

@ -2351,8 +2351,7 @@ async def test_guardrail_turn_detection_injection_tolerates_non_dict_value(
injected_turn_detection = session_obj.get("turn_detection") or session_obj.get("audio", {}).get("input", {}).get(
"turn_detection"
)
assert isinstance(injected_turn_detection, dict)
assert injected_turn_detection["create_response"] is False
assert injected_turn_detection == {"type": "server_vad", "create_response": False}
assert streaming._guardrail_turn_detection_update_sent is True
@ -2417,8 +2416,7 @@ async def test_subsequent_session_update_cannot_reenable_vad_when_guardrails_act
forwarded_turn_detection = session_obj.get("turn_detection") or session_obj.get("audio", {}).get("input", {}).get(
"turn_detection"
)
assert isinstance(forwarded_turn_detection, dict)
assert forwarded_turn_detection["create_response"] is False
assert forwarded_turn_detection == {"type": "server_vad", "create_response": False}
@pytest.mark.asyncio
@ -3927,3 +3925,317 @@ async def test_transcription_session_guardrail_block_only_reports_violation(
sent_to_backend: Final = [call.args[0] for call in backend_ws.send.await_args_list]
assert sent_to_backend == [], sent_to_backend
assert backend_ws.close.await_count == (1 if expect_session_closed else 0)
_GA_GUARDRAIL_GATE: Final = {
"type": "realtime",
"audio": {"input": {"turn_detection": {"type": "server_vad", "create_response": False}}},
}
async def _forwarded_ga_session_updates(
monkeypatch: pytest.MonkeyPatch,
*client_sessions: dict[str, object],
event_normalizer: XAIRealtimeNormalizer | None = None,
) -> tuple[dict[str, object], ...]:
"""Run a GA client's session.updates through a transcript-guardrailed proxy and return the forwarded sessions."""
monkeypatch.setattr(litellm, "callbacks", [_transcription_guardrail()])
client_ws: Final = AsyncMock()
client_ws.scope = {"headers": []}
client_ws.receive_text = AsyncMock(
side_effect=[
*(json.dumps({"type": "session.update", "session": session}) for session in client_sessions),
ConnectionClosed(None, None),
]
)
backend_ws: Final = MagicMock()
backend_ws.send = AsyncMock()
logging_obj: Final = MagicMock()
logging_obj.async_success_handler = AsyncMock()
logging_obj.success_handler = MagicMock()
await RealTimeStreaming(client_ws, backend_ws, logging_obj, event_normalizer=event_normalizer).client_ack_messages()
return tuple(json.loads(call.args[0])["session"] for call in backend_ws.send.await_args_list)
@pytest.mark.asyncio
async def test_first_ga_session_update_keeps_client_turn_detection_when_guardrail_disables_auto_response(
monkeypatch: pytest.MonkeyPatch,
):
"""A transcript guardrail only flips create_response: the client's nested GA
turn_detection, its type included, must reach the backend intact, or OpenAI
rejects the whole update and the client's session config is lost."""
forwarded: Final = await _forwarded_ga_session_updates(
monkeypatch,
{
"type": "realtime",
"instructions": "hi",
"audio": {
"input": {
"turn_detection": {"type": "server_vad", "silence_duration_ms": 700, "create_response": True},
"transcription": {"model": "gpt-4o-transcribe"},
}
},
},
)
gate, client_update = forwarded
assert gate == _GA_GUARDRAIL_GATE
assert "turn_detection" not in client_update
assert client_update["instructions"] == "hi"
assert client_update["audio"]["input"] == {
"turn_detection": {"type": "server_vad", "silence_duration_ms": 700, "create_response": False},
"transcription": {"model": "gpt-4o-transcribe"},
}
@pytest.mark.asyncio
async def test_first_ga_session_update_without_turn_detection_gets_a_typed_gate(monkeypatch: pytest.MonkeyPatch):
forwarded: Final = await _forwarded_ga_session_updates(monkeypatch, {"type": "realtime", "instructions": "hi"})
gate, client_update = forwarded
assert gate == _GA_GUARDRAIL_GATE
assert "turn_detection" not in client_update
assert client_update["audio"]["input"]["turn_detection"] == {"type": "server_vad", "create_response": False}
@pytest.mark.asyncio
async def test_later_ga_session_update_without_turn_detection_is_forwarded_without_one(
monkeypatch: pytest.MonkeyPatch,
):
"""OpenAI keeps the earlier create_response=false for a field left unset, while a
bare create_response carries no type and gets the whole update rejected."""
forwarded: Final = await _forwarded_ga_session_updates(
monkeypatch,
{"type": "realtime", "audio": {"input": {"turn_detection": {"type": "server_vad", "create_response": True}}}},
{"type": "realtime", "instructions": "later"},
)
gate, first, later = forwarded
assert gate == _GA_GUARDRAIL_GATE
assert first["audio"]["input"]["turn_detection"] == {"type": "server_vad", "create_response": False}
assert later == {"type": "realtime", "instructions": "later"}
_XAI_TYPED_VAD_SESSION: Final = {
"type": "realtime",
"audio": {"input": {"turn_detection": {"type": "server_vad", "silence_duration_ms": 700}}},
}
@pytest.mark.asyncio
@pytest.mark.parametrize(
"client_sessions",
[
pytest.param(({"type": "realtime", "instructions": "hi"},), id="no-turn-detection"),
pytest.param(({"type": "realtime", "audio": {"input": {"turn_detection": None}}},), id="push-to-talk"),
pytest.param((_XAI_TYPED_VAD_SESSION,), id="typed-server-vad"),
pytest.param(
({"type": "realtime", "instructions": "hi"}, _XAI_TYPED_VAD_SESSION), id="later-typed-server-vad"
),
],
)
async def test_xai_session_update_gate_leaves_out_the_vad_type(
monkeypatch: pytest.MonkeyPatch, client_sessions: tuple[dict[str, object], ...]
):
"""xAI answers on its own under a typed turn_detection even with create_response off,
so every turn_detection the proxy forwards to xAI under the guardrail must carry no type."""
forwarded: Final = await _forwarded_ga_session_updates(
monkeypatch, *client_sessions, event_normalizer=XAIRealtimeNormalizer()
)
assert [session["audio"]["input"]["turn_detection"] for session in forwarded] == [
{"create_response": False}
] * (len(client_sessions) + 1)
@pytest.mark.asyncio
async def test_xai_session_created_gate_leaves_out_the_vad_type(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(litellm, "callbacks", [_transcription_guardrail()])
client_ws: Final = MagicMock()
client_ws.send_text = AsyncMock()
backend_ws: Final = MagicMock()
backend_ws.send = AsyncMock()
backend_ws.recv = AsyncMock(
side_effect=[
json.dumps({"type": "session.created", "session": {"turn_detection": {"type": None}}}).encode(),
ConnectionClosed(None, None),
]
)
logging_obj: Final = _mock_logging()
logging_obj.async_success_handler = AsyncMock()
logging_obj.success_handler = MagicMock()
streaming: Final = RealTimeStreaming(client_ws, backend_ws, logging_obj, event_normalizer=XAIRealtimeNormalizer())
await streaming.backend_to_client_send_messages()
sent: Final = [json.loads(call.args[0]) for call in backend_ws.send.await_args_list]
assert sent == [
{
"type": "session.update",
"session": {"type": "realtime", "audio": {"input": {"turn_detection": {"create_response": False}}}},
}
]
@pytest.mark.asyncio
@pytest.mark.parametrize(
"client_turn_detection",
[
pytest.param({"type": "semantic_vad", "eagerness": "low"}, id="semantic-vad"),
pytest.param({"type": "server_vad", "silence_duration_ms": 700}, id="tuned-server-vad"),
],
)
async def test_guardrail_gate_goes_ahead_of_a_session_update_sent_before_session_created(
monkeypatch: pytest.MonkeyPatch, client_turn_detection: dict[str, object]
):
"""OpenAI applies events in order, so the gate must precede a client update sent on connect:
the client's turn_detection then wins, and when OpenAI refuses that update the gate still holds."""
monkeypatch.setattr(litellm, "callbacks", [_transcription_guardrail()])
client_ws: Final = AsyncMock()
client_ws.scope = {"headers": []}
client_ws.receive_text = AsyncMock(
side_effect=[
json.dumps(
{
"type": "session.update",
"session": {"type": "realtime", "audio": {"input": {"turn_detection": client_turn_detection}}},
}
),
ConnectionClosed(None, None),
]
)
backend_ws: Final = MagicMock()
backend_ws.send = AsyncMock()
backend_ws.recv = AsyncMock(
side_effect=[json.dumps({"type": "session.created", "session": {}}).encode(), ConnectionClosed(None, None)]
)
logging_obj: Final = _mock_logging()
logging_obj.async_success_handler = AsyncMock()
logging_obj.success_handler = MagicMock()
streaming: Final = RealTimeStreaming(client_ws, backend_ws, logging_obj)
await streaming.client_ack_messages()
await streaming.backend_to_client_send_messages()
sent: Final = [json.loads(call.args[0]) for call in backend_ws.send.await_args_list]
assert [event["session"]["audio"]["input"]["turn_detection"] for event in sent] == [
{"type": "server_vad", "create_response": False},
{**client_turn_detection, "create_response": False},
]
@pytest.mark.asyncio
async def test_client_never_sees_the_error_caused_by_the_guardrail_cancel(monkeypatch: pytest.MonkeyPatch):
"""OpenAI answers the guardrail's response.cancel with response_cancel_not_active when nothing
is answering yet, an error about an event the client never sent."""
class _BlockingGuardrail(CustomGuardrail):
async def apply_guardrail(self, inputs, request_data, input_type, logging_obj=None):
raise ValueError("blocked")
monkeypatch.setattr(
litellm,
"callbacks",
[
_BlockingGuardrail(
guardrail_name="blocking", event_hook=GuardrailEventHooks.realtime_input_transcription, default_on=True
)
],
)
client_ws: Final = MagicMock()
client_ws.send_text = AsyncMock()
backend_ws: Final = MagicMock()
backend_ws.send = AsyncMock()
def cancel_not_active(event_id: str) -> bytes:
return json.dumps(
{"type": "error", "error": {"code": "response_cancel_not_active", "event_id": event_id}}
).encode()
def proxy_cancel_event_id() -> str:
sent = [json.loads(call.args[0]) for call in backend_ws.send.await_args_list]
return next(event["event_id"] for event in sent if event["type"] == "response.cancel")
backend_frames: Final = iter(
(
lambda: _make_transcript_event("blocked words"),
lambda: cancel_not_active(proxy_cancel_event_id()),
lambda: cancel_not_active("evt_from_client"),
lambda: cancel_not_active("litellm_guardrail_from_client"),
)
)
async def recv(**_kwargs: object) -> bytes:
frame = next(backend_frames, None)
if frame is None:
raise ConnectionClosed(None, None)
return frame()
backend_ws.recv = recv
logging_obj: Final = _mock_logging()
logging_obj.async_success_handler = AsyncMock()
logging_obj.success_handler = MagicMock()
await RealTimeStreaming(client_ws, backend_ws, logging_obj).backend_to_client_send_messages()
client_errors: Final = [
event["error"] for call in client_ws.send_text.await_args_list if (event := json.loads(call.args[0]))["type"] == "error"
]
assert [(error["code"], error.get("event_id")) for error in client_errors] == [
("content_policy_violation", None),
("response_cancel_not_active", "evt_from_client"),
("response_cancel_not_active", "litellm_guardrail_from_client"),
]
@pytest.mark.asyncio
@pytest.mark.parametrize(
("event_normalizer", "expected_frames"),
[
(None, ["response.cancel", "conversation.item.delete", "conversation.item.create", "response.create"]),
(XAIRealtimeNormalizer(), ["response.cancel", "conversation.item.create", "response.create"]),
],
ids=["deletes_the_blocked_item", "xai_keeps_the_item_it_cannot_delete"],
)
async def test_blocked_transcript_is_removed_from_the_conversation_before_the_refusal_turn(
monkeypatch: pytest.MonkeyPatch,
event_normalizer: XAIRealtimeNormalizer | None,
expected_frames: list[str],
):
class _BlockingGuardrail(CustomGuardrail):
async def apply_guardrail(self, inputs, request_data, input_type, logging_obj=None):
raise ValueError("blocked")
monkeypatch.setattr(
litellm,
"callbacks",
[
_BlockingGuardrail(
guardrail_name="blocking", event_hook=GuardrailEventHooks.realtime_input_transcription, default_on=True
)
],
)
client_ws: Final = MagicMock()
client_ws.send_text = AsyncMock()
backend_ws: Final = MagicMock()
backend_ws.send = AsyncMock()
backend_frames: Final = iter((_make_transcript_event("blocked words", item_id="item_blocked"),))
async def recv(**_kwargs: object) -> bytes:
frame = next(backend_frames, None)
if frame is None:
raise ConnectionClosed(None, None)
return frame
backend_ws.recv = recv
logging_obj: Final = _mock_logging()
logging_obj.async_success_handler = AsyncMock()
logging_obj.success_handler = MagicMock()
await RealTimeStreaming(
client_ws, backend_ws, logging_obj, event_normalizer=event_normalizer
).backend_to_client_send_messages()
sent: Final = [json.loads(call.args[0]) for call in backend_ws.send.await_args_list]
assert [event["type"] for event in sent] == expected_frames
assert all(event["item_id"] == "item_blocked" for event in sent if event["type"] == "conversation.item.delete")

36
tests/unit/llms/gemini/realtime/test_gemini_realtime_transformation.py
View file

@ -1,6 +1,6 @@
import json
from collections.abc import Mapping
from typing import cast
from typing import Final, cast
from unittest.mock import MagicMock
import pytest
@ -1323,6 +1323,40 @@ def test_gemini_realtime_pipecat_semantic_vad_omits_realtime_input_config():
assert setup["tools"][0]["function_declarations"][0]["name"] == "terminate_call"
@pytest.mark.parametrize(
("create_response", "expected_detection"),
[
(False, {"disabled": True}),
(True, {"disabled": False, "prefixPaddingMs": 200, "silenceDurationMs": 700}),
],
)
def test_gemini_turn_detection_tuning_is_dropped_when_auto_response_is_off(
create_response: bool, expected_detection: dict[str, object]
):
session_update: Final = {
"type": "session.update",
"session": {
"audio": {
"input": {
"turn_detection": {
"type": "server_vad",
"create_response": create_response,
"prefix_padding_ms": 200,
"silence_duration_ms": 700,
}
},
},
},
}
messages: Final = GeminiRealtimeConfig().transform_realtime_request(
json.dumps(session_update),
"gemini-live-2.5-flash-native-audio",
session_configuration_request=None,
)
setup: Final = json.loads(messages[0])["setup"]
assert setup["realtimeInputConfig"]["automaticActivityDetection"] == expected_detection
def test_gemini_input_audio_buffer_commit_maps_to_audio_stream_end():
config = GeminiRealtimeConfig()
setup = {

18
tests/unit/llms/meta/realtime/test_meta_realtime_transformation.py
View file

@ -20,6 +20,8 @@ from litellm.llms.meta.realtime.transformation import (
session_created_event,
)
from litellm.types.llms.meta import MuseMode
from litellm.types.utils import LlmProviders
from litellm.utils import ProviderConfigManager
from litellm.types.realtime import RealtimeResponseTransformInput
EMPTY_TRANSFORM_INPUT: Final[RealtimeResponseTransformInput] = {
@ -164,6 +166,22 @@ def test_session_rejects_unsupported_audio_model_and_hints(session: dict[str, ob
parse_session_update(_event("session.update", session={"type": "transcription", **session}), MUSE_MODEL)
def _handshake(turn_detection: str, drop_params: bool) -> tuple[str | bytes, ...]:
config = ProviderConfigManager.get_provider_realtime_config(MUSE_MODEL, LlmProviders.META, drop_params=drop_params)
assert config is not None
config.validate_environment({}, MUSE_MODEL, api_key="secret-token")
return tuple(config.transform_realtime_request(_ga_session_update(turn_detection=turn_detection), MUSE_MODEL))
def test_semantic_vad_falls_back_to_server_vad_under_drop_params():
assert _handshake("semantic_vad", drop_params=True) == _handshake("server_vad", drop_params=False)
def test_semantic_vad_without_drop_params_names_the_drop_params_fallback():
with pytest.raises(MuseProtocolError, match="Set drop_params to use server_vad instead"):
_handshake("semantic_vad", drop_params=False)
def test_session_created_event_exposes_openai_transcription_shape():
config = parse_session_update(
_event(

21
tests/unit/llms/vertex_ai/audio_transcription/test_vertex_ai_realtime_transformation.py
View file

@ -17,6 +17,7 @@ from litellm.llms.vertex_ai.audio_transcription.realtime_transformation import (
parse_chirp_session_update,
)
from litellm.llms.vertex_ai.common_utils import VertexAIError
from litellm.llms.vertex_ai.realtime.transformation import vertex_realtime_config
from litellm.types.llms.vertex_ai_speech_to_text import (
VertexSpeechStreamingConfigured,
VertexSpeechStreamingResponse,
@ -173,6 +174,26 @@ def test_unsupported_session_settings_are_rejected(payload: str, message: str):
parse_chirp_session_update(payload, MODEL)
def _configure_command(turn_detection: str, drop_params: bool) -> list[object]:
config = vertex_realtime_config(
MODEL, access_token="token", resolve_access_token=_token, project="proj-1", location="us", drop_params=drop_params
)
config.transform_session_created_event(MODEL, "sess_1")
return [
json.loads(command) if isinstance(command, str) else command
for command in config.transform_realtime_request(_ga_session_update(turn_detection=turn_detection), MODEL)
]
def test_semantic_vad_falls_back_to_server_vad_under_drop_params():
assert _configure_command("semantic_vad", drop_params=True) == _configure_command("server_vad", drop_params=False)
def test_semantic_vad_without_drop_params_names_the_drop_params_fallback():
with pytest.raises(ChirpProtocolError, match="Set drop_params to use server_vad instead"):
_configure_command("semantic_vad", drop_params=False)
def test_session_update_configures_once_and_later_updates_are_ignored():
config = _config()
config.transform_session_created_event(MODEL, "sess_1")

30
tests/unit/llms/vertex_ai/realtime/test_vertex_ai_realtime_transformation.py
View file

@ -223,6 +223,36 @@ def test_vertex_session_update_normalizes_ga_remapped_fields(
)
def test_vertex_keeps_turn_detection_tuning_when_auto_response_is_off():
session_update: Final = {
"type": "session.update",
"session": {
"audio": {
"input": {
"turn_detection": {
"type": "server_vad",
"create_response": False,
"prefix_padding_ms": 200,
"silence_duration_ms": 700,
}
},
},
},
}
cfg: Final = VertexAIRealtimeConfig(access_token="tok", project="my-proj", location="us-central1")
messages: Final = cfg.transform_realtime_request(
json.dumps(session_update),
_NATIVE_AUDIO_MODEL,
session_configuration_request=None,
)
setup: Final = json.loads(messages[0])["setup"]
assert setup["realtimeInputConfig"]["automaticActivityDetection"] == {
"disabled": True,
"prefixPaddingMs": 200,
"silenceDurationMs": 700,
}
# ---------------------------------------------------------------------------
# Round-trip test: text-in / text-out via RealTimeStreaming
# ---------------------------------------------------------------------------