From 9978b092ccea4b0017edc2267f4bb780a74c8c95 Mon Sep 17 00:00:00 2001 From: michelligabriele Date: Thu, 28 May 2026 20:50:19 +0200 Subject: [PATCH] fix(realtime): stamp guardrail_information on read-path metadata --- .../litellm_core_utils/realtime_streaming.py | 18 ++- .../test_realtime_streaming.py | 113 ++++++++++++++++++ 2 files changed, 130 insertions(+), 1 deletion(-) diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index 33bb6d7d2ea..e5af444a2f8 100644 --- a/litellm/litellm_core_utils/realtime_streaming.py +++ b/litellm/litellm_core_utils/realtime_streaming.py @@ -405,10 +405,26 @@ class RealTimeStreaming: ): continue _already_run.add(id(callback)) + # Alias request_data["metadata"] to the read-path metadata dict on + # logging_obj.model_call_details so that the guardrail's + # standard_logging_guardrail_information stamp (start_time/end_time/ + # duration) lands where get_standard_logging_object_payload reads + # from. Without this aliasing the stamp goes into a throwaway dict + # and StandardLoggingPayload.guardrail_information stays empty. + litellm_params = self.logging_obj.model_call_details.setdefault( + "litellm_params", {} + ) + log_metadata = litellm_params.get("metadata") + if not isinstance(log_metadata, dict): + log_metadata = {} + litellm_params["metadata"] = log_metadata try: await callback.apply_guardrail( inputs={"texts": [transcript], "images": []}, - request_data={"user_api_key_dict": self.user_api_key_dict}, + request_data={ + "user_api_key_dict": self.user_api_key_dict, + "metadata": log_metadata, + }, input_type="request", ) except Exception as e: diff --git a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py index 7913efe8294..4f0299b31ec 100644 --- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py +++ b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py @@ -1770,3 +1770,116 @@ async def test_follow_up_setup_updates_cached_session_configuration_request(): await streaming.client_ack_messages() assert streaming.session_configuration_request == follow_up_setup + + +@pytest.mark.asyncio +async def test_realtime_guardrail_stamps_standard_logging_guardrail_information_on_read_path(): + """ + Regression test for the realtime guardrail observability gap. + + Before the fix, run_realtime_guardrails handed apply_guardrail a throwaway + request_data dict, so the standard_logging_guardrail_information stamp (with + start_time / end_time / duration) was discarded. After the fix, the stamp + must land on logging_obj.model_call_details["litellm_params"]["metadata"] + — the same dict get_standard_logging_object_payload reads from. + """ + + class TimingProbeGuardrail(CustomGuardrail): + async def apply_guardrail( + self, inputs, request_data, input_type, logging_obj=None + ): + self.add_standard_logging_guardrail_information_to_request_data( + guardrail_json_response={"ok": True}, + request_data=request_data, + guardrail_status="success", + start_time=1.0, + end_time=2.5, + duration=1.5, + event_type=GuardrailEventHooks.realtime_input_transcription, + ) + return inputs + + guardrail = TimingProbeGuardrail( + guardrail_name="test_timing_probe", + event_hook=GuardrailEventHooks.realtime_input_transcription, + default_on=True, + ) + litellm.callbacks = [guardrail] + try: + client_ws = MagicMock() + client_ws.send_text = AsyncMock() + backend_ws = MagicMock() + backend_ws.send = AsyncMock() + + read_path_metadata: dict = {} + logging_obj = MagicMock() + logging_obj.model_call_details = { + "litellm_params": {"metadata": read_path_metadata} + } + + streaming = RealTimeStreaming(client_ws, backend_ws, logging_obj) + blocked = await streaming.run_realtime_guardrails("hello world") + + assert blocked is False + stamped = read_path_metadata.get("standard_logging_guardrail_information") + assert stamped, ( + "Expected standard_logging_guardrail_information to be stamped on the " + "read-path metadata dict after run_realtime_guardrails returned." + ) + assert stamped[0]["guardrail_name"] == "test_timing_probe" + assert stamped[0]["start_time"] == 1.0 + assert stamped[0]["end_time"] == 2.5 + assert stamped[0]["duration"] == 1.5 + finally: + litellm.callbacks = [] + + +@pytest.mark.asyncio +async def test_realtime_guardrail_creates_metadata_when_missing(): + """ + Defensive guard: if model_call_details has no 'litellm_params.metadata' yet + (e.g. a non-proxy entry that skipped update_from_kwargs), the dispatcher + must create it rather than crash, and still land the stamp on it. + """ + + class StampingGuardrail(CustomGuardrail): + async def apply_guardrail( + self, inputs, request_data, input_type, logging_obj=None + ): + self.add_standard_logging_guardrail_information_to_request_data( + guardrail_json_response={"ok": True}, + request_data=request_data, + guardrail_status="success", + start_time=0.0, + end_time=0.1, + duration=0.1, + event_type=GuardrailEventHooks.realtime_input_transcription, + ) + return inputs + + guardrail = StampingGuardrail( + guardrail_name="test_defensive_stamp", + event_hook=GuardrailEventHooks.realtime_input_transcription, + default_on=True, + ) + litellm.callbacks = [guardrail] + try: + client_ws = MagicMock() + client_ws.send_text = AsyncMock() + backend_ws = MagicMock() + backend_ws.send = AsyncMock() + + logging_obj = MagicMock() + logging_obj.model_call_details = {} + + streaming = RealTimeStreaming(client_ws, backend_ws, logging_obj) + await streaming.run_realtime_guardrails("hello world") + + litellm_params = logging_obj.model_call_details.get("litellm_params") + assert isinstance(litellm_params, dict) + metadata = litellm_params.get("metadata") + assert isinstance(metadata, dict) + stamped = metadata.get("standard_logging_guardrail_information") + assert stamped and stamped[0]["guardrail_name"] == "test_defensive_stamp" + finally: + litellm.callbacks = []