From 0a02318a51db8241bed056ff7cb53347b5c0d5af Mon Sep 17 00:00:00 2001 From: a692570 Date: Wed, 29 Jul 2026 21:07:24 -0700 Subject: [PATCH 1/3] feat(telnyx): add Telnyx TTS plugin Streams text to speech over the Telnyx WebSocket endpoint and decodes the returned MP3 into PcmData as it arrives. Telnyx closes the socket after each stop frame, so the plugin connects per synthesis rather than holding one socket open. Follows the plugins/sarvam precedent of one vendor plugin covering LLM, STT, and TTS. --- plugins/telnyx/README.md | 20 ++ plugins/telnyx/pyproject.toml | 1 + plugins/telnyx/tests/test_telnyx_tts.py | 136 +++++++++ .../vision_agents/plugins/telnyx/__init__.py | 3 + .../vision_agents/plugins/telnyx/tts.py | 258 ++++++++++++++++++ uv.lock | 2 + 6 files changed, 420 insertions(+) create mode 100644 plugins/telnyx/tests/test_telnyx_tts.py create mode 100644 plugins/telnyx/vision_agents/plugins/telnyx/tts.py diff --git a/plugins/telnyx/README.md b/plugins/telnyx/README.md index dd4564e8b..8f6d5e3ac 100644 --- a/plugins/telnyx/README.md +++ b/plugins/telnyx/README.md @@ -12,6 +12,7 @@ real-time bidirectional media streaming. - **Audio Conversion**: PCMU, PCMA, and L16 RTP payload conversion - **WebSocket Management**: Handle Telnyx WebSocket media events - **Stream Bridge**: Attach a Telnyx phone participant to a Stream call +- **TTS**: Streaming text to speech over WebSocket ## Installation @@ -46,6 +47,24 @@ call.telnyx_stream = stream await stream.run() ``` +## TTS + +```python +from vision_agents.plugins import telnyx + +tts = telnyx.TTS(voice="AWS.Polly.Danielle-Neural") +``` + +Requires `TELNYX_API_KEY` in the environment, or an `api_key` argument. + +Voice ids come from `GET /v2/text-to-speech/voices`. The default is +`Telnyx.KokoroTTS.af_heart`. + +Telnyx serves each synthesis on its own WebSocket and closes the socket after +the stop frame, so the plugin reconnects per `stream_audio` call. Audio arrives +as MP3 and is decoded to `PcmData` as it streams. The output sample rate follows +the voice, so it is taken from the decoder rather than configured. + ## Examples See [examples/](examples/) for minimal inbound and outbound Telnyx phone @@ -189,5 +208,6 @@ payload = pcm_to_pcmu(pcm) ## Dependencies - vision-agents +- aiohttp - numpy - fastapi diff --git a/plugins/telnyx/pyproject.toml b/plugins/telnyx/pyproject.toml index 961aedd0e..2c74735d5 100644 --- a/plugins/telnyx/pyproject.toml +++ b/plugins/telnyx/pyproject.toml @@ -14,6 +14,7 @@ dependencies = [ "numpy>=1.24.0", # capped at <2.0 via workspace override in root pyproject.toml "cryptography>=44.0.0", "fastapi>=0.135.1", + "aiohttp>=3.13.3", ] [project.urls] diff --git a/plugins/telnyx/tests/test_telnyx_tts.py b/plugins/telnyx/tests/test_telnyx_tts.py new file mode 100644 index 000000000..29cd20476 --- /dev/null +++ b/plugins/telnyx/tests/test_telnyx_tts.py @@ -0,0 +1,136 @@ +"""Tests for the Telnyx TTS plugin.""" + +import os + +import pytest +from dotenv import load_dotenv +from vision_agents.plugins.telnyx import TTS +from vision_agents.plugins.telnyx.tts import _Id3Stripper + +load_dotenv() + + +def id3_tag(payload_size: int) -> bytes: + """Build an ID3v2 tag header plus a body of ``payload_size`` bytes.""" + synchsafe = bytes( + [ + (payload_size >> 21) & 0x7F, + (payload_size >> 14) & 0x7F, + (payload_size >> 7) & 0x7F, + payload_size & 0x7F, + ] + ) + return b"ID3\x04\x00\x00" + synchsafe + b"\xaa" * payload_size + + +class TestId3Stripper: + """Unit tests for the streaming ID3v2 tag stripper.""" + + def test_untagged_data_passes_through(self): + stripper = _Id3Stripper() + assert stripper.feed(b"\xff\xf3audio") == b"\xff\xf3audio" + + def test_leading_tag_removed(self): + stripper = _Id3Stripper() + assert stripper.feed(id3_tag(34) + b"\xff\xf3audio") == b"\xff\xf3audio" + + def test_zero_length_tag_removed(self): + stripper = _Id3Stripper() + assert stripper.feed(id3_tag(0) + b"\xff\xf3") == b"\xff\xf3" + + def test_tag_body_spanning_frames(self): + stripper = _Id3Stripper() + blob = id3_tag(40) + b"\xff\xf3audio" + assert stripper.feed(blob[:20]) == b"" + assert stripper.feed(blob[20:]) == b"\xff\xf3audio" + + def test_tag_header_spanning_frames(self): + stripper = _Id3Stripper() + blob = id3_tag(12) + b"\xff\xf3audio" + assert stripper.feed(blob[:4]) == b"" + assert stripper.feed(blob[4:]) == b"\xff\xf3audio" + + def test_tag_at_head_of_later_frame(self): + stripper = _Id3Stripper() + assert stripper.feed(b"\xff\xf3first") == b"\xff\xf3first" + assert stripper.feed(id3_tag(8) + b"\xff\xf3second") == b"\xff\xf3second" + + def test_id3_bytes_inside_audio_are_kept(self): + stripper = _Id3Stripper() + audio = b"\xff\xf3 padding ID3 more audio" + assert stripper.feed(audio) == audio + + def test_tag_consuming_whole_frame(self): + stripper = _Id3Stripper() + blob = id3_tag(100) + b"\xff\xf3audio" + assert stripper.feed(blob[:50]) == b"" + assert stripper.feed(blob[50:]) == b"\xff\xf3audio" + + +class TestTelnyxTTS: + """Unit tests for Telnyx TTS configuration.""" + + async def test_requires_api_key(self, monkeypatch): + monkeypatch.delenv("TELNYX_API_KEY", raising=False) + with pytest.raises(ValueError, match="TELNYX_API_KEY"): + TTS() + + async def test_default_configuration(self): + tts = TTS(api_key="KEY_test") + assert tts.voice == "Telnyx.KokoroTTS.af_heart" + assert tts.provider_name == "telnyx" + + async def test_custom_voice(self): + tts = TTS(api_key="KEY_test", voice="AWS.Polly.Danielle-Neural") + assert tts.voice == "AWS.Polly.Danielle-Neural" + + +@pytest.mark.skipif(not os.getenv("TELNYX_API_KEY"), reason="TELNYX_API_KEY not set") +@pytest.mark.integration +class TestTelnyxTTSIntegration: + """Integration tests against the real Telnyx streaming TTS.""" + + @pytest.fixture + async def tts(self): + instance = TTS(voice="AWS.Polly.Danielle-Neural") + try: + yield instance + finally: + await instance.close() + + async def test_stream_audio_yields_chunks(self, tts): + out = [] + async for item in tts.send_iter( + "This is a test of the Telnyx text to speech API." + ): + out.append(item) + + assert len(out) > 0 + assert out[0].data + assert out[-1].final + + async def test_decoded_audio_is_audible_and_long_enough(self, tts): + chunks = [ + item.data + async for item in tts.send_iter( + "The quick brown fox jumps over the lazy dog, and then keeps " + "running for a good while longer across the field." + ) + if item.data is not None + ] + + assert chunks + rates = {chunk.sample_rate for chunk in chunks} + assert len(rates) == 1 + total_samples = sum(len(chunk.samples) for chunk in chunks) + # A sentence this long spans more than one MP3 file, so a broken ID3 + # strip truncates the audio well before this threshold. + assert total_samples / rates.pop() > 3.0 + assert max(abs(int(chunk.samples.max())) for chunk in chunks) > 0 + + async def test_second_synthesis_reconnects(self, tts): + first = [item async for item in tts.send_iter("First utterance.")] + second = [item async for item in tts.send_iter("Second utterance.")] + + assert any(item.data is not None for item in first) + assert any(item.data is not None for item in second) diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py b/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py index cf5b476c1..5c8973565 100644 --- a/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py +++ b/plugins/telnyx/vision_agents/plugins/telnyx/__init__.py @@ -14,6 +14,7 @@ ) from .call_registry import TelnyxCall, TelnyxCallRegistry from .media_stream import TelnyxMediaFormat, TelnyxMediaStream, attach_phone_to_call +from .tts import TTS, TelnyxTTSError CallRegistry = TelnyxCallRegistry MediaStream = TelnyxMediaStream @@ -21,12 +22,14 @@ __all__ = [ "CallRegistry", "MediaStream", + "TTS", "TELNYX_DEFAULT_SAMPLE_RATE", "TELNYX_L16_SAMPLE_RATE", "TelnyxCall", "TelnyxCallRegistry", "TelnyxMediaFormat", "TelnyxMediaStream", + "TelnyxTTSError", "attach_phone_to_call", "l16_to_pcm", "pcma_to_pcm", diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/tts.py b/plugins/telnyx/vision_agents/plugins/telnyx/tts.py new file mode 100644 index 000000000..a83424d61 --- /dev/null +++ b/plugins/telnyx/vision_agents/plugins/telnyx/tts.py @@ -0,0 +1,258 @@ +"""Telnyx Text-to-Speech via WebSocket streaming. + +Docs: https://developers.telnyx.com/api/call-control/text-to-speech + +Two properties of the wire protocol drive this implementation: + +- A synthesis is primed with an init frame, then one or more text frames, then + an empty-text stop frame. Telnyx closes the socket once the stop frame has + been served, so a connection cannot be reused across ``stream_audio`` calls + the way a persistent-socket provider allows. +- The audio frames carry slices of MP3, and a synthesis can span several + concatenated MP3 files, each introduced by its own ID3v2 tag at the head of + a WebSocket frame. Those tags have to be dropped before the bytes reach the + decoder, otherwise decoding fails part way through the utterance. + +The decoded sample rate depends on the voice (Polly voices return 24 kHz, +Kokoro voices 22.05 kHz), so the rate reported by the decoder is used rather +than a configured one. +""" + +import asyncio +import base64 +import json +import logging +import os +from typing import Any, AsyncIterator, Optional, cast +from urllib.parse import urlencode + +import aiohttp +import av +from getstream.video.rtc.track_util import AudioFormat, PcmData +from vision_agents.core import tts + +logger = logging.getLogger(__name__) + +WS_TTS_URL = "wss://api.telnyx.com/v2/text-to-speech/speech" + +DEFAULT_VOICE = "Telnyx.KokoroTTS.af_heart" + +ID3_HEADER_SIZE = 10 + + +class TelnyxTTSError(Exception): + """Raised when Telnyx TTS returns an error frame over WebSocket.""" + + +class _Id3Stripper: + """Removes the ID3v2 tag heading each MP3 file in the audio stream. + + Telnyx concatenates one MP3 file per synthesised segment. Each file starts + with an ID3v2 tag at the head of a WebSocket frame, so only the head of a + frame is inspected. A tag whose header or body spans frames is carried + across calls. + """ + + def __init__(self) -> None: + self._skip = 0 + self._pending = b"" + + def feed(self, data: bytes) -> bytes: + """Return ``data`` with any leading ID3v2 tag removed.""" + if self._skip: + consumed = min(self._skip, len(data)) + data = data[consumed:] + self._skip -= consumed + if not data: + return b"" + + if self._pending: + data = self._pending + data + self._pending = b"" + + if data[:3] != b"ID3": + return data + + if len(data) < ID3_HEADER_SIZE: + self._pending = data + return b"" + + # ID3v2 stores the tag size as four synchsafe bytes (7 bits each). + size = data[6] << 21 | data[7] << 14 | data[8] << 7 | data[9] + body = data[ID3_HEADER_SIZE:] + consumed = min(size, len(body)) + self._skip = size - consumed + return body[consumed:] + + +class TTS(tts.TTS): + """Telnyx streaming Text-to-Speech. + + Opens one WebSocket per synthesis, streams MP3 slices back, and decodes + them into ``PcmData`` as they arrive. + + Examples: + + from vision_agents.plugins import telnyx + tts = telnyx.TTS(voice="AWS.Polly.Danielle-Neural") + """ + + def __init__( + self, + api_key: Optional[str] = None, + voice: str = DEFAULT_VOICE, + idle_timeout: float = 10.0, + ) -> None: + """Initialize Telnyx TTS. + + Args: + api_key: Telnyx API key. Falls back to the ``TELNYX_API_KEY`` env var. + voice: Voice id as listed by ``GET /v2/text-to-speech/voices``, + for example ``Telnyx.KokoroTTS.af_heart`` or + ``AWS.Polly.Danielle-Neural``. + idle_timeout: Seconds of server silence before synthesis is treated + as finished. Normally the server marks the last frame with + ``isFinal``; this is a safety net. + """ + super().__init__(provider_name="telnyx") + + self._api_key = api_key or os.environ.get("TELNYX_API_KEY") + if not self._api_key: + raise ValueError( + "TELNYX_API_KEY env var or api_key parameter required for Telnyx TTS" + ) + + self.voice = voice + self._idle_timeout = idle_timeout + + self._session: Optional[aiohttp.ClientSession] = None + self._ws: Optional[aiohttp.ClientWebSocketResponse] = None + self._lock = asyncio.Lock() + self._stop_event = asyncio.Event() + + async def close(self) -> None: + """Close the current WebSocket and release the aiohttp session.""" + await super().close() + if self._session is not None and not self._session.closed: + await self._session.close() + self._session = None + self._on_disconnected() + + async def stream_audio( + self, text: str, *_: Any, **__: Any + ) -> AsyncIterator[PcmData]: + """Stream TTS audio chunks for ``text``. + + Returns: + Async iterator yielding ``PcmData`` chunks. + """ + + async def _stream() -> AsyncIterator[PcmData]: + self._stop_event.clear() + async with self._lock: + ws = await self._connect() + try: + # Telnyx rejects a text frame that is not preceded by an + # init frame with "Invalid message". + await ws.send_str(json.dumps({"text": " "})) + await ws.send_str(json.dumps({"text": text})) + await ws.send_str(json.dumps({"text": ""})) + async for chunk in self._receive_audio(ws): + yield chunk + finally: + await self._close_ws() + + return _stream() + + async def stop_audio(self) -> None: + """Cancel any in-flight synthesis and drop the connection.""" + self._stop_event.set() + await self._close_ws() + + async def _connect(self) -> aiohttp.ClientWebSocketResponse: + if self._session is None or self._session.closed: + self._session = aiohttp.ClientSession() + + url = f"{WS_TTS_URL}?{urlencode({'voice': self.voice})}" + ws = await self._session.ws_connect( + url, headers={"Authorization": f"Bearer {self._api_key}"} + ) + self._ws = ws + self._on_connected() + logger.debug("Telnyx TTS websocket connected for voice %s", self.voice) + return ws + + async def _close_ws(self) -> None: + if self._ws is not None and not self._ws.closed: + try: + await self._ws.close() + except (aiohttp.ClientError, ConnectionError): + logger.debug("Error closing Telnyx TTS websocket") + self._ws = None + + async def _receive_audio( + self, ws: aiohttp.ClientWebSocketResponse + ) -> AsyncIterator[PcmData]: + """Yield PcmData until the final frame, a stop, an idle timeout, or a close.""" + decoder = cast(av.AudioCodecContext, av.CodecContext.create("mp3", "r")) + resampler = av.AudioResampler(format="s16", layout="mono") + stripper = _Id3Stripper() + + while True: + if self._stop_event.is_set(): + break + try: + msg = await asyncio.wait_for(ws.receive(), timeout=self._idle_timeout) + except asyncio.TimeoutError: + logger.debug("Telnyx TTS idle timeout, ending synthesis") + break + + if msg.type in ( + aiohttp.WSMsgType.CLOSED, + aiohttp.WSMsgType.CLOSING, + aiohttp.WSMsgType.ERROR, + ): + break + if msg.type != aiohttp.WSMsgType.TEXT: + continue + + try: + data = json.loads(msg.data) + except json.JSONDecodeError: + logger.warning("Telnyx TTS sent non-JSON text: %s", msg.data) + continue + + if data.get("error"): + raise TelnyxTTSError(str(data["error"])) + + encoded = data.get("audio") + if encoded: + for pcm in self._decode( + base64.b64decode(encoded), decoder, resampler, stripper + ): + yield pcm + + if data.get("isFinal"): + break + + def _decode( + self, + audio: bytes, + decoder: av.AudioCodecContext, + resampler: av.AudioResampler, + stripper: _Id3Stripper, + ) -> list[PcmData]: + """Decode one WebSocket audio payload into PcmData chunks.""" + chunks: list[PcmData] = [] + for packet in decoder.parse(stripper.feed(audio)): + for frame in decoder.decode(packet): + for resampled in resampler.resample(frame): + chunks.append( + PcmData( + samples=resampled.to_ndarray().reshape(-1), + sample_rate=resampled.sample_rate, + channels=1, + format=AudioFormat.S16, + ) + ) + return chunks diff --git a/uv.lock b/uv.lock index d0db9afe3..92743d527 100644 --- a/uv.lock +++ b/uv.lock @@ -7535,6 +7535,7 @@ dev = [ name = "vision-agents-plugins-telnyx" source = { editable = "plugins/telnyx" } dependencies = [ + { name = "aiohttp" }, { name = "cryptography" }, { name = "fastapi" }, { name = "numpy" }, @@ -7549,6 +7550,7 @@ dev = [ [package.metadata] requires-dist = [ + { name = "aiohttp", specifier = ">=3.13.3" }, { name = "cryptography", specifier = ">=44.0.0" }, { name = "fastapi", specifier = ">=0.135.1" }, { name = "numpy", specifier = ">=1.24.0" }, From 6511663c8a75f74d79e300ae274d2352013d0622 Mon Sep 17 00:00:00 2001 From: a692570 Date: Wed, 29 Jul 2026 21:30:42 -0700 Subject: [PATCH 2/3] fix(telnyx): guard TTS stop race, handshake timeout, and malformed audio --- plugins/telnyx/tests/test_telnyx_tts.py | 46 +++++++++++++++ .../vision_agents/plugins/telnyx/tts.py | 59 +++++++++++++------ 2 files changed, 87 insertions(+), 18 deletions(-) diff --git a/plugins/telnyx/tests/test_telnyx_tts.py b/plugins/telnyx/tests/test_telnyx_tts.py index 29cd20476..3f215c643 100644 --- a/plugins/telnyx/tests/test_telnyx_tts.py +++ b/plugins/telnyx/tests/test_telnyx_tts.py @@ -1,7 +1,10 @@ """Tests for the Telnyx TTS plugin.""" import os +from types import SimpleNamespace +import aiohttp +import av import pytest from dotenv import load_dotenv from vision_agents.plugins.telnyx import TTS @@ -85,6 +88,49 @@ async def test_custom_voice(self): assert tts.voice == "AWS.Polly.Danielle-Neural" +class TestTelnyxTTSMalformedPayloads: + """The receive loop tolerates junk from the server without aborting.""" + + @staticmethod + def fake_ws(payloads: list[str]) -> object: + messages = [ + SimpleNamespace(type=aiohttp.WSMsgType.TEXT, data=payload) + for payload in payloads + ] + [SimpleNamespace(type=aiohttp.WSMsgType.CLOSED, data=None)] + + class FakeWS: + def __init__(self) -> None: + self._queue = list(messages) + + async def receive(self): + return self._queue.pop(0) + + return FakeWS() + + async def test_non_dict_payload_is_skipped(self): + tts = TTS(api_key="KEY_test") + ws = self.fake_ws(['["not", "a", "dict"]']) + + assert [chunk async for chunk in tts._receive_audio(ws)] == [] + + async def test_invalid_base64_audio_is_skipped(self): + tts = TTS(api_key="KEY_test") + ws = self.fake_ws(['{"audio": "!!!not base64!!!"}']) + + assert [chunk async for chunk in tts._receive_audio(ws)] == [] + + async def test_undecodable_audio_is_dropped(self): + tts = TTS(api_key="KEY_test") + + class FailingDecoder: + def parse(self, data: bytes): + raise av.InvalidDataError(1094995529, "Invalid data") + + assert ( + tts._decode(b"\xff\xf3junk", FailingDecoder(), None, _Id3Stripper()) == [] + ) + + @pytest.mark.skipif(not os.getenv("TELNYX_API_KEY"), reason="TELNYX_API_KEY not set") @pytest.mark.integration class TestTelnyxTTSIntegration: diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/tts.py b/plugins/telnyx/vision_agents/plugins/telnyx/tts.py index a83424d61..1d73cf337 100644 --- a/plugins/telnyx/vision_agents/plugins/telnyx/tts.py +++ b/plugins/telnyx/vision_agents/plugins/telnyx/tts.py @@ -20,6 +20,7 @@ import asyncio import base64 +import binascii import json import logging import os @@ -148,8 +149,11 @@ async def stream_audio( """ async def _stream() -> AsyncIterator[PcmData]: - self._stop_event.clear() async with self._lock: + # Cleared under the lock so a stop_audio() aimed at an + # in-flight synthesis cannot leave the event set for a call + # queued behind it. + self._stop_event.clear() ws = await self._connect() try: # Telnyx rejects a text frame that is not preceded by an @@ -174,8 +178,13 @@ async def _connect(self) -> aiohttp.ClientWebSocketResponse: self._session = aiohttp.ClientSession() url = f"{WS_TTS_URL}?{urlencode({'voice': self.voice})}" - ws = await self._session.ws_connect( - url, headers={"Authorization": f"Bearer {self._api_key}"} + # aiohttp's default client timeout for the handshake is 300s, far past + # the point where a caller waiting on speech should give up. + ws = await asyncio.wait_for( + self._session.ws_connect( + url, headers={"Authorization": f"Bearer {self._api_key}"} + ), + timeout=self._idle_timeout, ) self._ws = ws self._on_connected() @@ -187,7 +196,7 @@ async def _close_ws(self) -> None: try: await self._ws.close() except (aiohttp.ClientError, ConnectionError): - logger.debug("Error closing Telnyx TTS websocket") + logger.debug("Error closing Telnyx TTS websocket", exc_info=True) self._ws = None async def _receive_audio( @@ -222,14 +231,21 @@ async def _receive_audio( logger.warning("Telnyx TTS sent non-JSON text: %s", msg.data) continue + if not isinstance(data, dict): + logger.warning("Telnyx TTS sent unexpected payload: %r", data) + continue + if data.get("error"): raise TelnyxTTSError(str(data["error"])) encoded = data.get("audio") if encoded: - for pcm in self._decode( - base64.b64decode(encoded), decoder, resampler, stripper - ): + try: + raw = base64.b64decode(encoded) + except (binascii.Error, ValueError): + logger.warning("Telnyx TTS sent audio that is not valid base64") + continue + for pcm in self._decode(raw, decoder, resampler, stripper): yield pcm if data.get("isFinal"): @@ -242,17 +258,24 @@ def _decode( resampler: av.AudioResampler, stripper: _Id3Stripper, ) -> list[PcmData]: - """Decode one WebSocket audio payload into PcmData chunks.""" + """Decode one WebSocket audio payload into PcmData chunks. + + A corrupt payload is dropped rather than allowed to abort the + synthesis; the decoder recovers on the next packet boundary. + """ chunks: list[PcmData] = [] - for packet in decoder.parse(stripper.feed(audio)): - for frame in decoder.decode(packet): - for resampled in resampler.resample(frame): - chunks.append( - PcmData( - samples=resampled.to_ndarray().reshape(-1), - sample_rate=resampled.sample_rate, - channels=1, - format=AudioFormat.S16, + try: + for packet in decoder.parse(stripper.feed(audio)): + for frame in decoder.decode(packet): + for resampled in resampler.resample(frame): + chunks.append( + PcmData( + samples=resampled.to_ndarray().reshape(-1), + sample_rate=resampled.sample_rate, + channels=1, + format=AudioFormat.S16, + ) ) - ) + except av.FFmpegError: + logger.warning("Telnyx TTS sent undecodable audio, dropping payload") return chunks From 1510363d8af8c06251aead311213c39c823ad85c Mon Sep 17 00:00:00 2001 From: a692570 Date: Thu, 30 Jul 2026 00:16:16 -0700 Subject: [PATCH 3/3] fix(telnyx): treat barge-in socket close as normal TTS end and validate base64 --- plugins/telnyx/tests/test_telnyx_tts.py | 50 +++++++++++++++++++ .../vision_agents/plugins/telnyx/tts.py | 11 ++-- 2 files changed, 58 insertions(+), 3 deletions(-) diff --git a/plugins/telnyx/tests/test_telnyx_tts.py b/plugins/telnyx/tests/test_telnyx_tts.py index 3f215c643..f51278aaf 100644 --- a/plugins/telnyx/tests/test_telnyx_tts.py +++ b/plugins/telnyx/tests/test_telnyx_tts.py @@ -88,6 +88,50 @@ async def test_custom_voice(self): assert tts.voice == "AWS.Polly.Danielle-Neural" +class TestTelnyxTTSBargeIn: + """A socket dropped by stop_audio() ends synthesis instead of raising.""" + + @staticmethod + def tts_with_dropped_socket(stop_before_drop: bool) -> TTS: + instance = TTS(api_key="KEY_test") + + class DroppedWS: + closed = True + + async def send_str(self, data: str) -> None: + if stop_before_drop: + instance._stop_event.set() + raise aiohttp.ClientConnectionResetError("Cannot write to closing") + + async def close(self) -> None: + return None + + class FakeSession: + closed = False + + async def ws_connect(self, url: str, headers: dict[str, str]): + return DroppedWS() + + instance._session = FakeSession() + return instance + + async def test_stop_during_synthesis_ends_quietly(self): + """A socket closed by a concurrent stop_audio() is a barge-in.""" + tts = self.tts_with_dropped_socket(stop_before_drop=True) + + stream = await tts.stream_audio("hello") + assert [chunk async for chunk in stream] == [] + + async def test_connection_drop_without_stop_propagates(self): + """A stale stop must not silence a genuine failure in a new synthesis.""" + tts = self.tts_with_dropped_socket(stop_before_drop=False) + await tts.stop_audio() + + stream = await tts.stream_audio("hello") + with pytest.raises(aiohttp.ClientConnectionError): + [chunk async for chunk in stream] + + class TestTelnyxTTSMalformedPayloads: """The receive loop tolerates junk from the server without aborting.""" @@ -119,6 +163,12 @@ async def test_invalid_base64_audio_is_skipped(self): assert [chunk async for chunk in tts._receive_audio(ws)] == [] + async def test_non_string_audio_is_skipped(self): + tts = TTS(api_key="KEY_test") + ws = self.fake_ws(['{"audio": 12345}']) + + assert [chunk async for chunk in tts._receive_audio(ws)] == [] + async def test_undecodable_audio_is_dropped(self): tts = TTS(api_key="KEY_test") diff --git a/plugins/telnyx/vision_agents/plugins/telnyx/tts.py b/plugins/telnyx/vision_agents/plugins/telnyx/tts.py index 1d73cf337..cc28ebfdd 100644 --- a/plugins/telnyx/vision_agents/plugins/telnyx/tts.py +++ b/plugins/telnyx/vision_agents/plugins/telnyx/tts.py @@ -154,8 +154,8 @@ async def _stream() -> AsyncIterator[PcmData]: # in-flight synthesis cannot leave the event set for a call # queued behind it. self._stop_event.clear() - ws = await self._connect() try: + ws = await self._connect() # Telnyx rejects a text frame that is not preceded by an # init frame with "Invalid message". await ws.send_str(json.dumps({"text": " "})) @@ -163,6 +163,11 @@ async def _stream() -> AsyncIterator[PcmData]: await ws.send_str(json.dumps({"text": ""})) async for chunk in self._receive_audio(ws): yield chunk + except aiohttp.ClientConnectionError: + # stop_audio() closes the socket underneath us, which is a + # normal barge-in rather than a synthesis failure. + if not self._stop_event.is_set(): + raise finally: await self._close_ws() @@ -241,8 +246,8 @@ async def _receive_audio( encoded = data.get("audio") if encoded: try: - raw = base64.b64decode(encoded) - except (binascii.Error, ValueError): + raw = base64.b64decode(encoded, validate=True) + except (binascii.Error, TypeError, ValueError): logger.warning("Telnyx TTS sent audio that is not valid base64") continue for pcm in self._decode(raw, decoder, resampler, stripper):