fix(realtime): fail closed on a non-string transcript under a guardrail and give each turn its own guardrail metadata

litellm/litellm_core_utils/realtime_guardrails.py

@@ -62,6 +62,11 @@ def _json_object(value: object) -> Mapping[str, object] | None:
         return None
 
 
+def _turn_copy(value: object) -> object:
+    mapping: Final = _json_object(value)
+    return value if mapping is None else dict(mapping)
+
+
 def _typed_text(part: object) -> str | None:
     part_object: Final = _json_object(part)
     if part_object is None or part_object.get("type") not in _TYPED_TEXT_PART_TYPES:
@@ -264,7 +269,10 @@ async def realtime_guardrail_verdict(
         if guardrail is not None and _in_role_scope(guardrail, role)
     }
     in_scope: Final = tuple(guardrails.values())
-    guardrail_request_data: Final = {**request_data, "user_api_key_dict": user_api_key_dict}
+    guardrail_request_data: Final = {
+        **{key: _turn_copy(value) for key, value in request_data.items()},
+        "user_api_key_dict": user_api_key_dict,
+    }
     joined: Final = _joins(texts)
     screened: Final = await _screen_each(in_scope, joined, guardrail_request_data)
     if isinstance(screened, RealtimeGuardrailBlock):

litellm/litellm_core_utils/realtime_streaming.py

@@ -1670,12 +1670,18 @@ class RealTimeStreaming:
         Returns True if blocked (synthetic warning already sent to client).
         Returns False if clean (caller should send response.create to the backend).
         """
-        if not isinstance(transcript, str):
-            return False
-        verdict: Final = await self.apply_realtime_guardrails(
-            (transcript,), "user", item_id, pre_block_backend_message, event_hooks
-        )
-        return isinstance(verdict, RealtimeGuardrailBlock)
+        from litellm.types.guardrails import GuardrailEventHooks
+
+        if isinstance(transcript, str):
+            verdict: Final = await self.apply_realtime_guardrails(
+                (transcript,), "user", item_id, pre_block_backend_message, event_hooks
+            )
+            return isinstance(verdict, RealtimeGuardrailBlock)
+        if self._has_realtime_guardrails_for_event_hooks(
+            event_hooks or (GuardrailEventHooks.realtime_input_transcription,)
+        ):
+            raise TypeError(f"Realtime guardrails cannot screen a {type(transcript).__name__} transcript")
+        return False
 
     async def _screen_client_item(self, guarded_item: RealtimeGuardedItem | None) -> RealtimeGuardrailVerdict:
         from litellm.types.guardrails import GuardrailEventHooks

tests/unit/litellm_core_utils/test_realtime_guardrails.py

@@ -339,6 +339,44 @@ async def test_a_mask_in_a_message_with_more_text_parts_than_the_recursion_limit
     assert verdict == RealtimeGuardrailRewrite(texts=(*parts[:-1], "my code is [MASKED]."))
 
 
+class GuardrailLogRecordingGuardrail(CustomGuardrail):
+    def __init__(self) -> None:
+        super().__init__(guardrail_name="log-recorder", event_hook=GuardrailEventHooks.pre_call, default_on=True)
+        self.guardrail_log_lengths: list[int] = []
+
+    async def apply_guardrail(
+        self,
+        inputs: GenericGuardrailAPIInputs,
+        request_data: dict,
+        input_type: Literal["request", "response"],
+        logging_obj: LiteLLMLogging | None = None,
+    ) -> GenericGuardrailAPIInputs:
+        self.guardrail_log_lengths.append(
+            len(request_data["litellm_metadata"].get("standard_logging_guardrail_information", []))
+        )
+        return inputs
+
+
+@pytest.mark.asyncio
+async def test_a_guardrail_sees_only_the_current_turns_guardrail_log_however_long_the_session_runs() -> None:
+    session_request_data: Final = {"litellm_metadata": {"user_api_key_team_id": "team-a"}}
+    content_filter: Final = ContentFilterGuardrail(
+        guardrail_name="phrase-filter",
+        blocked_words=[BlockedWord(keyword="hunter2", action=ContentFilterAction.BLOCK)],
+        event_hook=GuardrailEventHooks.pre_call,
+        default_on=True,
+    )
+    recorder: Final = GuardrailLogRecordingGuardrail()
+
+    for turn in ("hello", "how are you", "goodbye"):
+        await realtime_guardrail_verdict(
+            (turn,), "user", (GuardrailEventHooks.pre_call,), session_request_data, None, (content_filter, recorder)
+        )
+
+    assert recorder.guardrail_log_lengths == [1, 1, 1]
+    assert session_request_data == {"litellm_metadata": {"user_api_key_team_id": "team-a"}}
+
+
 class ScopedGuardrail(RaisingGuardrail):
     def __init__(
         self, *, skip_system_message: bool | None, skip_tool_message: bool | None, scan_only_tool_results: bool | None

tests/unit/litellm_core_utils/test_realtime_streaming.py

@@ -1081,6 +1081,35 @@ async def test_a_non_string_transcript_does_not_end_the_session(transcript: obje
     backend_ws.send.assert_not_called()
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("transcript", [None, 123, ["a"]])
+async def test_a_non_string_transcript_fails_the_session_when_a_transcript_guardrail_would_screen_it(
+    transcript: object, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "callbacks", [_transcription_guardrail()])
+    client_ws: Final = MagicMock()
+    client_ws.send_text = AsyncMock()
+    transcript_event: Final = json.dumps(
+        {"type": "conversation.item.input_audio_transcription.completed", "transcript": transcript, "item_id": "i"}
+    )
+    later_event: Final = json.dumps({"type": "response.done", "response": {"id": "resp_1"}})
+    backend_ws: Final = MagicMock()
+    backend_ws.recv = AsyncMock(
+        side_effect=[transcript_event.encode(), later_event.encode(), ConnectionClosed(None, None)]
+    )
+    backend_ws.send = AsyncMock()
+    logging_obj: Final = _mock_logging()
+    logging_obj.model_call_details = {}
+    logging_obj.dispatch_success_handlers = AsyncMock()
+
+    close: Final = await RealTimeStreaming(client_ws, backend_ws, logging_obj).backend_to_client_send_messages()
+
+    assert close.code == 1011
+    sent_types: Final = [json.loads(call.args[0])["type"] for call in client_ws.send_text.call_args_list]
+    assert sent_types == ["conversation.item.input_audio_transcription.completed"]
+    backend_ws.send.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_transcription_session_captures_usage_and_skips_response_create():
     """