fix(realtime): send TEXT frames and valid guardrail session.update (#28848)

* fix(realtime): send TEXT frames and valid guardrail session.update

Decode backend recv bytes before send_text so clients receive OP_TEXT JSON
events. Include turn_detection.type server_vad in injected session.update.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(realtime): skip non-UTF-8 backend binary frames

Avoid terminating the forwarding loop on UnicodeDecodeError when the
backend sends unexpected binary payloads.

Co-authored-by: Cursor <cursoragent@cursor.com>

---------

Co-authored-by: Cursor <cursoragent@cursor.com>

litellm/litellm_core_utils/realtime_streaming.py

@@ -267,18 +267,16 @@ class RealTimeStreaming:
 
     def _make_disable_auto_response_message(self) -> str:
         """Return a session.update that disables VAD auto-response."""
+        turn_detection: Dict[str, Any] = {
+            "type": "server_vad",
+            "create_response": False,
+        }
         if self._backend_uses_beta_protocol:
-            session: Dict[str, Any] = {
-                "turn_detection": {"create_response": False},
-            }
+            session: Dict[str, Any] = {"turn_detection": turn_detection}
         else:
             session = {
                 "type": "realtime",
-                "audio": {
-                    "input": {
-                        "turn_detection": {"create_response": False},
-                    }
-                },
+                "audio": {"input": {"turn_detection": turn_detection}},
             }
         return json.dumps({"type": "session.update", "session": session})
 
@@ -564,10 +562,19 @@ class RealTimeStreaming:
                 try:
                     raw_response = await self.backend_ws.recv(  # type: ignore[union-attr]
                         decode=False
-                    )  # improves performance
+                    )
                 except TypeError:
                     raw_response = await self.backend_ws.recv()  # type: ignore[union-attr, assignment]
 
+                if isinstance(raw_response, bytes):
+                    try:
+                        raw_response = raw_response.decode("utf-8")
+                    except UnicodeDecodeError:
+                        verbose_logger.warning(
+                            "Received non-UTF-8 binary frame from backend, skipping."
+                        )
+                        continue
+
                 if self.provider_config:
                     try:
                         await self._handle_provider_config_message(raw_response)

tests/test_litellm/litellm_core_utils/test_realtime_streaming.py

@@ -133,7 +133,9 @@ def test_make_disable_auto_response_message_produces_ga_shape():
         "turn_detection" not in session
     ), "turn_detection must not be at the top-level session (beta shape); use audio.input"
     # turn_detection must be nested under audio.input
-    assert session["audio"]["input"]["turn_detection"]["create_response"] is False
+    td = session["audio"]["input"]["turn_detection"]
+    assert td["type"] == "server_vad"
+    assert td["create_response"] is False
 
 
 def test_make_disable_auto_response_message_produces_beta_shape_for_beta_clients():
@@ -148,7 +150,55 @@ def test_make_disable_auto_response_message_produces_beta_shape_for_beta_clients
 
     assert msg["type"] == "session.update"
     session = msg["session"]
-    assert session == {"turn_detection": {"create_response": False}}
+    assert session == {
+        "turn_detection": {"type": "server_vad", "create_response": False}
+    }
+
+
+@pytest.mark.asyncio
+async def test_backend_to_client_send_text_receives_str_not_bytes():
+    client_ws = MagicMock()
+    client_ws.send_text = AsyncMock()
+    backend_ws = MagicMock()
+    backend_ws.recv = AsyncMock(
+        side_effect=[
+            json.dumps({"type": "session.created", "session": {}}).encode(),
+            ConnectionClosed(None, None),
+        ]
+    )
+    logging_obj = MagicMock()
+    logging_obj.async_success_handler = AsyncMock()
+    logging_obj.success_handler = MagicMock()
+    streaming = RealTimeStreaming(client_ws, backend_ws, logging_obj)
+
+    await streaming.backend_to_client_send_messages()
+
+    assert client_ws.send_text.called
+    sent = client_ws.send_text.call_args_list[0].args[0]
+    assert isinstance(sent, str)
+
+
+@pytest.mark.asyncio
+async def test_backend_to_client_skips_non_utf8_binary_frames():
+    client_ws = MagicMock()
+    client_ws.send_text = AsyncMock()
+    backend_ws = MagicMock()
+    backend_ws.recv = AsyncMock(
+        side_effect=[
+            b"\xff\xfe",
+            json.dumps({"type": "session.created", "session": {}}).encode(),
+            ConnectionClosed(None, None),
+        ]
+    )
+    logging_obj = MagicMock()
+    logging_obj.async_success_handler = AsyncMock()
+    logging_obj.success_handler = MagicMock()
+    streaming = RealTimeStreaming(client_ws, backend_ws, logging_obj)
+
+    await streaming.backend_to_client_send_messages()
+
+    assert client_ws.send_text.call_count == 1
+    assert isinstance(client_ws.send_text.call_args_list[0].args[0], str)
 
 
 @pytest.mark.asyncio