From a627597bca5ef22bea9eb6995965f58232b69b52 Mon Sep 17 00:00:00 2001 From: James Hush Date: Tue, 3 Feb 2026 16:25:07 +0800 Subject: [PATCH 1/6] Fix StopAsyncIteration in parse_telephony_websocket Handle WebSocket disconnections gracefully when telephony providers send fewer messages than expected. Adds explicit StopAsyncIteration handling for both first and second message retrieval. Co-Authored-By: Claude Sonnet 4.5 --- src/pipecat/runner/utils.py | 27 ++++++++++++++++++--------- 1 file changed, 18 insertions(+), 9 deletions(-) diff --git a/src/pipecat/runner/utils.py b/src/pipecat/runner/utils.py index f54453d93..2a56bb0f4 100644 --- a/src/pipecat/runner/utils.py +++ b/src/pipecat/runner/utils.py @@ -147,19 +147,28 @@ async def parse_telephony_websocket(websocket: WebSocket): try: # First message - first_message_raw = await start_data.__anext__() - logger.trace(f"First message: {first_message_raw}") try: - first_message = json.loads(first_message_raw) - except json.JSONDecodeError: - first_message = {} + first_message_raw = await start_data.__anext__() + logger.trace(f"First message: {first_message_raw}") + try: + first_message = json.loads(first_message_raw) + except json.JSONDecodeError: + first_message = {} + except StopAsyncIteration: + logger.error("WebSocket closed without sending any messages") + raise ValueError("WebSocket closed before receiving telephony handshake messages") # Second message - second_message_raw = await start_data.__anext__() - logger.trace(f"Second message: {second_message_raw}") try: - second_message = json.loads(second_message_raw) - except json.JSONDecodeError: + second_message_raw = await start_data.__anext__() + logger.trace(f"Second message: {second_message_raw}") + try: + second_message = json.loads(second_message_raw) + except json.JSONDecodeError: + second_message = {} + except StopAsyncIteration: + # Only one message received - use it for detection + logger.warning("Only received one WebSocket message, expected two") second_message = {} # Try auto-detection on both messages From b030f1178dc4b42e07e89276346a730c850f8a83 Mon Sep 17 00:00:00 2001 From: James Hush Date: Tue, 3 Feb 2026 16:26:09 +0800 Subject: [PATCH 2/6] Add changelog and improve docstring for parse_telephony_websocket - Added changelog entry for bug fix - Enhanced docstring with Args and Raises sections Co-Authored-By: Claude Sonnet 4.5 --- changelog/3629.fixed.md | 1 + src/pipecat/runner/utils.py | 6 ++++++ 2 files changed, 7 insertions(+) create mode 100644 changelog/3629.fixed.md diff --git a/changelog/3629.fixed.md b/changelog/3629.fixed.md new file mode 100644 index 000000000..12792f47d --- /dev/null +++ b/changelog/3629.fixed.md @@ -0,0 +1 @@ +- Fixed `StopAsyncIteration` exceptions in `parse_telephony_websocket()` when WebSocket connections close before sending expected messages. diff --git a/src/pipecat/runner/utils.py b/src/pipecat/runner/utils.py index 2a56bb0f4..a3be74436 100644 --- a/src/pipecat/runner/utils.py +++ b/src/pipecat/runner/utils.py @@ -96,6 +96,9 @@ def _detect_transport_type_from_message(message_data: dict) -> str: async def parse_telephony_websocket(websocket: WebSocket): """Parse telephony WebSocket messages and return transport type and call data. + Args: + websocket: FastAPI WebSocket connection from telephony provider. + Returns: tuple: (transport_type: str, call_data: dict) @@ -136,6 +139,9 @@ async def parse_telephony_websocket(websocket: WebSocket): "to": str, } + Raises: + ValueError: If WebSocket closes before sending any messages. + Example usage:: transport_type, call_data = await parse_telephony_websocket(websocket) From b427d534ae5889e6e1e8f5c224a08eb04cff8b50 Mon Sep 17 00:00:00 2001 From: James Hush Date: Tue, 3 Feb 2026 16:33:36 +0800 Subject: [PATCH 3/6] Add tests for parse_telephony_websocket StopAsyncIteration handling Tests cover: - No messages received (raises ValueError) - One message received (logs warning, continues) - Two messages received (normal operation) - All telephony providers (Twilio, Telnyx, Plivo, Exotel) Co-Authored-By: Claude Sonnet 4.5 --- tests/test_runner_utils.py | 153 +++++++++++++++++++++++++++++++++++++ 1 file changed, 153 insertions(+) create mode 100644 tests/test_runner_utils.py diff --git a/tests/test_runner_utils.py b/tests/test_runner_utils.py new file mode 100644 index 000000000..e5fe28786 --- /dev/null +++ b/tests/test_runner_utils.py @@ -0,0 +1,153 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import json +import unittest +from unittest.mock import AsyncMock, MagicMock + +from pipecat.runner.utils import parse_telephony_websocket + + +class MockAsyncIterator: + """Mock async iterator for WebSocket messages.""" + + def __init__(self, messages): + self.messages = messages + self.index = 0 + + def __aiter__(self): + return self + + async def __anext__(self): + if self.index >= len(self.messages): + raise StopAsyncIteration + message = self.messages[self.index] + self.index += 1 + return message + + +class TestParseTelephonyWebSocket(unittest.IsolatedAsyncioTestCase): + async def test_no_messages_raises_value_error(self): + """Test that no messages raises ValueError.""" + mock_websocket = MagicMock() + mock_websocket.iter_text.return_value = MockAsyncIterator([]) + + with self.assertRaises(ValueError) as context: + await parse_telephony_websocket(mock_websocket) + + self.assertIn("WebSocket closed before receiving", str(context.exception)) + + async def test_one_message_logs_warning_and_continues(self): + """Test that one message logs warning but continues processing.""" + twilio_message = json.dumps( + { + "event": "start", + "start": { + "streamSid": "MZ123", + "callSid": "CA123", + "customParameters": {"user_id": "test_user"}, + }, + } + ) + + mock_websocket = MagicMock() + mock_websocket.iter_text.return_value = MockAsyncIterator([twilio_message]) + + transport_type, call_data = await parse_telephony_websocket(mock_websocket) + + self.assertEqual(transport_type, "twilio") + self.assertEqual(call_data["stream_id"], "MZ123") + self.assertEqual(call_data["call_id"], "CA123") + + async def test_two_messages_normal_operation(self): + """Test normal operation with two messages.""" + first_message = json.dumps({"event": "connected"}) + twilio_message = json.dumps( + { + "event": "start", + "start": { + "streamSid": "MZ456", + "callSid": "CA456", + "customParameters": {}, + }, + } + ) + + mock_websocket = MagicMock() + mock_websocket.iter_text.return_value = MockAsyncIterator([first_message, twilio_message]) + + transport_type, call_data = await parse_telephony_websocket(mock_websocket) + + self.assertEqual(transport_type, "twilio") + self.assertEqual(call_data["stream_id"], "MZ456") + self.assertEqual(call_data["call_id"], "CA456") + + async def test_telnyx_detection(self): + """Test Telnyx provider detection.""" + telnyx_message = json.dumps( + { + "stream_id": "stream_123", + "start": { + "call_control_id": "cc_123", + "media_format": {"encoding": "PCMU"}, + "from": "+15551234567", + "to": "+15559876543", + }, + } + ) + + mock_websocket = MagicMock() + mock_websocket.iter_text.return_value = MockAsyncIterator([telnyx_message]) + + transport_type, call_data = await parse_telephony_websocket(mock_websocket) + + self.assertEqual(transport_type, "telnyx") + self.assertEqual(call_data["stream_id"], "stream_123") + self.assertEqual(call_data["call_control_id"], "cc_123") + + async def test_plivo_detection(self): + """Test Plivo provider detection.""" + plivo_message = json.dumps( + {"start": {"streamId": "stream_plivo_123", "callId": "call_plivo_123"}} + ) + + mock_websocket = MagicMock() + mock_websocket.iter_text.return_value = MockAsyncIterator([plivo_message]) + + transport_type, call_data = await parse_telephony_websocket(mock_websocket) + + self.assertEqual(transport_type, "plivo") + self.assertEqual(call_data["stream_id"], "stream_plivo_123") + self.assertEqual(call_data["call_id"], "call_plivo_123") + + async def test_exotel_detection(self): + """Test Exotel provider detection.""" + exotel_message = json.dumps( + { + "event": "start", + "start": { + "stream_sid": "stream_exo_123", + "call_sid": "call_exo_123", + "account_sid": "acc_123", + "from": "+15551111111", + "to": "+15552222222", + }, + } + ) + + mock_websocket = MagicMock() + mock_websocket.iter_text.return_value = MockAsyncIterator([exotel_message]) + + transport_type, call_data = await parse_telephony_websocket(mock_websocket) + + self.assertEqual(transport_type, "exotel") + self.assertEqual(call_data["stream_id"], "stream_exo_123") + self.assertEqual(call_data["call_id"], "call_exo_123") + self.assertEqual(call_data["account_sid"], "acc_123") + + +if __name__ == "__main__": + unittest.main() From 90bead06ab8fce226bb25f5e28b830dbeb88b151 Mon Sep 17 00:00:00 2001 From: James Hush Date: Tue, 3 Feb 2026 16:42:13 +0800 Subject: [PATCH 4/6] Apply suggestions from code review Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- src/pipecat/runner/utils.py | 2 +- tests/test_runner_utils.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/pipecat/runner/utils.py b/src/pipecat/runner/utils.py index a3be74436..5188beae2 100644 --- a/src/pipecat/runner/utils.py +++ b/src/pipecat/runner/utils.py @@ -161,7 +161,7 @@ async def parse_telephony_websocket(websocket: WebSocket): except json.JSONDecodeError: first_message = {} except StopAsyncIteration: - logger.error("WebSocket closed without sending any messages") + raise ValueError("WebSocket closed before receiving telephony handshake messages") # Second message diff --git a/tests/test_runner_utils.py b/tests/test_runner_utils.py index e5fe28786..18f156cbb 100644 --- a/tests/test_runner_utils.py +++ b/tests/test_runner_utils.py @@ -6,7 +6,7 @@ import json import unittest -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import MagicMock from pipecat.runner.utils import parse_telephony_websocket From 803a20cc00e26be38a56d7cf2201713bce1cf469 Mon Sep 17 00:00:00 2001 From: James Hush Date: Tue, 3 Feb 2026 16:46:44 +0800 Subject: [PATCH 5/6] Fix formatting: remove extra blank line Co-Authored-By: Claude Sonnet 4.5 --- src/pipecat/runner/utils.py | 1 - 1 file changed, 1 deletion(-) diff --git a/src/pipecat/runner/utils.py b/src/pipecat/runner/utils.py index 5188beae2..2208b7db4 100644 --- a/src/pipecat/runner/utils.py +++ b/src/pipecat/runner/utils.py @@ -161,7 +161,6 @@ async def parse_telephony_websocket(websocket: WebSocket): except json.JSONDecodeError: first_message = {} except StopAsyncIteration: - raise ValueError("WebSocket closed before receiving telephony handshake messages") # Second message From c1857d255d394915536c4537f8686453b6d04500 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 3 Feb 2026 12:00:04 -0500 Subject: [PATCH 6/6] Avoid nesting try/excepts --- src/pipecat/runner/utils.py | 44 ++++++++++++++++++------------------- 1 file changed, 21 insertions(+), 23 deletions(-) diff --git a/src/pipecat/runner/utils.py b/src/pipecat/runner/utils.py index 2208b7db4..d0bb44a88 100644 --- a/src/pipecat/runner/utils.py +++ b/src/pipecat/runner/utils.py @@ -149,33 +149,31 @@ async def parse_telephony_websocket(websocket: WebSocket): user_id = call_data["body"]["user_id"] """ # Read first two messages - start_data = websocket.iter_text() + message_stream = websocket.iter_text() + first_message = {} + second_message = {} try: - # First message - try: - first_message_raw = await start_data.__anext__() - logger.trace(f"First message: {first_message_raw}") - try: - first_message = json.loads(first_message_raw) - except json.JSONDecodeError: - first_message = {} - except StopAsyncIteration: - raise ValueError("WebSocket closed before receiving telephony handshake messages") + # First message - required + first_message_raw = await message_stream.__anext__() + logger.trace(f"First message: {first_message_raw}") + first_message = json.loads(first_message_raw) if first_message_raw else {} + except json.JSONDecodeError: + pass + except StopAsyncIteration: + raise ValueError("WebSocket closed before receiving telephony handshake messages") - # Second message - try: - second_message_raw = await start_data.__anext__() - logger.trace(f"Second message: {second_message_raw}") - try: - second_message = json.loads(second_message_raw) - except json.JSONDecodeError: - second_message = {} - except StopAsyncIteration: - # Only one message received - use it for detection - logger.warning("Only received one WebSocket message, expected two") - second_message = {} + try: + # Second message - optional, some providers may only send one + second_message_raw = await message_stream.__anext__() + logger.trace(f"Second message: {second_message_raw}") + second_message = json.loads(second_message_raw) if second_message_raw else {} + except json.JSONDecodeError: + pass + except StopAsyncIteration: + logger.warning("Only received one WebSocket message, expected two") + try: # Try auto-detection on both messages detected_type_first = _detect_transport_type_from_message(first_message) detected_type_second = _detect_transport_type_from_message(second_message)