From a2b68b743b3818e302b25c5d8a4e0feb942a050b Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Tue, 7 Jul 2026 15:12:23 -0700 Subject: [PATCH] test(realtime): assert guardrail block on backend wire traffic instead of model refusal wording test_text_message_blocked_by_guardrail_no_ai_response classified the model's reply against a safe_markers keyword list to decide whether the guardrail had blocked the message. gpt-realtime words its refusal of the guardrail's "say exactly" voice prompt nondeterministically, so any new phrasing outside the list turned CI red on unrelated PRs; the list had already been extended in #28191, #28200 and #29477, and drifted again to "Sorry, I can't comply with that request" (11 of the 13 failed realtime_translation_testing runs since 2026-06-24, e.g. CircleCI job 2009316 on #32380). Record every frame the proxy sends to the backend through a RecordingBackendWebSocket wrapper and assert the invariant the product actually guarantees: the blocked phrase never reaches OpenAI, only the guardrail's own conversation.item.create and response.create are forwarded (the client's reflexive response.create is dropped), and the blocked phrase never appears in AI output. Replace the fixed 0.3s/3.0s sleeps with an event-driven wait for response.done; client frames are processed sequentially so no inter-message sleep is needed. Verified by mutation: disabling the response.create drop fails the response.create count assertion, and disabling the guardrail fails the guardrail_violation assertion. --- .../test_realtime_guardrails_openai.py | 186 +++++++----------- 1 file changed, 72 insertions(+), 114 deletions(-) diff --git a/tests/llm_translation/realtime/test_realtime_guardrails_openai.py b/tests/llm_translation/realtime/test_realtime_guardrails_openai.py index 413f5d1ff8b5..cf596aa597e9 100644 --- a/tests/llm_translation/realtime/test_realtime_guardrails_openai.py +++ b/tests/llm_translation/realtime/test_realtime_guardrails_openai.py @@ -4,7 +4,8 @@ These tests require OPENAI_API_KEY and are skipped if not set. They verify end-to-end that: - 1. A text message blocked by a guardrail -> error event sent to client, NO AI response. + 1. A text message blocked by a guardrail -> error event sent to client, the blocked + message never reaches OpenAI, and the client's response.create is not forwarded. 2. A voice transcript blocked by a guardrail -> error event sent, response.create NOT sent. 3. A clean text message passes through and triggers a real OpenAI response. @@ -55,9 +56,25 @@ def _make_guardrail(event_hook=GuardrailEventHooks.pre_call): ) -async def _wait_for_event( - client_events: List[dict], event_type: str, timeout: float = 15.0 -) -> dict: +class RecordingBackendWebSocket: + """Wraps a real backend WebSocket and records every frame sent to it.""" + + def __init__(self, backend_ws): + self._backend_ws = backend_ws + self.sent_messages: List[str] = [] + + async def send(self, message): + self.sent_messages.append(message) + await self._backend_ws.send(message) + + async def recv(self, *args, **kwargs): + return await self._backend_ws.recv(*args, **kwargs) + + async def close(self): + await self._backend_ws.close() + + +async def _wait_for_event(client_events: List[dict], event_type: str, timeout: float = 15.0) -> dict: """Poll client_events list until an event with matching type appears.""" deadline = asyncio.get_event_loop().time() + timeout while asyncio.get_event_loop().time() < deadline: @@ -65,9 +82,7 @@ async def _wait_for_event( if matching: return matching[0] await asyncio.sleep(0.05) - raise TimeoutError( - f"Timed out waiting for '{event_type}'. Got so far: {[e.get('type') for e in client_events]}" - ) + raise TimeoutError(f"Timed out waiting for '{event_type}'. Got so far: {[e.get('type') for e in client_events]}") async def _build_streaming(client_events: List[dict], backend_ws, request_data=None): @@ -99,12 +114,21 @@ async def send_text(data: str): @pytest.mark.asyncio async def test_text_message_blocked_by_guardrail_no_ai_response(): """ - Send a text message containing the blocked phrase. + Send a text message containing the blocked phrase, immediately followed by + response.create (the reflexive client pattern). Guardrail must: - Send error event (guardrail_violation) to client. - Send response.output_audio_transcript.delta (or beta-protocol - response.audio_transcript.delta) with the block message to client. - - NOT forward response.create to OpenAI (no AI response). + response.audio_transcript.delta) to client. + - NEVER forward the blocked message to OpenAI. + - Drop the client's response.create; the only response.create OpenAI sees + is the guardrail's own (which voices the block message), so the model + can never answer the blocked content. + + Assertions are on the recorded backend wire traffic, not on the model's + reply wording: gpt-realtime phrases its voicing/refusal of the guardrail + prompt nondeterministically, which made wording-based assertions flaky + (see PRs #28191, #28200, #29477). """ import websockets @@ -119,21 +143,16 @@ async def test_text_message_blocked_by_guardrail_no_ai_response(): additional_headers={ "Authorization": f"Bearer {OPENAI_API_KEY}", }, - ) as backend_ws: + ) as raw_backend_ws: + backend_ws = RecordingBackendWebSocket(raw_backend_ws) streaming, input_queue = await _build_streaming(client_events, backend_ws) - # Start backend -> client forwarding - backend_task = asyncio.create_task( - streaming.backend_to_client_send_messages() - ) - # Start client -> backend forwarding (reads from input_queue) + backend_task = asyncio.create_task(streaming.backend_to_client_send_messages()) client_task = asyncio.create_task(streaming.client_ack_messages()) try: - # Wait until session is ready await _wait_for_event(client_events, "session.created", timeout=15) - # Send the blocked message + response.create blocked_item = json.dumps( { "type": "conversation.item.create", @@ -149,34 +168,23 @@ async def test_text_message_blocked_by_guardrail_no_ai_response(): } ) await input_queue.put(blocked_item) - # Give guardrail time to process before the follow-up response.create - await asyncio.sleep(0.3) await input_queue.put(json.dumps({"type": "response.create"})) - # Allow time for guardrail round-trip - await asyncio.sleep(3.0) + await _wait_for_event(client_events, "response.done", timeout=30) finally: backend_task.cancel() client_task.cancel() await asyncio.gather(backend_task, client_task, return_exceptions=True) - # --- Assertions --- event_types = [e.get("type") for e in client_events] - # 1. Must have received guardrail error (may not be the first error event - # if the OpenAI session emits other errors, e.g. missing parameters) error_events = [e for e in client_events if e.get("type") == "error"] - guardrail_errors = [ - e - for e in error_events - if e.get("error", {}).get("type") == "guardrail_violation" - ] - assert ( - len(guardrail_errors) >= 1 - ), f"Expected at least one guardrail_violation error but got: {[e.get('error', {}).get('type') for e in error_events]}" + guardrail_errors = [e for e in error_events if e.get("error", {}).get("type") == "guardrail_violation"] + assert len(guardrail_errors) >= 1, ( + f"Expected at least one guardrail_violation error but got: {[e.get('error', {}).get('type') for e in error_events]}" + ) - # 2. Must have the guardrail message surfaced as an AI transcript delta transcript_deltas = [ e for e in client_events @@ -186,64 +194,30 @@ async def test_text_message_blocked_by_guardrail_no_ai_response(): "response.audio_transcript.delta", ) ] - assert ( - len(transcript_deltas) >= 1 - ), f"Expected guardrail message in transcript delta, got: {event_types}" - - # 3. No *real* AI response to the blocked content should have been - # generated. The original user message is blocked BEFORE it is - # forwarded to OpenAI, so the only thing the model ever sees is the - # guardrail's "say exactly: " prompt - # (see realtime_streaming.py). Two safe outcomes are possible: - # - the model voices the block message verbatim (older realtime - # snapshots did this -> text contains "blocked"), or - # - the model declines to repeat it (gpt-realtime tends to refuse - # verbatim-repeat instructions, e.g. "I'm sorry, but I can't - # repeat that message."). - # Both mean the blocked prompt itself was never answered, so we - # accept either. The hard invariant is that the blocked phrase must - # never leak into AI output, and the model must not have produced a - # normal answer to the user (which would have neither a block nor a - # refusal marker). - safe_markers = ( - "block", - "guardrail", - "content filter", - "policy", - "can't repeat", - "cannot repeat", - "can't say", - "cannot say", - "won't repeat", - "can't assist", - "can't help", - "unable to", - "i'm sorry", - "i am sorry", + assert len(transcript_deltas) >= 1, f"Expected guardrail message in transcript delta, got: {event_types}" + + sent_frames = backend_ws.sent_messages + assert all(BLOCKED_PHRASE not in frame for frame in sent_frames), ( + f"Blocked message was forwarded to OpenAI: {sent_frames}" ) + + sent_types = [json.loads(frame).get("type") for frame in sent_frames] + assert sent_types.count("response.create") == 1, ( + f"Expected only the guardrail's response.create to reach OpenAI, got backend frames: {sent_types}" + ) + assert sent_types.count("conversation.item.create") == 1, ( + f"Expected only the guardrail's conversation.item.create to reach OpenAI, got backend frames: {sent_types}" + ) + done_events = [e for e in client_events if e.get("type") == "response.done"] + assert len(done_events) >= 1, f"Expected response.done, got: {event_types}" for done in done_events: output = done.get("response", {}).get("output", []) ai_texts = [ - c.get("text", "") or c.get("transcript", "") - for item in output - for c in item.get("content", []) + c.get("text", "") or c.get("transcript", "") for item in output for c in item.get("content", []) ] real_ai_text = " ".join(ai_texts).strip() - if real_ai_text: - assert ( - BLOCKED_PHRASE not in real_ai_text - ), f"Blocked phrase leaked into AI response: {real_ai_text!r}" - normalized_ai_text = ( - real_ai_text.lower() - .replace("\u2019", "'") - .replace("\u2018", "'") - .replace("\u201c", '"') - .replace("\u201d", '"') - ) - assert any( - marker in normalized_ai_text for marker in safe_markers - ), f"AI responded with non-guardrail content even though message was blocked: {real_ai_text!r}" + assert BLOCKED_PHRASE not in real_ai_text, f"Blocked phrase leaked into AI response: {real_ai_text!r}" finally: litellm.callbacks = [] @@ -289,9 +263,7 @@ async def test_voice_transcript_blocked_by_guardrail(): # 1. Error event must be sent to client error_events = [e for e in client_events if e.get("type") == "error"] - assert ( - len(error_events) >= 1 - ), f"Expected guardrail error event, got: {event_types}" + assert len(error_events) >= 1, f"Expected guardrail error event, got: {event_types}" assert error_events[0]["error"]["type"] == "guardrail_violation" # 2. Check what was sent to backend. @@ -299,16 +271,12 @@ async def test_voice_transcript_blocked_by_guardrail(): # + response.create (to speak the block message). That's acceptable. # What we assert is that a response.cancel was sent (blocking the original). sent_to_backend = [ - json.loads(c.args[0]) - for c in backend_ws.send.call_args_list - if c.args and isinstance(c.args[0], str) - ] - response_cancels = [ - e for e in sent_to_backend if e.get("type") == "response.cancel" + json.loads(c.args[0]) for c in backend_ws.send.call_args_list if c.args and isinstance(c.args[0], str) ] - assert ( - len(response_cancels) >= 1 or len(sent_to_backend) == 0 - ), f"Guardrail should have sent response.cancel or nothing, got: {sent_to_backend}" + response_cancels = [e for e in sent_to_backend if e.get("type") == "response.cancel"] + assert len(response_cancels) >= 1 or len(sent_to_backend) == 0, ( + f"Guardrail should have sent response.cancel or nothing, got: {sent_to_backend}" + ) # Note: The guardrail may or may not send transcript deltas; the error event # (assertion #1) is the primary signal that the blocked content was handled. @@ -339,9 +307,7 @@ async def test_clean_text_message_passes_through_to_openai(): ) as backend_ws: streaming, input_queue = await _build_streaming(client_events, backend_ws) - backend_task = asyncio.create_task( - streaming.backend_to_client_send_messages() - ) + backend_task = asyncio.create_task(streaming.backend_to_client_send_messages()) client_task = asyncio.create_task(streaming.client_ack_messages()) try: @@ -353,9 +319,7 @@ async def test_clean_text_message_passes_through_to_openai(): "type": "conversation.item.create", "item": { "role": "user", - "content": [ - {"type": "input_text", "text": "Reply with just: OK"} - ], + "content": [{"type": "input_text", "text": "Reply with just: OK"}], }, } ) @@ -373,20 +337,14 @@ async def test_clean_text_message_passes_through_to_openai(): # No guardrail error should have been sent error_events = [e for e in client_events if e.get("type") == "error"] - guardrail_errors = [ - e - for e in error_events - if e.get("error", {}).get("type") == "guardrail_violation" - ] - assert ( - len(guardrail_errors) == 0 - ), f"Clean message should not trigger guardrail, got: {guardrail_errors}" + guardrail_errors = [e for e in error_events if e.get("error", {}).get("type") == "guardrail_violation"] + assert len(guardrail_errors) == 0, f"Clean message should not trigger guardrail, got: {guardrail_errors}" # AI response must be present done_events = [e for e in client_events if e.get("type") == "response.done"] - assert ( - len(done_events) >= 1 - ), f"Expected response.done from OpenAI, got: {[e.get('type') for e in client_events]}" + assert len(done_events) >= 1, ( + f"Expected response.done from OpenAI, got: {[e.get('type') for e in client_events]}" + ) finally: litellm.callbacks = []