Realtime: forward exceptions from transport layer (#1107)

The model listens to websocket events in a separate task, and the
session receives those events in on_event. So if there's e.g. a
websocket or json exception, it's silently dropped. Similar to the
streaming text agents, we should:
- capture exceptions separately
- raise them in `async for event in session` loop.

Added tests to verify.






---
[//]: # (BEGIN SAPLING FOOTER)
* #1112
* #1111
* __->__ #1107
* #1106

Author: rm-openai

src/agents/realtime/config.py

@@ -29,6 +29,7 @@
 class RealtimeClientMessage(TypedDict):
     type: str  # explicitly required
     other_data: NotRequired[dict[str, Any]]
+    """Merged into the message body."""
 
 
 class RealtimeUserInputText(TypedDict):

src/agents/realtime/model_events.py

@@ -130,6 +130,16 @@ class RealtimeModelOtherEvent:
     type: Literal["other"] = "other"
 
 
+@dataclass
+class RealtimeModelExceptionEvent:
+    """Exception occurred during model operation."""
+
+    exception: Exception
+    context: str | None = None
+
+    type: Literal["exception"] = "exception"
+
+
 # TODO (rm) Add usage events
 
 
@@ -147,4 +157,5 @@ class RealtimeModelOtherEvent:
     RealtimeModelTurnStartedEvent,
     RealtimeModelTurnEndedEvent,
     RealtimeModelOtherEvent,
+    RealtimeModelExceptionEvent,
 ]

src/agents/realtime/openai_realtime.py

@@ -39,6 +39,7 @@
     RealtimeModelAudioInterruptedEvent,
     RealtimeModelErrorEvent,
     RealtimeModelEvent,
+    RealtimeModelExceptionEvent,
     RealtimeModelInputAudioTranscriptionCompletedEvent,
     RealtimeModelItemDeletedEvent,
     RealtimeModelItemUpdatedEvent,
@@ -130,48 +131,84 @@ async def _listen_for_messages(self):
 
         try:
             async for message in self._websocket:
-                parsed = json.loads(message)
-                await self._handle_ws_event(parsed)
+                try:
+                    parsed = json.loads(message)
+                    await self._handle_ws_event(parsed)
+                except json.JSONDecodeError as e:
+                    await self._emit_event(
+                        RealtimeModelExceptionEvent(
+                            exception=e, context="Failed to parse WebSocket message as JSON"
+                        )
+                    )
+                except Exception as e:
+                    await self._emit_event(
+                        RealtimeModelExceptionEvent(
+                            exception=e, context="Error handling WebSocket event"
+                        )
+                    )
 
-        except websockets.exceptions.ConnectionClosed:
-            # TODO connection closed handling (event, cleanup)
-            logger.warning("WebSocket connection closed")
+        except websockets.exceptions.ConnectionClosedOK:
+            # Normal connection closure - no exception event needed
+            logger.info("WebSocket connection closed normally")
+        except websockets.exceptions.ConnectionClosed as e:
+            await self._emit_event(
+                RealtimeModelExceptionEvent(
+                    exception=e, context="WebSocket connection closed unexpectedly"
+                )
+            )
         except Exception as e:
-            logger.error(f"WebSocket error: {e}")
+            await self._emit_event(
+                RealtimeModelExceptionEvent(
+                    exception=e, context="WebSocket error in message listener"
+                )
+            )
 
     async def send_event(self, event: RealtimeClientMessage) -> None:
         """Send an event to the model."""
         assert self._websocket is not None, "Not connected"
-        converted_event = {
-            "type": event["type"],
-        }
 
-        converted_event.update(event.get("other_data", {}))
+        try:
+            converted_event = {
+                "type": event["type"],
+            }
 
-        await self._websocket.send(json.dumps(converted_event))
+            converted_event.update(event.get("other_data", {}))
+
+            await self._websocket.send(json.dumps(converted_event))
+        except Exception as e:
+            await self._emit_event(
+                RealtimeModelExceptionEvent(
+                    exception=e, context=f"Failed to send event: {event.get('type', 'unknown')}"
+                )
+            )
 
     async def send_message(
         self, message: RealtimeUserInput, other_event_data: dict[str, Any] | None = None
     ) -> None:
         """Send a message to the model."""
-        message = (
-            message
-            if isinstance(message, dict)
-            else {
-                "type": "message",
-                "role": "user",
-                "content": [{"type": "input_text", "text": message}],
+        try:
+            message = (
+                message
+                if isinstance(message, dict)
+                else {
+                    "type": "message",
+                    "role": "user",
+                    "content": [{"type": "input_text", "text": message}],
+                }
+            )
+            other_data = {
+                "item": message,
             }
-        )
-        other_data = {
-            "item": message,
-        }
-        if other_event_data:
-            other_data.update(other_event_data)
+            if other_event_data:
+                other_data.update(other_event_data)
 
-        await self.send_event({"type": "conversation.item.create", "other_data": other_data})
+            await self.send_event({"type": "conversation.item.create", "other_data": other_data})
 
-        await self.send_event({"type": "response.create"})
+            await self.send_event({"type": "response.create"})
+        except Exception as e:
+            await self._emit_event(
+                RealtimeModelExceptionEvent(exception=e, context="Failed to send message")
+            )
 
     async def send_audio(self, audio: bytes, *, commit: bool = False) -> None:
         """Send a raw audio chunk to the model.
@@ -182,17 +219,23 @@ async def send_audio(self, audio: bytes, *, commit: bool = False) -> None:
             detection, this can be used to indicate the turn is completed.
         """
         assert self._websocket is not None, "Not connected"
-        base64_audio = base64.b64encode(audio).decode("utf-8")
-        await self.send_event(
-            {
-                "type": "input_audio_buffer.append",
-                "other_data": {
-                    "audio": base64_audio,
-                },
-            }
-        )
-        if commit:
-            await self.send_event({"type": "input_audio_buffer.commit"})
+
+        try:
+            base64_audio = base64.b64encode(audio).decode("utf-8")
+            await self.send_event(
+                {
+                    "type": "input_audio_buffer.append",
+                    "other_data": {
+                        "audio": base64_audio,
+                    },
+                }
+            )
+            if commit:
+                await self.send_event({"type": "input_audio_buffer.commit"})
+        except Exception as e:
+            await self._emit_event(
+                RealtimeModelExceptionEvent(exception=e, context="Failed to send audio")
+            )
 
     async def send_tool_output(
         self, tool_call: RealtimeModelToolCallEvent, output: str, start_response: bool
@@ -342,8 +385,13 @@ async def _handle_ws_event(self, event: dict[str, Any]):
                 OpenAIRealtimeServerEvent
             ).validate_python(event)
         except Exception as e:
-            logger.error(f"Invalid event: {event} - {e}")
-            # await self._emit_event(RealtimeModelErrorEvent(error=f"Invalid event: {event} - {e}"))
+            event_type = event.get("type", "unknown") if isinstance(event, dict) else "unknown"
+            await self._emit_event(
+                RealtimeModelExceptionEvent(
+                    exception=e,
+                    context=f"Failed to validate server event: {event_type}",
+                )
+            )
             return
 
         if parsed.type == "response.audio.delta":

src/agents/realtime/session.py

@@ -84,6 +84,7 @@ def __init__(
         self._run_config = run_config or {}
         self._event_queue: asyncio.Queue[RealtimeSessionEvent] = asyncio.Queue()
         self._closed = False
+        self._stored_exception: Exception | None = None
 
         # Guardrails state tracking
         self._interrupted_by_guardrail = False
@@ -130,17 +131,20 @@ async def __aiter__(self) -> AsyncIterator[RealtimeSessionEvent]:
         """Iterate over events from the session."""
         while not self._closed:
             try:
+                # Check if there's a stored exception to raise
+                if self._stored_exception is not None:
+                    # Clean up resources before raising
+                    await self._cleanup()
+                    raise self._stored_exception
+
                 event = await self._event_queue.get()
                 yield event
             except asyncio.CancelledError:
                 break
 
     async def close(self) -> None:
         """Close the session."""
-        self._closed = True
-        self._cleanup_guardrail_tasks()
-        self._model.remove_listener(self)
-        await self._model.close()
+        await self._cleanup()
 
     async def send_message(self, message: RealtimeUserInput) -> None:
         """Send a message to the model."""
@@ -228,6 +232,9 @@ async def on_event(self, event: RealtimeModelEvent) -> None:
                     info=self._event_info,
                 )
             )
+        elif event.type == "exception":
+            # Store the exception to be raised in __aiter__
+            self._stored_exception = event.exception
         elif event.type == "other":
             pass
         else:
@@ -403,3 +410,17 @@ def _cleanup_guardrail_tasks(self) -> None:
             if not task.done():
                 task.cancel()
         self._guardrail_tasks.clear()
+
+    async def _cleanup(self) -> None:
+        """Clean up all resources and mark session as closed."""
+        # Cancel and cleanup guardrail tasks
+        self._cleanup_guardrail_tasks()
+
+        # Remove ourselves as a listener
+        self._model.remove_listener(self)
+
+        # Close the model connection
+        await self._model.close()
+
+        # Mark as closed
+        self._closed = True

tests/realtime/test_openai_realtime.py

@@ -174,39 +174,34 @@ class TestEventHandlingRobustness(TestOpenAIRealtimeWebSocketModel):
 
     @pytest.mark.asyncio
     async def test_handle_malformed_json_logs_error_continues(self, model):
-        """Test that malformed JSON is logged as error but doesn't crash."""
+        """Test that malformed JSON emits exception event but doesn't crash."""
         mock_listener = AsyncMock()
         model.add_listener(mock_listener)
 
         # Malformed JSON should not crash the handler
-        with patch("agents.realtime.openai_realtime.logger") as mock_logger:
-            await model._handle_ws_event("invalid json {")
+        await model._handle_ws_event("invalid json {")
 
-            # Should log error but not crash
-            mock_logger.error.assert_called_once()
-            assert "Invalid event" in mock_logger.error.call_args[0][0]
-
-            # Should not emit any events to listeners
-            mock_listener.on_event.assert_not_called()
+        # Should emit exception event to listeners
+        mock_listener.on_event.assert_called_once()
+        exception_event = mock_listener.on_event.call_args[0][0]
+        assert exception_event.type == "exception"
+        assert "Failed to validate server event: unknown" in exception_event.context
 
     @pytest.mark.asyncio
     async def test_handle_invalid_event_schema_logs_error(self, model):
-        """Test that events with invalid schema are logged but don't crash."""
+        """Test that events with invalid schema emit exception events but don't crash."""
         mock_listener = AsyncMock()
         model.add_listener(mock_listener)
 
         invalid_event = {"type": "response.audio.delta"}  # Missing required fields
 
-        with patch("agents.realtime.openai_realtime.logger") as mock_logger:
-            await model._handle_ws_event(invalid_event)
+        await model._handle_ws_event(invalid_event)
 
-            # Should log validation error
-            mock_logger.error.assert_called_once()
-            error_msg = mock_logger.error.call_args[0][0]
-            assert "Invalid event" in error_msg
-
-            # Should not emit events to listeners
-            mock_listener.on_event.assert_not_called()
+        # Should emit exception event to listeners
+        mock_listener.on_event.assert_called_once()
+        exception_event = mock_listener.on_event.call_args[0][0]
+        assert exception_event.type == "exception"
+        assert "Failed to validate server event: response.audio.delta" in exception_event.context
 
     @pytest.mark.asyncio
     async def test_handle_unknown_event_type_ignored(self, model):