scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-13 23:11:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 100

Merge pull request #31782 from BerriAI/litellm_backport_1_90_x_bp_31519_31733

Browse source 
chore(release): backport #31519, #31733 to stable/1.90.x and cut 1.90.2
This commit is contained in:
yuneng-jiang 2026-06-30 19:09:44 -07:00 • committed by GitHub
commit 1e60f26596
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 436 additions and 273 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

41
litellm/litellm_core_utils/realtime_streaming.py
View file

@ -5,6 +5,7 @@ from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union, cast
import litellm
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig
from litellm.types.llms.openai import (
OpenAIRealtimeEvents,
@ -327,8 +328,10 @@ class RealTimeStreaming:
self.tool_calls
)
## ASYNC LOGGING
# Create an event loop for the new thread
asyncio.create_task(self.logging_obj.async_success_handler(self.messages))
# Route through the bounded logging worker (per-coroutine timeout +
# concurrency cap) instead of a bare create_task, so a slow callback
# can't leave suspended tasks pinning each call's response in memory.
GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue(self.logging_obj.async_success_handler(self.messages))
## SYNC LOGGING
executor.submit(self.logging_obj.success_handler(self.messages))
@ -357,6 +360,7 @@ class RealTimeStreaming:
# send, causing subsequent client session.update messages to
# be treated as "subsequent" and dropped even though the
# backend never received the original setup.
msg = self._maybe_inject_guardrail_auto_response_disable(msg)
await self.backend_ws.send(msg) # type: ignore[union-attr, attr-defined]
self._cache_session_configuration_request(msg)
sent = True
@ -617,6 +621,39 @@ class RealTimeStreaming:
if sent:
self._guardrail_turn_detection_update_sent = True
def _maybe_inject_guardrail_auto_response_disable(self, setup_message: str) -> str:
"""Fold the transcription-guardrail auto-response disable into the setup.
Gemini/Vertex Live reject a second ``setup`` (1007), so the guardrail's
``automaticActivityDetection.disabled=true`` cannot be delivered as a
follow-up session.update; it must live in the one-and-only setup, or a
``realtime_input_transcription`` guardrail is bypassed (the model
auto-responds before the proxy can gate the turn). Applies only to the
bidi ``setup`` shape; OpenAI sessions accept follow-up updates and so are
left untouched (handled by ``_maybe_send_guardrail_turn_detection_update``).
"""
if self._guardrail_turn_detection_update_sent:
return setup_message
if not self._has_audio_transcription_guardrails():
return setup_message
try:
obj = json.loads(setup_message)
except (json.JSONDecodeError, TypeError):
return setup_message
setup = obj.get("setup") if isinstance(obj, dict) else None
if not isinstance(setup, dict):
return setup_message
automatic = setup.setdefault("realtimeInputConfig", {}).setdefault(
"automaticActivityDetection", {}
)
automatic["disabled"] = True
self._guardrail_turn_detection_update_sent = True
verbose_logger.debug(
"Realtime: folded automaticActivityDetection.disabled=true into setup "
"for transcription-guardrail gating"
)
return json.dumps(obj)
def _has_realtime_guardrails_for_event_hooks(
self,
event_hooks: List[Any],

94
litellm/llms/custom_httpx/llm_http_handler.py
View file

@ -5457,6 +5457,58 @@ class BaseLLMHTTPHandler:
new_query = parsed.query + ("&" if parsed.query else "") + urlencode(extras)
return urlunparse(parsed._replace(query=new_query))
@staticmethod
async def _open_realtime_backend_ws(
websockets_module: Any,
url: str,
headers: dict,
ssl_context: Any,
*,
open_timeout: float = 8.0,
max_attempts: int = 3,
) -> Any:
"""Open the backend realtime websocket, retrying a hung open handshake.
The upstream Live handshake (e.g. Gemini Live) intermittently hangs on
open; waiting longer never recovers a hung attempt, but a fresh attempt
almost always connects in ~1s. So bound each attempt with ``open_timeout``
and retry, instead of surfacing one slow handshake to the caller as a
fatal 1011. A bounded attempt that timed out already spaced out the
retry, so no extra backoff is needed. Deterministic rejections (auth /
handshake status) are not retried.
"""
# Handshake-status rejections are deterministic (auth / 4xx): retrying
# cannot help and the caller must see the upstream status, not a generic
# 1011. websockets <15 raises InvalidStatusCode, >=15 raises InvalidStatus.
deterministic_errors = tuple(
exc
for exc in (
getattr(websockets_module.exceptions, "InvalidStatus", None),
getattr(websockets_module.exceptions, "InvalidStatusCode", None),
)
if exc is not None
)
last_exc: Optional[BaseException] = None
for _ in range(max_attempts):
try:
return await websockets_module.connect(
url,
additional_headers=headers,
max_size=REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES,
ssl=ssl_context,
open_timeout=open_timeout,
)
except deterministic_errors:
raise
except (
TimeoutError,
OSError,
websockets_module.exceptions.WebSocketException,
) as e:
last_exc = e
assert last_exc is not None # loop only exits via return or a captured exc
raise last_exc
async def async_realtime(
self,
model: str,
@ -5491,22 +5543,10 @@ class BaseLLMHTTPHandler:
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE
async with websockets.connect( # type: ignore
url,
additional_headers=headers,
max_size=REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES,
ssl=ssl_context,
) as backend_ws:
# Auto-send session setup if the provider requires it
# (e.g. Gemini/Vertex AI Live needs a `setup` message before any realtime_input)
_session_config: Optional[str] = None
if provider_config.requires_session_configuration():
_session_config = provider_config.session_configuration_request(
model
)
if _session_config:
await backend_ws.send(_session_config)
backend_ws = await self._open_realtime_backend_ws(
websockets, url, headers, ssl_context
)
async with backend_ws:
_request_data: Dict[str, Any] = {}
if litellm_metadata:
_request_data["litellm_metadata"] = litellm_metadata
@ -5524,8 +5564,26 @@ class BaseLLMHTTPHandler:
else None
),
)
if _session_config:
realtime_streaming.session_configuration_request = _session_config
# Auto-send session setup if the provider requires it (e.g.
# Gemini/Vertex AI Live needs a `setup` before any realtime_input).
# Build the streaming handler first so a transcription guardrail's
# auto-response disable can be folded into this one setup: Gemini
# rejects a second setup, so a follow-up disable would be dropped
# and the guardrail bypassed.
_session_config: Optional[str] = None
if provider_config.requires_session_configuration():
_session_config = provider_config.session_configuration_request(
model
)
if _session_config:
_session_config = (
realtime_streaming._maybe_inject_guardrail_auto_response_disable(
_session_config
)
)
await backend_ws.send(_session_config)
realtime_streaming.session_configuration_request = _session_config
# For providers that defer setup until client session.update, optionally
# send synthetic session.created to unblock clients waiting on connect.

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

@ -433,17 +433,11 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
Handle session.update by sending setup to Gemini.
On the FIRST session.update (when session_configuration_request is None),
the full setup with all configuration is sent.
Subsequent session.update messages are forwarded as a follow-up setup
with the new fields merged into the original setup. Gemini Live treats
a follow-up BidiGenerateContentSetup as a full session replacement
rather than a partial merge, so we carry forward the previous setup
(tools, generationConfig, inputAudioTranscription, systemInstruction,
...) and overlay the new fields on top. This preserves the old
behavior where clients could refine the session via session.update
(e.g. add tools after the auto-setup on connect), and also keeps the
guardrail-driven turn_detection update working.
the full setup with all configuration is sent. Gemini Live accepts setup
as the first-and-only client message, so every later session.update is
dropped rather than forwarded as a second setup (which Gemini rejects
with a 1007, tearing the session down). To carry tools/instructions, send
them on the first session.update before any conversation content.
"""
session_payload = json_message.get("session") or {}
# Normalize GA-remapped fields (``output_modalities``,
@ -472,82 +466,32 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
)
]
if not new_overrides:
# Gemini Live accepts exactly one ``setup`` message: the first and only
# client message. A second ``setup`` closes the socket with
# ``1007 Request contains an invalid argument``, so a session.update
# after the initial setup must not be forwarded as a follow-up setup.
# Every GA client (pipecat included) sends several session.updates while
# configuring the session; forwarding a second one tears the session down
# before the first turn, which surfaces to callers as silence after the
# first response, reconnect/retry latency churn, and 1011 errors. Drop
# it. The Vertex subclass already drops subsequent setups for this exact
# reason; the constraint is identical on AI Studio.
client_turn_detection = self._extract_turn_detection(session_payload)
if (
isinstance(client_turn_detection, dict)
and client_turn_detection.get("create_response") is False
):
verbose_logger.warning(
"Gemini Realtime: Dropping subsequent session.update "
"(turn_detection.create_response=False) — Gemini Live rejects a "
"second setup message, so audio-transcription guardrails cannot "
"suppress the model's auto-response mid-session."
)
else:
verbose_logger.debug(
"Gemini Realtime: Ignoring session.update (no mappable fields)"
"Gemini Realtime: Ignoring session.update (setup already sent)"
)
return []
try:
original_setup = cast(
BidiGenerateContentSetup,
json.loads(session_configuration_request).get("setup", {}),
)
except (json.JSONDecodeError, AttributeError):
original_setup = {}
# Deep-merge ``generationConfig`` and ``realtimeInputConfig`` so a
# partial session.update (e.g. only ``temperature`` or only
# ``modalities``) does not silently drop unrelated sub-keys
# (``responseModalities``, ``maxOutputTokens``, ...) from the original
# setup.
follow_up_setup: BidiGenerateContentSetup = {
**original_setup,
**new_overrides,
"model": f"models/{model}",
}
original_generation_config = original_setup.get("generationConfig")
new_generation_config = new_overrides.get("generationConfig")
if isinstance(original_generation_config, dict) and isinstance(
new_generation_config, dict
):
follow_up_setup["generationConfig"] = {
**original_generation_config,
**new_generation_config,
}
original_realtime_input_config = original_setup.get("realtimeInputConfig")
new_realtime_input_config = new_overrides.get("realtimeInputConfig")
if isinstance(original_realtime_input_config, dict) and isinstance(
new_realtime_input_config, dict
):
merged_realtime_input_config = {
**original_realtime_input_config,
**new_realtime_input_config,
}
# Deep-merge ``automaticActivityDetection`` so a partial VAD
# update (e.g. the guardrail-injected ``disabled: True`` from
# ``create_response: False``) does not silently drop unrelated
# knobs like ``silenceDurationMs`` / ``prefixPaddingMs`` from
# the original setup.
original_automatic_activity_detection = original_realtime_input_config.get(
"automaticActivityDetection"
)
new_automatic_activity_detection = new_realtime_input_config.get(
"automaticActivityDetection"
)
if isinstance(original_automatic_activity_detection, dict) and isinstance(
new_automatic_activity_detection, dict
):
merged_realtime_input_config["automaticActivityDetection"] = {
**original_automatic_activity_detection,
**new_automatic_activity_detection,
}
follow_up_setup["realtimeInputConfig"] = cast(
BidiGenerateContentRealtimeInputConfig,
merged_realtime_input_config,
)
verbose_logger.debug(
"Gemini Realtime: Forwarding session.update as follow-up setup"
)
return [
json.dumps(
{
"setup": self._finalize_gemini_live_setup(
model, cast(Dict[str, Any], follow_up_setup)
)
}
)
]
return []
def _handle_conversation_item(self, json_message: dict) -> List[str]:
"""

4
pyproject.toml
View file

@ -1,6 +1,6 @@
[project]
name = "litellm"
version = "1.90.1"
version = "1.90.2"
description = "Library to easily interface with LLM API providers"
readme = "README.md"
requires-python = ">=3.10, <3.14"
@ -272,7 +272,7 @@ source-exclude = [
profile = "black"
[tool.commitizen]
version = "1.90.1"
version = "1.90.2"
version_files = [
"pyproject.toml:^version",
]

92
tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
View file

@ -2681,3 +2681,95 @@ async def test_deferred_setup_clear_drops_appends_when_buffered():
streaming._buffer_pending_message_until_setup(new_audio)
assert streaming._pending_messages_until_setup == [new_audio]
def _transcription_guardrail():
"""A minimal real CustomGuardrail registered for the realtime transcript hook."""
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.types.guardrails import GuardrailEventHooks
class _TranscriptionGuardrail(CustomGuardrail):
async def apply_guardrail(
self, inputs, request_data, input_type, logging_obj=None
):
return inputs
return _TranscriptionGuardrail(
guardrail_name="test_transcription_guard",
event_hook=GuardrailEventHooks.realtime_input_transcription,
default_on=True,
)
def test_setup_folds_in_auto_response_disable_when_transcription_guardrail_active():
"""Gemini rejects a second setup, so a transcription guardrail's auto-response
disable must be folded into the one-and-only setup; otherwise the model
auto-responds and the guardrail is bypassed."""
import litellm
litellm.callbacks = [_transcription_guardrail()]
try:
streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock())
setup = json.dumps(
{
"setup": {
"model": "models/gemini-3.1-flash-live-preview",
"generationConfig": {"responseModalities": ["AUDIO"]},
"inputAudioTranscription": {},
}
}
)
out = json.loads(streaming._maybe_inject_guardrail_auto_response_disable(setup))
aad = out["setup"]["realtimeInputConfig"]["automaticActivityDetection"]
assert aad["disabled"] is True
finally:
litellm.callbacks = []
def test_setup_unchanged_without_transcription_guardrail():
import litellm
litellm.callbacks = []
streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock())
setup = json.dumps(
{"setup": {"model": "x", "generationConfig": {"responseModalities": ["AUDIO"]}}}
)
out = streaming._maybe_inject_guardrail_auto_response_disable(setup)
assert json.loads(out) == json.loads(setup)
def test_non_bidi_setup_left_untouched_for_followup_capable_providers():
"""OpenAI realtime accepts a follow-up session.update, so a non-bidi message
(no top-level 'setup' key) must be left untouched even with a guardrail on."""
import litellm
litellm.callbacks = [_transcription_guardrail()]
try:
streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock())
msg = json.dumps({"type": "session.update", "session": {"instructions": "hi"}})
assert streaming._maybe_inject_guardrail_auto_response_disable(msg) == msg
finally:
litellm.callbacks = []
@pytest.mark.asyncio
async def test_log_messages_routes_async_logging_through_bounded_worker():
"""Realtime success logging must go through GLOBAL_LOGGING_WORKER (bounded
queue + per-coroutine timeout), not a bare asyncio.create_task. A bare task
has no timeout/concurrency cap, so when a logging callback is slow every
realtime turn leaves a suspended task pinning its response in memory -> an
unbounded leak. Regression for that fix."""
logging_obj = MagicMock()
streaming = RealTimeStreaming(MagicMock(), MagicMock(), logging_obj)
streaming.messages = [{"type": "session.created"}]
with (
patch("litellm.litellm_core_utils.realtime_streaming.GLOBAL_LOGGING_WORKER") as mock_worker,
patch("litellm.litellm_core_utils.realtime_streaming.asyncio.create_task") as mock_create_task,
patch("litellm.litellm_core_utils.realtime_streaming.executor.submit"),
):
await streaming.log_messages()
mock_worker.ensure_initialized_and_enqueue.assert_called_once()
# the bare create_task path must no longer be used for success logging
mock_create_task.assert_not_called()

93
tests/test_litellm/llms/custom_httpx/test_llm_http_handler.py
View file

@ -1051,3 +1051,96 @@ def test_async_compact_handler_sends_json_when_not_signed():
)
assert kwargs.get("json") == {"model": "openai.gpt-5.5", "input": "hi"}
assert "data" not in kwargs
class _FakeWSExceptions:
class WebSocketException(Exception):
pass
class InvalidStatusCode(WebSocketException):
def __init__(self) -> None:
super().__init__("HTTP 403")
# websockets>=15 raises InvalidStatus (not InvalidStatusCode) for a rejected
# client handshake; both must be treated as deterministic.
class InvalidStatus(WebSocketException):
def __init__(self) -> None:
super().__init__("HTTP 401")
class _FakeWebsocketsModule:
"""Stand-in for the ``websockets`` module so the realtime backend-open retry
can be exercised without a real network handshake (dependency injection,
no monkeypatching)."""
def __init__(self, outcomes):
# outcomes: list where each item is either an Exception to raise or a
# sentinel object to return as the "connected" websocket.
self._outcomes = list(outcomes)
self.exceptions = _FakeWSExceptions
self.attempts = 0
self.open_timeouts: list = []
async def connect(self, *args, **kwargs):
self.attempts += 1
self.open_timeouts.append(kwargs.get("open_timeout"))
outcome = self._outcomes.pop(0)
if isinstance(outcome, Exception):
raise outcome
return outcome
@pytest.mark.asyncio
async def test_realtime_backend_open_retries_then_succeeds():
"""A hung/slow open handshake is retried; a later fresh attempt connects.
Regression for intermittent ``1011 timed out during opening handshake``:
the proxy used to surface a single slow upstream handshake to the caller as
a fatal 1011 with no retry.
"""
sentinel = object()
fake = _FakeWebsocketsModule(
[TimeoutError("timed out during opening handshake"), sentinel]
)
result = await BaseLLMHTTPHandler._open_realtime_backend_ws(
fake, "wss://backend.example/live", {"Authorization": "Bearer x"}, None
)
assert result is sentinel
assert fake.attempts == 2
# Each attempt must be bounded by a finite open_timeout (not the default/None).
assert all(t is not None and t > 0 for t in fake.open_timeouts)
@pytest.mark.asyncio
async def test_realtime_backend_open_raises_after_max_attempts():
"""When every attempt times out, the final error propagates (so the caller
still closes the client socket) rather than looping forever."""
fake = _FakeWebsocketsModule([TimeoutError("hang")] * 2)
with pytest.raises(TimeoutError):
await BaseLLMHTTPHandler._open_realtime_backend_ws(
fake, "wss://backend.example/live", {}, None, max_attempts=2
)
assert fake.attempts == 2
@pytest.mark.asyncio
@pytest.mark.parametrize(
"rejection",
[_FakeWSExceptions.InvalidStatusCode, _FakeWSExceptions.InvalidStatus],
)
async def test_realtime_backend_open_does_not_retry_auth_failure(rejection):
"""A deterministic handshake-status rejection (auth/4xx) must not be retried;
retrying cannot help and the upstream status must surface, not a 1011. Both
the websockets<15 (InvalidStatusCode) and >=15 (InvalidStatus) shapes apply."""
fake = _FakeWebsocketsModule([rejection()])
with pytest.raises(_FakeWSExceptions.WebSocketException):
await BaseLLMHTTPHandler._open_realtime_backend_ws(
fake, "wss://backend.example/live", {}, None
)
assert fake.attempts == 1

267
tests/test_litellm/llms/gemini/realtime/test_gemini_realtime_transformation.py
View file

@ -1145,60 +1145,6 @@ def test_gemini_function_call_output_includes_name():
assert "response" in function_response
def test_gemini_subsequent_session_update_forwards_tools_merged_with_original_setup():
"""A client session.update sent after the auto-setup must forward tools/
instructions as a follow-up setup, merged with the original setup so we
don't drop the pre-existing config (model, generationConfig, etc.)."""
config = GeminiRealtimeConfig()
original_setup = {
"setup": {
"model": "models/gemini-2.5-flash-native-audio",
"generationConfig": {"responseModalities": ["AUDIO"]},
"inputAudioTranscription": {},
"systemInstruction": {"role": "user", "parts": [{"text": "original"}]},
}
}
session_update = {
"type": "session.update",
"session": {
"tools": [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get weather.",
"parameters": {
"type": "object",
"properties": {"location": {"type": "string"}},
"required": ["location"],
},
},
}
],
"instructions": "Be concise.",
},
}
messages = config.transform_realtime_request(
json.dumps(session_update),
"gemini-2.5-flash-native-audio",
session_configuration_request=json.dumps(original_setup),
)
assert len(messages) == 1
follow_up = json.loads(messages[0])["setup"]
assert "tools" in follow_up
assert follow_up["tools"][0]["function_declarations"][0]["name"] == "get_weather"
# systemInstruction overwritten by client's instructions
assert follow_up["systemInstruction"]["parts"][0]["text"] == "Be concise."
# Original generationConfig / model / inputAudioTranscription preserved
assert follow_up["generationConfig"]["responseModalities"] == ["AUDIO"]
assert follow_up["model"] == "models/gemini-2.5-flash-native-audio"
assert follow_up["inputAudioTranscription"] == {}
def test_gemini_realtime_pipecat_ga_session_voice_and_tools():
"""Pipecat OpenAIRealtimeSessionProperties: output_modalities, nested tools,
and audio.output.voice (e.g. Kore) must map into Gemini setup."""
@ -1340,116 +1286,6 @@ def test_gemini_input_audio_buffer_commit_maps_to_activity_end_when_manual_vad()
assert json.loads(messages[0]) == {"realtimeInput": {"activityEnd": True}}
def test_gemini_subsequent_session_update_with_turn_detection_only_preserves_original_tools():
"""A subsequent session.update carrying only turn_detection (the
guardrail-injected disable) must keep the original tools/generationConfig."""
config = GeminiRealtimeConfig()
original_setup = {
"setup": {
"model": "models/gemini-2.5-flash-native-audio",
"generationConfig": {"responseModalities": ["AUDIO"]},
"inputAudioTranscription": {},
"tools": [
{
"function_declarations": [
{"name": "lookup", "description": "x", "parameters": {}}
]
}
],
}
}
session_update = {
"type": "session.update",
"session": {"turn_detection": {"create_response": False}},
}
messages = config.transform_realtime_request(
json.dumps(session_update),
"gemini-2.5-flash-native-audio",
session_configuration_request=json.dumps(original_setup),
)
assert len(messages) == 1
follow_up = json.loads(messages[0])["setup"]
assert follow_up["tools"] == original_setup["setup"]["tools"]
assert (
follow_up["realtimeInputConfig"]["automaticActivityDetection"]["disabled"]
is True
)
def test_gemini_follow_up_session_update_preserves_response_modalities_on_partial_generation_config():
"""A follow-up session.update that only sets `temperature` (or any other
generationConfig sub-field) must not wipe `responseModalities` from the
original setup."""
config = GeminiRealtimeConfig()
original_setup = {
"setup": {
"model": "models/gemini-2.5-flash-native-audio",
"generationConfig": {
"responseModalities": ["AUDIO"],
"maxOutputTokens": 2048,
},
"inputAudioTranscription": {},
}
}
session_update = {
"type": "session.update",
"session": {"temperature": 0.7},
}
messages = config.transform_realtime_request(
json.dumps(session_update),
"gemini-2.5-flash-native-audio",
session_configuration_request=json.dumps(original_setup),
)
follow_up = json.loads(messages[0])["setup"]
assert follow_up["generationConfig"]["responseModalities"] == ["AUDIO"]
assert follow_up["generationConfig"]["maxOutputTokens"] == 2048
assert follow_up["generationConfig"]["temperature"] == 0.7
def test_gemini_subsequent_session_update_preserves_automatic_activity_detection_subfields():
config = GeminiRealtimeConfig()
original_setup = {
"setup": {
"model": "models/gemini-2.5-flash-native-audio",
"generationConfig": {"responseModalities": ["AUDIO"]},
"realtimeInputConfig": {
"automaticActivityDetection": {
"disabled": False,
"silenceDurationMs": 500,
"prefixPaddingMs": 100,
}
},
}
}
session_update = {
"type": "session.update",
"session": {"turn_detection": {"create_response": False}},
}
messages = config.transform_realtime_request(
json.dumps(session_update),
"gemini-2.5-flash-native-audio",
session_configuration_request=json.dumps(original_setup),
)
automatic_activity_detection = json.loads(messages[0])["setup"][
"realtimeInputConfig"
]["automaticActivityDetection"]
assert automatic_activity_detection["disabled"] is True
assert automatic_activity_detection["silenceDurationMs"] == 500
assert automatic_activity_detection["prefixPaddingMs"] == 100
def test_gemini_tool_call_id_to_name_evicts_oldest_when_capped():
"""The call_id → name LRU must evict the oldest entry once the cap is
reached so long sessions with many tool calls don't grow unboundedly,
@ -1732,3 +1568,106 @@ def test_gemini_in_frame_usage_metadata_clears_pending_buffer():
assert usage["output_tokens"] == 2
assert usage["total_tokens"] == 5
assert config._pending_usage_metadata is None
def test_gemini_subsequent_session_update_is_dropped_not_resent_as_setup():
"""Regression: Gemini Live accepts exactly one ``setup`` message; a second
one closes the socket with ``1007 Request contains an invalid argument``.
Once the initial setup has been sent (``session_configuration_request`` is
set), a follow-up session.update must be dropped rather than forwarded as
another setup. Forwarding it tore the session down before the first turn,
which surfaced to callers as silence after the first response and 1011s.
"""
config = GeminiRealtimeConfig()
initial_setup = json.dumps(
{
"setup": {
"model": "models/gemini-2.5-flash",
"generationConfig": {"responseModalities": ["AUDIO"]},
"inputAudioTranscription": {},
}
}
)
follow_up = {
"type": "session.update",
"session": {"instructions": "Updated instructions", "temperature": 0.4},
}
messages = config.transform_realtime_request(
json.dumps(follow_up),
"gemini-2.5-flash",
session_configuration_request=initial_setup,
)
assert messages == [], (
"a session.update after the initial setup must be dropped, never "
"forwarded as a second setup (Gemini Live rejects it with 1007)"
)
def test_gemini_subsequent_session_update_with_new_tools_is_dropped():
"""Regression: even a follow-up session.update that *differs* from the initial
setup (e.g. registers tools after connect) must be dropped.
The previous dedup only skipped follow-ups identical to the initial setup; a
changed one was merged and re-sent as a second setup, still hitting the 1007.
Tools must instead ride on the first session.update.
"""
config = GeminiRealtimeConfig()
initial_setup = json.dumps(
{
"setup": {
"model": "models/gemini-2.5-flash",
"generationConfig": {"responseModalities": ["AUDIO"]},
}
}
)
follow_up = {
"type": "session.update",
"session": {
"tools": [
{
"type": "function",
"function": {
"name": "terminate_call",
"description": "End the call.",
"parameters": {"type": "object", "properties": {}},
},
}
]
},
}
messages = config.transform_realtime_request(
json.dumps(follow_up),
"gemini-2.5-flash",
session_configuration_request=initial_setup,
)
assert messages == []
def test_gemini_subsequent_guardrail_session_update_dropped_with_warning(caplog):
"""A dropped follow-up carrying ``turn_detection.create_response=False`` (the
transcription-guardrail signal) is still dropped, but warns so operators know
the guardrail cannot gate the model's auto-response mid-session on Gemini.
"""
import logging
config = GeminiRealtimeConfig()
initial_setup = json.dumps({"setup": {"model": "models/gemini-2.5-flash"}})
follow_up = {
"type": "session.update",
"session": {"turn_detection": {"type": "server_vad", "create_response": False}},
}
with caplog.at_level(logging.WARNING, logger="LiteLLM"):
messages = config.transform_realtime_request(
json.dumps(follow_up),
"gemini-2.5-flash",
session_configuration_request=initial_setup,
)
assert messages == []
assert any("Dropping subsequent session.update" in record.message for record in caplog.records)

4
uv.lock generated
View file

@ -9,7 +9,7 @@ resolution-markers = [
]
[options]
exclude-newer = "2026-06-27T01:16:05.524641Z"
exclude-newer = "2026-06-28T02:01:12.691586Z"
exclude-newer-span = "P3D"
[manifest]
@ -3245,7 +3245,7 @@ wheels = [
[[package]]
name = "litellm"
version = "1.90.1"
version = "1.90.2"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },