fix: enforce realtime text guardrails and synchronize streaming cancellation

.agents/references/realtime-session-lifecycle.md

@@ -42,7 +42,9 @@ Do not add a new side effect before a failure point without defining who release
 ## Guardrails and Response Ordering
 
 - Realtime output guardrails inspect accumulated transcript text at configured debounce thresholds, not each token and not a final `Runner` output object. They emit `guardrail_tripped` instead of raising a normal Runner tripwire exception.
-- A tripped output guardrail marks the response interrupted before awaiting transport work, emits one trip event per response, forces response cancellation, and sends safe follow-up input naming the guardrail. Concurrent guardrail tasks must not interrupt or message the same response twice.
+- A tripped output guardrail marks the response interrupted before awaiting transport work, emits one trip event per response, forces response cancellation, and sends safe follow-up input naming the guardrail. Correlate that follow-up's `response.create` request with the exact response that starts; do not classify the next arbitrary turn as recovery. Custom models that cannot echo the request correlation ID must preserve response request ordering so the session can use its ordered fallback for both successful starts and create failures. If the correlated recovery response also trips, reject it normally without starting another automatic recovery; a later user turn starts a new recovery chain. Concurrent guardrail tasks must not interrupt or message the same response twice.
+- Bind each response to the agent that was active when the turn started. Later guardrail debounce thresholds must use that source-agent snapshot even if `update_agent()` or a handoff changes the current agent first.
+- Text-only guardrail trips cancel the identified response without truncating playback state from an earlier audio response. Audio transcript trips still use the playback interrupt path so consumers receive `audio_interrupted`.
 - Guardrail callbacks can run after audio has already been buffered or played. Consumers must treat `audio_interrupted` as the signal to stop local playback; text rejection alone cannot retract audio already delivered.
 - An exception from one output guardrail is logged and skipped so it does not silently terminate the live session. Exceptions that escape the background guardrail task must become a `RealtimeError` event rather than disappearing.
 - Realtime function-tool input guardrails follow the same optional pre-approval and mandatory post-approval ordering as standard function tools, but their rejection is returned through Realtime tool output and events.

examples/realtime/app/README.md

@@ -30,6 +30,26 @@ cd examples/realtime/app && LOG_LEVEL=DEBUG uv run python server.py
 
 The debug logs include concise summaries for server, model, session, history, tool, handoff, error, and usage events. Audio frames and high-volume delta events are omitted, and transcript content is not logged. Uvicorn and WebSocket protocol logging remain at INFO so `LOG_LEVEL=DEBUG` does not dump wire payloads.
 
+### Testing delayed output guardrails
+
+Enable the manual delayed guardrail when validating response lifecycle behavior:
+
+```bash
+cd examples/realtime/app
+REALTIME_GUARDRAIL_TEST=1 \
+REALTIME_GUARDRAIL_TEST_DELAY_SECONDS=1 \
+REALTIME_GUARDRAIL_TEST_TRIGGER_PHRASE="red balloon" \
+REALTIME_GUARDRAIL_TEST_DEBOUNCE_TEXT_LENGTH=1 \
+LOG_LEVEL=DEBUG \
+uv run python server.py
+```
+
+Ask the agent to say "red balloon" in a short response. Test mode lowers the debounce threshold to 1 by default so even this short output is checked. The delay makes it likely that the response finishes before the guardrail result is ready. The app should still receive a `guardrail_tripped` event and generate the safe follow-up without cancelling a newer response.
+
+The guardrail rejects every response containing the trigger phrase. If the automatic safe follow-up also contains it, the follow-up should be rejected with another `guardrail_tripped` event, but the SDK should not generate a third response automatically. A new user turn should still be checked normally and can trigger the guardrail again.
+
+To exercise the overlapping-response case, speak again immediately after the triggering response finishes but before the delayed guardrail returns. The newer response should continue without being interrupted.
+
 ## Customization
 
 To use the same UI with your own agents, edit `agent.py` and ensure get_starting_agent() returns the right starting agent for your use case.

examples/realtime/app/agent.py

@@ -1,13 +1,50 @@
 import asyncio
+import os
+from typing import Any
 
-from agents import function_tool
+from agents import GuardrailFunctionOutput, OutputGuardrail, function_tool
 from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX
 from agents.realtime import RealtimeAgent, realtime_handoff
 
 """
 When running the UI example locally, you can edit this file to change the setup. THe server
 will use the agent returned from get_starting_agent() as the starting agent."""
 
+
+async def delayed_manual_test_guardrail(
+    _context: Any,
+    _agent: Any,
+    output: str,
+) -> GuardrailFunctionOutput:
+    """Delay output validation so response lifecycle races are easy to test manually."""
+    delay_seconds = float(os.getenv("REALTIME_GUARDRAIL_TEST_DELAY_SECONDS", "1.0"))
+    trigger_phrase = os.getenv("REALTIME_GUARDRAIL_TEST_TRIGGER_PHRASE", "red balloon")
+    await asyncio.sleep(max(delay_seconds, 0.0))
+    return GuardrailFunctionOutput(
+        output_info={"trigger_phrase": trigger_phrase, "delay_seconds": delay_seconds},
+        tripwire_triggered=bool(trigger_phrase) and trigger_phrase.casefold() in output.casefold(),
+    )
+
+
+def get_manual_guardrail_test_debounce_text_length() -> int | None:
+    """Return the short manual-test debounce threshold when the guardrail is enabled."""
+    if os.getenv("REALTIME_GUARDRAIL_TEST", "").casefold() not in {"1", "true", "yes", "on"}:
+        return None
+    configured = int(os.getenv("REALTIME_GUARDRAIL_TEST_DEBOUNCE_TEXT_LENGTH", "1"))
+    return max(configured, 1)
+
+
+manual_test_output_guardrails = (
+    [
+        OutputGuardrail(
+            guardrail_function=delayed_manual_test_guardrail,
+            name="delayed_manual_test_guardrail",
+        )
+    ]
+    if get_manual_guardrail_test_debounce_text_length() is not None
+    else []
+)
+
 ### TOOLS
 
 
@@ -65,6 +102,7 @@ def get_weather(city: str) -> str:
     2. Use the faq lookup tool to answer the question. Do not rely on your own knowledge.
     3. If you cannot answer the question, transfer back to the triage agent.""",
     tools=[faq_lookup_tool],
+    output_guardrails=manual_test_output_guardrails,
 )
 
 seat_booking_agent = RealtimeAgent(
@@ -79,6 +117,7 @@ def get_weather(city: str) -> str:
     3. Use the update seat tool to update the seat on the flight.
     If the customer asks a question that is not related to the routine, transfer back to the triage agent. """,
     tools=[update_seat],
+    output_guardrails=manual_test_output_guardrails,
 )
 
 triage_agent = RealtimeAgent(
@@ -89,6 +128,7 @@ def get_weather(city: str) -> str:
         "You are a helpful triaging agent. You can use your tools to delegate questions to other appropriate agents."
     ),
     tools=[get_weather],
+    output_guardrails=manual_test_output_guardrails,
     handoffs=[
         realtime_handoff(faq_agent, tool_name_override="transfer_to_faq_agent"),
         realtime_handoff(seat_booking_agent, tool_name_override="transfer_to_seat_booking_agent"),

examples/realtime/app/server.py

@@ -14,7 +14,7 @@
 from typing_extensions import assert_never
 
 from agents.realtime import RealtimeRunner, RealtimeSession, RealtimeSessionEvent
-from agents.realtime.config import RealtimeUserInputMessage
+from agents.realtime.config import RealtimeRunConfig, RealtimeUserInputMessage
 from agents.realtime.items import RealtimeItem
 from agents.realtime.model import RealtimeModelConfig
 from agents.realtime.model_events import (
@@ -27,15 +27,15 @@
 # Import TwilioHandler class - handle both module and package use cases
 if TYPE_CHECKING:
     # For type checking, use the relative import
-    from .agent import get_starting_agent
+    from .agent import get_manual_guardrail_test_debounce_text_length, get_starting_agent
 else:
     # At runtime, try both import styles
     try:
         # Try relative import first (when used as a package)
-        from .agent import get_starting_agent
+        from .agent import get_manual_guardrail_test_debounce_text_length, get_starting_agent
     except ImportError:
         # Fall back to direct import (when run as a script)
-        from agent import get_starting_agent
+        from agent import get_manual_guardrail_test_debounce_text_length, get_starting_agent
 
 
 _requested_log_level = os.getenv("LOG_LEVEL", "INFO").upper()
@@ -45,6 +45,13 @@
 logger.setLevel(_log_level)
 
 
+def _get_runner_config() -> RealtimeRunConfig | None:
+    debounce_text_length = get_manual_guardrail_test_debounce_text_length()
+    if debounce_text_length is None:
+        return None
+    return {"guardrails_settings": {"debounce_text_length": debounce_text_length}}
+
+
 class RealtimeWebSocketManager:
     def __init__(self):
         self.active_sessions: dict[str, RealtimeSession] = {}
@@ -56,7 +63,7 @@ async def connect(self, websocket: WebSocket, session_id: str):
         self.websockets[session_id] = websocket
 
         agent = get_starting_agent()
-        runner = RealtimeRunner(agent)
+        runner = RealtimeRunner(agent, config=_get_runner_config())
         # If you want to customize the runner behavior, you can pass options:
         # runner_config = RealtimeRunConfig(async_tool_calls=False)
         # runner = RealtimeRunner(agent, config=runner_config)
@@ -205,7 +212,7 @@ def _log_debug_event(self, session_id: str, event: RealtimeSessionEvent) -> None
 
     def _log_debug_model_event(self, session_id: str, event: Any) -> None:
         model_event = event.data
-        if model_event.type in {"audio", "transcript_delta"}:
+        if model_event.type in {"audio", "output_text_delta", "transcript_delta"}:
             return
 
         if isinstance(model_event, RealtimeModelRawServerEvent):

src/agents/realtime/__init__.py

@@ -70,6 +70,7 @@
     RealtimeModelItemDeletedEvent,
     RealtimeModelItemUpdatedEvent,
     RealtimeModelOtherEvent,
+    RealtimeModelOutputTextDeltaEvent,
     RealtimeModelOutputTokensDetails,
     RealtimeModelToolCallEvent,
     RealtimeModelTranscriptDeltaEvent,
@@ -173,6 +174,7 @@
     "RealtimeModelItemDeletedEvent",
     "RealtimeModelItemUpdatedEvent",
     "RealtimeModelOtherEvent",
+    "RealtimeModelOutputTextDeltaEvent",
     "RealtimeModelOutputTokensDetails",
     "RealtimeModelToolCallEvent",
     "RealtimeModelTranscriptDeltaEvent",

src/agents/realtime/model_events.py

@@ -16,6 +16,10 @@ class RealtimeModelErrorEvent:
     error: Any
 
     type: Literal["error"] = "error"
+    response_create_id: str | None = None
+    """Correlation ID of the failed response.create request, when available."""
+    is_guardrail_recovery: bool | None = None
+    """Whether the failed response.create was requested for guardrail recovery, when known."""
 
 
 @dataclass
@@ -106,6 +110,17 @@ class RealtimeModelTranscriptDeltaEvent:
     type: Literal["transcript_delta"] = "transcript_delta"
 
 
+@dataclass
+class RealtimeModelOutputTextDeltaEvent:
+    """Partial text output update."""
+
+    item_id: str
+    delta: str
+    response_id: str
+
+    type: Literal["output_text_delta"] = "output_text_delta"
+
+
 @dataclass
 class RealtimeModelItemUpdatedEvent:
     """Item added to the history or updated."""
@@ -138,6 +153,12 @@ class RealtimeModelTurnStartedEvent:
     """Triggered when the model starts generating a response for a turn."""
 
     type: Literal["turn_started"] = "turn_started"
+    response_id: str | None = None
+    """Provider response ID for this turn, when available."""
+    response_create_id: str | None = None
+    """Correlation ID copied from the response.create request, when available."""
+    is_guardrail_recovery: bool | None = None
+    """Whether this turn was requested for guardrail recovery, when known."""
 
 
 @dataclass
@@ -189,6 +210,8 @@ class RealtimeModelTurnEndedEvent:
     """Triggered when the model finishes generating a response for a turn."""
 
     type: Literal["turn_ended"] = "turn_ended"
+    response_id: str | None = None
+    """Provider response ID for this turn, when available."""
 
 
 @dataclass
@@ -208,6 +231,10 @@ class RealtimeModelExceptionEvent:
     context: str | None = None
 
     type: Literal["exception"] = "exception"
+    response_create_id: str | None = None
+    """Correlation ID of the failed response.create request, when available."""
+    is_guardrail_recovery: bool | None = None
+    """Whether the failed response.create was requested for guardrail recovery, when known."""
 
 
 @dataclass
@@ -228,6 +255,7 @@ class RealtimeModelRawServerEvent:
     | RealtimeModelInputAudioTimeoutTriggeredEvent
     | RealtimeModelInputAudioTranscriptionCompletedEvent
     | RealtimeModelTranscriptDeltaEvent
+    | RealtimeModelOutputTextDeltaEvent
     | RealtimeModelItemUpdatedEvent
     | RealtimeModelItemDeletedEvent
     | RealtimeModelConnectionStatusEvent

src/agents/realtime/model_inputs.py

@@ -68,6 +68,12 @@ class RealtimeModelSendUserInput:
     user_input: RealtimeModelUserInput
     """The user input to send."""
 
+    response_create_id: str | None = None
+    """Optional correlation ID for the response.create triggered by this input.
+
+    Custom models should echo this value on the matching turn_started event when possible.
+    """
+
 
 @dataclass
 class RealtimeModelSendAudio:
@@ -98,6 +104,12 @@ class RealtimeModelSendInterrupt:
     force_response_cancel: bool = False
     """Force sending a response.cancel event even if automatic cancellation is enabled."""
 
+    response_id: str | None = None
+    """Limit the interrupt to the response that triggered it, when supported."""
+
+    cancel_response_only: bool = False
+    """Cancel the response without interrupting or truncating audio playback."""
+
 
 @dataclass
 class RealtimeModelSendSessionUpdate: