scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-08 22:21:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 78

fix(realtime): detect an upstream refusal from received frames, not the session log

Browse source 
The refusal predicate also required the session log to be empty, but that
log is not limited to upstream frames. With gemini_live_defer_setup the
handler stores a synthetic session.created before the relay starts, and
the transcription usage flush appends a usage event before the check
runs, so an upstream policy close with no received frames was still
logged as a $0 success. Key the check off the received-frames flag only
This commit is contained in:
mateo-berri 2026-09-04 21:24:36 -07:00
parent 14f8677bfc
commit 412c36bb8e
2 changed files with 37 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
litellm/litellm_core_utils/realtime_streaming.py
View file

@ -1135,7 +1135,7 @@ class RealTimeStreaming:
return BackendClose(code=1011, reason="proxy failed while relaying the upstream websocket")
def _backend_refused_session(self, close: BackendClose) -> bool:
return close.code != 1000 and not self._backend_sent_frames and not self.messages
return close.code != 1000 and not self._backend_sent_frames
async def log_backend_refusal(self, error: Exception) -> None:
if not self.logging_obj:

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

@ -3031,12 +3031,12 @@ async def test_session_close_flushes_unbilled_transcription_usage():
messages before log_messages runs, and never forwarded to the client."""
from typing import Final
from litellm.types.realtime import RealtimeInputAudioTranscriptionUsage
from litellm.types.realtime import RealtimeInputAudioTranscriptionUsage, RealtimeResponseTypedDict
client_ws: Final = MagicMock()
client_ws.send_text = AsyncMock()
backend_ws: Final = MagicMock()
backend_ws.recv = AsyncMock(side_effect=ConnectionClosed(None, None))
backend_ws.recv = AsyncMock(side_effect=[b'{"serverContent": {}}', ConnectionClosed(None, None)])
logging_obj: Final = MagicMock()
logging_obj.async_success_handler = AsyncMock()
logging_obj.success_handler = MagicMock()
@ -3048,7 +3048,24 @@ async def test_session_close_flushes_unbilled_transcription_usage():
"total_tokens": 171,
"input_token_details": {"text_tokens": 0, "audio_tokens": 153},
}
transcript_frame: Final[RealtimeResponseTypedDict] = {
"response": {
"type": "conversation.item.input_audio_transcription.completed",
"event_id": "event_1",
"transcript": "ahoy",
"item_id": "item_1",
"content_index": 0,
},
"current_output_item_id": None,
"current_response_id": None,
"current_delta_chunks": None,
"current_conversation_id": None,
"current_item_chunks": None,
"current_delta_type": None,
"session_configuration_request": None,
}
provider_config: Final = MagicMock()
provider_config.transform_realtime_response = MagicMock(return_value=transcript_frame)
provider_config.unbilled_usage_on_session_close = MagicMock(return_value=usage)
streaming: Final = RealTimeStreaming(
@ -3080,7 +3097,9 @@ async def test_session_close_flushes_unbilled_transcription_usage():
)
assert len(flushed) == 1
assert flushed[0] in logged_snapshots[0]
assert not client_ws.send_text.called
forwarded: Final = tuple(json.loads(call.args[0]) for call in client_ws.send_text.await_args_list)
assert [event.get("transcript") for event in forwarded] == ["ahoy"]
assert all("usage" not in event for event in forwarded)
@pytest.mark.asyncio
@ -3246,6 +3265,20 @@ async def test_upstream_refusal_before_any_frame_logs_a_failure_not_a_success():
assert session.logging.logged_sessions == ()
@pytest.mark.asyncio
async def test_upstream_refusal_after_a_synthetic_session_created_still_logs_a_failure():
"""LIT-6973: deferred Gemini Live setup stores a synthetic ``session.created`` before
the relay starts. It is not an upstream frame, so a refusal after it is still a refusal."""
upstream_close: Final = ConnectionClosed(Close(1008, _UPSTREAM_REFUSAL), None)
session: Final = _relay_session(_client_ws_that_never_sends(), _backend_ws_closing_with(upstream_close))
session.streaming.store_message(json.dumps({"type": "session.created", "session": {"id": "sess_synthetic"}}))
await session.run()
assert session.logging.logged_failures == (upstream_close,)
assert session.logging.logged_sessions == ()
@pytest.mark.asyncio
async def test_upstream_close_after_relayed_events_still_logs_the_session_as_success():
client_ws: Final = _client_ws_that_never_sends()