fix(realtime): only inject response.create when a transcript guardrail suppressed VAD auto-response (#43791)

* fix(realtime): only inject response.create when a transcript guardrail suppressed VAD auto-response

The proxy sent `response.create` to the backend after every
`conversation.item.input_audio_transcription.completed`, whether or not a
`realtime_input_transcription` guardrail had disabled the backend's server-VAD
auto-response. Without such a guardrail the backend already created the turn's
response, so the injected one was a duplicate and the backend answered every
user turn with `conversation_already_has_active_response`.

Gate the re-trigger on the same predicate as the suppress side
(`_has_audio_transcription_guardrails`), via `_proxy_drives_turns()`, in both
the raw-backend and the provider_config paths. Guardrails still run on every
transcript; only the injection is gated. Transcription-only sessions keep
skipping the injection as before.

Flip the unit test that asserted the old behaviour and add a provider-path
regression test.

Fixes #31726

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* test(realtime): assert guardrail-gated response.create on the provider_config path

Mirror of test_realtime_guardrail_allows_clean_transcript for backends that go through
provider_config. With a realtime_input_transcription guardrail the proxy disabled the
backend's auto-response, so a clean transcript must trigger exactly one response.create.
Locks in the other half of the suppress/re-trigger pair introduced for #31726.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* test(realtime): type the passthrough provider_config helper's frame argument

* test(realtime): expect no proxy response.create on an unguarded voice session in the integration suite

* test(realtime): audit cells for client-owned turns without a transcript guardrail

* test(realtime): wait for the killed worker to exit, not for its parent to reap it

* test(realtime): pin each worker-kill session's outcome to the worker that accepted it

* test(realtime): drive the provider_config path through a concrete passthrough realtime config

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>

litellm/litellm_core_utils/realtime_streaming.py

@@ -1031,7 +1031,7 @@ class RealTimeStreaming:
                     cast(str, transcript),
                     item_id=cast(str | None, event.get("item_id")),
                 )
-                if not blocked and not self._is_transcription_session:
+                if not blocked and self._should_disable_vad_auto_response():
                     await self._send_to_backend(json.dumps({"type": "response.create"}))
                 continue
             ## LOGGING
@@ -1078,7 +1078,7 @@ class RealTimeStreaming:
                 transcript,
                 item_id=event_obj.get("item_id"),
             )
-            if not blocked and not self._is_transcription_session:
+            if not blocked and self._should_disable_vad_auto_response():
                 await self._send_to_backend(json.dumps({"type": "response.create"}))
             return True
         return False

tests/integration/providers/test_realtime_transcription_guardrail_session_update.py

@@ -1139,7 +1139,7 @@ def test_muse_session_without_intent_still_delivers_the_transcript_and_the_verdi
         assert _muse_handshake(observed)["model"] == MUSE_MODEL, _muse_handshake(observed)
 
 
-def test_without_a_guardrail_the_proxy_sends_no_session_update_of_its_own(gateway: Gateway) -> None:
+def test_without_a_guardrail_the_proxy_sends_no_session_update_or_response_create_of_its_own(gateway: Gateway) -> None:
     with gateway.scenario() as scenario:
         transcription: Final = _scripted(scenario, _transcription_scenario(BLOCKED_TRANSCRIPT))
         voice: Final = _scripted(scenario, _voice_scenario(BLOCKED_TRANSCRIPT))
@@ -1147,22 +1147,29 @@ def test_without_a_guardrail_the_proxy_sends_no_session_update_of_its_own(gatewa
         transcribe_model: Final = _openai_deployment(scenario, transcription.scenario_id)
         voice_model: Final = _openai_deployment(scenario, voice.scenario_id, model=VOICE_MODEL)
         transcribed: Final = _transcribe(_ws_base(_proxy_url()), f"model={transcribe_model}&{TRANSCRIPTION_QUERY}", key)
-        talked: Final = _talk(_ws_base(_proxy_url()), f"model={voice_model}", key, _voice_frames(VOICE_UPDATE))
+        talked: Final = _talk_without_a_reply(_ws_base(_proxy_url()), f"model={voice_model}", key)
         observed: Final = _observed_all(gateway.upstream_url)
         transcription_observed: Final = _belonging(observed, transcription.scenario_id)
-        voice_observed: Final = _belonging(observed, voice.scenario_id)
         _assert_transcription_left_alone(
             transcribed, transcription_observed, BLOCKED_TRANSCRIPT, update=GA_TRANSCRIPTION_UPDATE
         )
-        assert talked.types == ("session.created", "session.updated", TRANSCRIPT_COMPLETED, RESPONSE_DONE), talked
-        assert talked.errors == (), talked
-        assert _session_updates(voice_observed) == (VOICE_UPDATE,), _session_updates(voice_observed)
+        _assert_voice_session_ungated(talked, _belonging(observed, voice.scenario_id))
+
+
+def _talk_without_a_reply(ws_base: str, query: str, key: str) -> Session:
+    return _drive(
+        ws_base,
+        query,
+        key,
+        (_step(VOICE_UPDATE_FRAME, COMMIT, until=TRANSCRIPT_COMPLETED), _step(until=RESPONSE_DONE)),
+    )
 
 
 def _assert_voice_session_ungated(session: Session, observed: tuple[dict[str, JsonValue], ...]) -> None:
-    assert session.types == ("session.created", "session.updated", TRANSCRIPT_COMPLETED, RESPONSE_DONE), session
+    assert session.types == ("session.created", "session.updated", TRANSCRIPT_COMPLETED, "timeout"), session
     assert session.transcripts == (BLOCKED_TRANSCRIPT,), session
     assert session.errors == (), session
+    assert _sent_types(observed) == ("session.update", "input_audio_buffer.commit"), _sent(observed)
     assert _session_updates(observed) == (VOICE_UPDATE,), _session_updates(observed)
 
 
@@ -1180,9 +1187,7 @@ def test_key_opted_out_of_the_transcript_guardrail_is_not_gated_by_the_prompt_gu
         handle: Final = _scripted(scenario, _voice_scenario(BLOCKED_TRANSCRIPT))
         key: Final = scenario.key(metadata={"opted_out_global_guardrails": [TRANSCRIPT_GUARDRAIL]})
         model: Final = _openai_deployment(scenario, handle.scenario_id, model=VOICE_MODEL)
-        session: Final = _talk(
-            _ws_base(_owned_url(guardrail_proxy)), f"model={model}", key, _voice_frames(VOICE_UPDATE)
-        )
+        session: Final = _talk_without_a_reply(_ws_base(_owned_url(guardrail_proxy)), f"model={model}", key)
         _assert_voice_session_ungated(session, _observed(guardrail_proxy.gateway.upstream_url, handle.scenario_id))
 
 
@@ -1199,9 +1204,7 @@ def test_team_opted_out_of_the_transcript_guardrail_is_not_gated(guardrail_proxy
         team: Final = scenario.team(metadata={"opted_out_global_guardrails": [TRANSCRIPT_GUARDRAIL]})
         key: Final = scenario.key(team_id=team)
         model: Final = _openai_deployment(scenario, handle.scenario_id, model=VOICE_MODEL)
-        session: Final = _talk(
-            _ws_base(_owned_url(guardrail_proxy)), f"model={model}", key, _voice_frames(VOICE_UPDATE)
-        )
+        session: Final = _talk_without_a_reply(_ws_base(_owned_url(guardrail_proxy)), f"model={model}", key)
         _assert_voice_session_ungated(session, _observed(guardrail_proxy.gateway.upstream_url, handle.scenario_id))
 
 

tests/unit/litellm_core_utils/test_realtime_streaming.py

@@ -1,6 +1,6 @@
 import asyncio
 import json
-from collections.abc import Coroutine, Mapping
+from collections.abc import Coroutine, Mapping, Sequence
 from dataclasses import dataclass
 from typing import Final, Literal
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -17,8 +17,11 @@ from litellm.litellm_core_utils.realtime_streaming import (
     RealTimeStreaming,
     client_sent_openai_beta_realtime_header,
 )
+from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
+from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig
 from litellm.llms.xai.realtime.transformation import XAIRealtimeNormalizer
 from litellm.types.guardrails import GuardrailEventHooks
+from litellm.types.realtime import RealtimeResponseTransformInput, RealtimeResponseTypedDict
 from litellm.types.utils import GenericGuardrailAPIInputs
 
 
@@ -903,37 +906,75 @@ async def test_transcription_session_captures_usage_and_skips_response_create():
     )
 
 
-@pytest.mark.asyncio
-async def test_non_transcription_completed_event_still_triggers_response_create():
-    """
-    Regression guard: a normal (non-transcription) session with no guardrails must
-    keep triggering response.create on a completed transcription event.
-    """
-    client_ws = MagicMock()
-    client_ws.send_text = AsyncMock()
+class _PassthroughRealtimeConfig(BaseRealtimeConfig):
+    def validate_environment(self, headers: dict, model: str, api_key: str | None = None) -> dict:
+        return headers
 
-    completed = json.dumps(
-        {
-            "type": "conversation.item.input_audio_transcription.completed",
-            "transcript": "hi",
-            "item_id": "item_1",
+    def get_complete_url(self, api_base: str | None, model: str, api_key: str | None = None) -> str:
+        return api_base or ""
+
+    def transform_realtime_request(
+        self, message: str, model: str, session_configuration_request: str | None = None
+    ) -> Sequence[str | bytes]:
+        return (message,)
+
+    def transform_realtime_response(
+        self,
+        message: str | bytes,
+        model: str,
+        logging_obj: LiteLLMLoggingObj,
+        realtime_response_transform_input: RealtimeResponseTransformInput,
+    ) -> RealtimeResponseTypedDict:
+        return {
+            "response": [json.loads(message)],
+            "current_output_item_id": realtime_response_transform_input["current_output_item_id"],
+            "current_response_id": realtime_response_transform_input["current_response_id"],
+            "current_delta_chunks": realtime_response_transform_input["current_delta_chunks"],
+            "current_conversation_id": realtime_response_transform_input["current_conversation_id"],
+            "current_item_chunks": realtime_response_transform_input["current_item_chunks"],
+            "current_delta_type": realtime_response_transform_input["current_delta_type"],
+            "session_configuration_request": realtime_response_transform_input["session_configuration_request"],
         }
-    ).encode()
 
-    backend_ws = MagicMock()
-    backend_ws.recv = AsyncMock(side_effect=[completed, ConnectionClosed(None, None)])
-    backend_ws.send = AsyncMock()
 
-    logging_obj = MagicMock()
-    logging_obj.async_success_handler = AsyncMock()
-    logging_obj.success_handler = MagicMock()
+@pytest.mark.asyncio
+@pytest.mark.parametrize("through_provider_config", [False, True], ids=["raw", "provider_config"])
+@pytest.mark.parametrize(
+    "event_hook,expected_response_creates",
+    [
+        (None, 0),
+        (GuardrailEventHooks.pre_call, 0),
+        (GuardrailEventHooks.post_call, 0),
+        (GuardrailEventHooks.realtime_input_transcription, 1),
+    ],
+    ids=["no_guardrail", "pre_call", "post_call", "realtime_input_transcription"],
+)
+async def test_transcript_triggers_response_create_only_when_a_transcript_guardrail_owns_the_turn(
+    monkeypatch: pytest.MonkeyPatch,
+    through_provider_config: bool,
+    event_hook: GuardrailEventHooks | None,
+    expected_response_creates: int,
+) -> None:
+    monkeypatch.setattr(litellm, "callbacks", [_transcription_guardrail(event_hook)] if event_hook else [])
+    completed: Final = _make_transcript_event("what are the opening hours tomorrow")
+    backend_ws: Final = MagicMock(
+        recv=AsyncMock(side_effect=[completed, ConnectionClosed(None, None)]),
+        send=AsyncMock(),
+    )
+    client_ws: Final = MagicMock(send_text=AsyncMock())
+    streaming: Final = RealTimeStreaming(
+        client_ws,
+        backend_ws,
+        MagicMock(async_success_handler=AsyncMock()),
+        provider_config=_PassthroughRealtimeConfig() if through_provider_config else None,
+    )
 
-    streaming = RealTimeStreaming(client_ws, backend_ws, logging_obj)
     await streaming.backend_to_client_send_messages()
 
-    assert streaming._is_transcription_session is False
-    sent_to_backend = [json.loads(c.args[0]) for c in backend_ws.send.call_args_list if c.args]
-    assert any(e.get("type") == "response.create" for e in sent_to_backend)
+    assert [json.loads(call.args[0]) for call in backend_ws.send.await_args_list] == [
+        {"type": "response.create"}
+    ] * expected_response_creates
+    assert [json.loads(call.args[0]) for call in client_ws.send_text.await_args_list] == [json.loads(completed)]
 
 
 def test_client_session_update_does_not_mark_transcription_session():
@@ -2865,10 +2906,10 @@ async def test_deferred_setup_clear_drops_appends_when_buffered():
     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
+def _transcription_guardrail(
+    event_hook: GuardrailEventHooks = GuardrailEventHooks.realtime_input_transcription,
+) -> CustomGuardrail:
+    """A minimal real CustomGuardrail registered for one realtime event hook."""
 
     class _TranscriptionGuardrail(CustomGuardrail):
         async def apply_guardrail(self, inputs, request_data, input_type, logging_obj=None):
@@ -2876,7 +2917,7 @@ def _transcription_guardrail():
 
     return _TranscriptionGuardrail(
         guardrail_name="test_transcription_guard",
-        event_hook=GuardrailEventHooks.realtime_input_transcription,
+        event_hook=event_hook,
         default_on=True,
     )