Skip to content

test(realtime): assert guardrail block on backend wire traffic instead of model refusal wording #32388

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mateo-berri merged 1 commit into from
Jul 7, 2026
Merged

test(realtime): assert guardrail block on backend wire traffic instead of model refusal wording #32388

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
186 changes: 72 additions & 114 deletions tests/llm_translation/realtime/test_realtime_guardrails_openai.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down Expand Up @@ -55,19 +56,33 @@ 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:
matching = [e for e in client_events if e.get("type") == event_type]
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):
Expand Down Expand Up @@ -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

Expand All @@ -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",
Expand All @@ -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
Expand All @@ -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: <block message>" 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 = []
Expand Down Expand Up @@ -289,26 +263,20 @@ 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.
# The guardrail may send response.cancel + conversation.item.create (block msg)
# + 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.
Expand Down Expand Up @@ -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:
Expand All @@ -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"}],
},
}
)
Expand All @@ -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 = []
Loading