From 70e1815860e6073e3ef31acfeea98752723e7d11 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Wed, 8 Apr 2026 11:23:10 +0000 Subject: [PATCH 1/5] docs: improve TTS extension guide with WebSocket patterns, config passthrough, graph wiring add complete serial WebSocket client and extension examples based on production deepgram_tts implementation. add critical patterns section covering finish_request(), _finalize_request(), TTFB measurement, invalid text handling, cancel/reconnect, and audio_start ordering. add params passthrough pattern showing update_params() convention used by all TTS extensions. add voice-assistant graph wiring section. fix config examples to use Field(default_factory=dict) and dict[str, Any]. replace fictional test config filenames with real property_*.json layout. add InvalidStatus handling to WebSocket _connect() example. fix graph schema to use "graph": {"nodes": ..., "connections": ...}. --- .../extension_dev/create_tts_extension.mdx | 592 +++++++++++++++++- 1 file changed, 580 insertions(+), 12 deletions(-) diff --git a/content/docs/ten_agent_examples/extension_dev/create_tts_extension.mdx b/content/docs/ten_agent_examples/extension_dev/create_tts_extension.mdx index 87a801f..a518a85 100644 --- a/content/docs/ten_agent_examples/extension_dev/create_tts_extension.mdx +++ b/content/docs/ten_agent_examples/extension_dev/create_tts_extension.mdx @@ -525,6 +525,427 @@ def _clear_queues(self) -> None: 6. **Set timeouts**: Avoid long blocking periods. 7. **Log properly**: Include vendor errors, state changes, request flow, and response flow. See [Logging Specifications](#logging-specifications). +### Simple Serial WebSocket Pattern + +Not all WebSocket TTS vendors need the full duplex send/receive loop shown above. Many vendors (such as Deepgram) support a simpler serial pattern: send text, receive audio chunks, repeat. This pattern is easier to implement and debug. Choose this path when the vendor API processes one utterance at a time over a persistent connection. + +#### Complete Client Example + +```python title="{vendor}_tts.py" +import asyncio +import json +from datetime import datetime +from typing import AsyncIterator +from urllib.parse import urlencode + +import websockets +from websockets.asyncio.client import ClientConnection +from websockets.exceptions import InvalidStatus + +from .config import VendorTTSConfig + +# Event types yielded back to the extension +EVENT_TTS_RESPONSE = 1 # Audio bytes +EVENT_TTS_END = 2 # Vendor finished this utterance +EVENT_TTS_ERROR = 3 # Vendor error +EVENT_TTS_TTFB_METRIC = 5 # Time-to-first-byte measurement + +WS_RECV_TIMEOUT = 8.0 # Seconds before giving up on a response + + +class VendorTTSClient: + """Serial WebSocket TTS client. + + Each get() call sends a text request and streams audio until done. + The connection is reused across calls and reconnected when needed. + """ + + def __init__(self, config: VendorTTSConfig, ten_env): + self.config = config + self.ten_env = ten_env + self._ws: ClientConnection | None = None + self._is_cancelled = False + self._needs_reconnect = False + self._sent_ts: datetime | None = None + self._ttfb_sent: bool = False + self._ws_url = self._build_ws_url() + + def _build_ws_url(self) -> str: + """Build the WebSocket URL with query parameters. + + Known config keys become fixed query params. Any extra keys + in config.params are forwarded as-is — see the params + passthrough pattern in Configuration Management Design. + """ + query_params: dict[str, str | int | float | bool] = { + "model": self.config.model, + "encoding": self.config.encoding, + "sample_rate": self.config.sample_rate, + } + for key, value in self.config.params.items(): + if key not in {"api_key", "base_url"} and value is not None: + query_params[key] = value + return f"{self.config.base_url}?{urlencode(query_params)}" + + async def start(self) -> None: + """Preheat: establish initial connection during on_init.""" + await self._connect() + + async def stop(self) -> None: + """Tear down the connection during on_stop.""" + if self._ws: + try: + await self._ws.close() + except Exception: + pass + self._ws = None + + async def cancel(self) -> None: + """Cancel current TTS. Sends Flush and drains until Flushed + so the connection is left in a clean state for the next request.""" + self._is_cancelled = True + if self._ws: + try: + await self._ws.send(json.dumps({"type": "Flush"})) + await asyncio.wait_for(self._drain_until_flushed(), timeout=3.0) + except Exception: + self._needs_reconnect = True + + def reset_ttfb(self) -> None: + """Reset TTFB tracking for a new request.""" + self._sent_ts = None + self._ttfb_sent = False + + async def get(self, text: str) -> AsyncIterator[tuple[bytes | int | None, int]]: + """Send text and yield (data, event_type) tuples. + + Callers iterate this to receive audio chunks, TTFB metrics, + end-of-stream, or error events. + """ + if not text.strip(): + yield None, EVENT_TTS_END + return + + if self._needs_reconnect: + await self._reconnect() + await self._ensure_connection() + + if not self._ttfb_sent: + self._sent_ts = datetime.now() + + # Clear cancel flag just before sending — avoids a race + # where a concurrent cancel() call gets overwritten + self._is_cancelled = False + + # Send Speak + Flush to trigger audio generation + await self._ws.send(json.dumps({"type": "Speak", "text": text})) + await self._ws.send(json.dumps({"type": "Flush"})) + + # Stream audio until Flushed or error + try: + while not self._is_cancelled: + try: + message = await asyncio.wait_for( + self._ws.recv(), timeout=WS_RECV_TIMEOUT + ) + except asyncio.TimeoutError: + self._needs_reconnect = True + yield b"Timeout waiting for audio", EVENT_TTS_ERROR + break + + if isinstance(message, bytes): + if self._is_cancelled: + break + # TTFB on first audio chunk + if self._sent_ts and not self._ttfb_sent: + ttfb_ms = int( + (datetime.now() - self._sent_ts).total_seconds() * 1000 + ) + yield ttfb_ms, EVENT_TTS_TTFB_METRIC + self._ttfb_sent = True + yield message, EVENT_TTS_RESPONSE + else: + data = json.loads(message) + msg_type = data.get("type", "") + if msg_type == "Flushed": + yield None, EVENT_TTS_END + break + elif msg_type == "Error": + self._needs_reconnect = True + yield data.get("err_msg", "").encode(), EVENT_TTS_ERROR + break + except Exception as e: + self._needs_reconnect = True + yield str(e).encode(), EVENT_TTS_ERROR + + async def _connect(self) -> None: + headers = {"Authorization": f"Token {self.config.api_key}"} + try: + self._ws = await websockets.connect( + self._ws_url, additional_headers=headers + ) + except InvalidStatus as e: + # Surface auth / HTTP status failures explicitly so the + # extension can map them to the correct TEN error code. + status_code = e.response.status_code + raise RuntimeError( + f"WebSocket connection failed with status {status_code}" + ) from e + + async def _ensure_connection(self) -> None: + if not self._ws: + await self._connect() + + async def _reconnect(self) -> None: + await self.stop() + await self._connect() + self._needs_reconnect = False + + async def _drain_until_flushed(self) -> None: + """Read and discard messages until Flushed.""" + while self._ws: + msg = await self._ws.recv() + if isinstance(msg, str): + if json.loads(msg).get("type") == "Flushed": + return +``` + +#### Complete Extension Example + +The extension below is a complete, working implementation that handles the full lifecycle: initialization, request processing, TTFB metrics, cancellation, error handling, and cleanup. + +**Key patterns to note:** +- **`_finalize_request()`** consolidates all cleanup into one place (audio end, finish request, state reset) +- **`finish_request()`** is called to signal the base class queue that the current request is done — **forgetting this will hang the queue** +- **`_audio_start_sent`** ensures `tts_audio_start` is always sent before `tts_audio_end`, even on errors + +```python title="extension.py" +from datetime import datetime +from ten_ai_base.tts2 import AsyncTTS2BaseExtension +from ten_ai_base.message import ( + ModuleError, ModuleErrorCode, ModuleType, + ModuleErrorVendorInfo, TTSAudioEndReason, +) +from ten_ai_base.struct import TTSTextInput +from ten_ai_base.const import LOG_CATEGORY_VENDOR, LOG_CATEGORY_KEY_POINT +from .config import VendorTTSConfig +from .vendor_tts import ( + VendorTTSClient, EVENT_TTS_RESPONSE, EVENT_TTS_END, + EVENT_TTS_TTFB_METRIC, EVENT_TTS_ERROR, +) + + +class VendorTTSExtension(AsyncTTS2BaseExtension): + def __init__(self, name: str) -> None: + super().__init__(name) + self.config: VendorTTSConfig | None = None + self.client: VendorTTSClient | None = None + self.current_request_id: str | None = None + self.current_request_finished: bool = False + self.total_audio_bytes: int = 0 + self.sent_ts: datetime | None = None + self._audio_start_sent: bool = False + + async def on_init(self, ten_env) -> None: + await super().on_init(ten_env) + config_json, _ = await self.ten_env.get_property_to_json("") + self.config = VendorTTSConfig.model_validate_json(config_json) + self.config.update_params() + ten_env.log_info( + self.config.to_str(sensitive_handling=True), + category=LOG_CATEGORY_KEY_POINT, + ) + # Await start() directly — do not use asyncio.create_task() + # so preheat failures surface during initialization + self.client = VendorTTSClient(config=self.config, ten_env=ten_env) + await self.client.start() + + async def on_stop(self, ten_env) -> None: + if self.client: + await self.client.stop() + await super().on_stop(ten_env) + + def vendor(self) -> str: + return "vendor_name" + + def synthesize_audio_sample_rate(self) -> int: + return self.config.sample_rate if self.config else 24000 + + async def cancel_tts(self) -> None: + """Called by the base class on flush. Cancel in-flight audio + and finalize the request as INTERRUPTED.""" + self.current_request_finished = True + if self.current_request_id and self.client: + await self.client.cancel() + await self._finalize_request(TTSAudioEndReason.INTERRUPTED) + + async def request_tts(self, t: TTSTextInput) -> None: + """Called by the base class for each queued text input.""" + # --- New request setup --- + if t.request_id != self.current_request_id: + if self.client: + self.client.reset_ttfb() + self.current_request_id = t.request_id + self.current_request_finished = False + self.total_audio_bytes = 0 + self.sent_ts = None + self._audio_start_sent = False + + if t.text_input_end: + self.current_request_finished = True + + text = t.text.strip() + if not text: + # Empty text with text_input_end: finalize immediately + if t.text_input_end: + await self._finalize_request(TTSAudioEndReason.REQUEST_END) + return + + # --- Stream audio from vendor --- + self.ten_env.log_debug( + f"send_text_to_tts_server: {text} of request_id: {t.request_id}", + category=LOG_CATEGORY_VENDOR, + ) + if self.sent_ts is None: + self.sent_ts = datetime.now() + + async for data, event in self.client.get(text): + if event == EVENT_TTS_RESPONSE and isinstance(data, bytes): + self.total_audio_bytes += len(data) + await self.send_tts_audio_data(data) + + elif event == EVENT_TTS_TTFB_METRIC and isinstance(data, int): + self.sent_ts = datetime.now() + await self.send_tts_audio_start( + request_id=self.current_request_id, + ) + self._audio_start_sent = True + await self.send_tts_ttfb_metrics( + request_id=self.current_request_id, + ttfb_ms=data, + extra_metadata={"model": self.config.model}, + ) + + elif event == EVENT_TTS_END: + if t.text_input_end: + await self._finalize_request(TTSAudioEndReason.REQUEST_END) + break + + elif event == EVENT_TTS_ERROR: + error = ModuleError( + message=str(data), + module=ModuleType.TTS, + code=ModuleErrorCode.NON_FATAL_ERROR, + vendor_info=ModuleErrorVendorInfo(vendor=self.vendor()), + ) + if t.text_input_end: + await self._finalize_request( + TTSAudioEndReason.ERROR, error=error + ) + break + + async def _finalize_request( + self, reason: TTSAudioEndReason, error: ModuleError | None = None + ) -> None: + """Consolidate all request cleanup into one place. + + This pattern avoids duplicate finalization code scattered + across multiple branches. Every exit path calls this method. + """ + # Ensure audio_start was sent (protocol requires it before audio_end) + if not self._audio_start_sent: + await self.send_tts_audio_start( + request_id=self.current_request_id, + ) + self._audio_start_sent = True + + interval_ms = self._current_request_interval_ms() + duration_ms = self._calculate_audio_duration_ms() + + await self.send_tts_audio_end( + request_id=self.current_request_id, + request_event_interval_ms=interval_ms, + request_total_audio_duration_ms=duration_ms, + reason=reason, + ) + + # CRITICAL: Signal the base class queue that this request is done. + # Forgetting this call will hang the queue permanently. + await self.finish_request( + request_id=self.current_request_id, + reason=reason, + error=error, + ) + self.sent_ts = None + + def _current_request_interval_ms(self) -> int: + if not self.sent_ts: + return 0 + return int((datetime.now() - self.sent_ts).total_seconds() * 1000) + + def _calculate_audio_duration_ms(self) -> int: + if not self.config: + return 0 + bytes_per_sample = 2 # 16-bit PCM + return int( + self.total_audio_bytes + / (self.synthesize_audio_sample_rate() * bytes_per_sample) + * 1000 + ) +``` + +### Critical WebSocket Implementation Patterns + +These patterns are essential for passing all guarder tests. They apply to both the duplex and serial WebSocket patterns. + +#### 1. Always Call `finish_request()` + +When inheriting from `AsyncTTS2BaseExtension` directly (WebSocket mode), **you must call `finish_request()`** to signal the base class queue that the current request is complete. The HTTP base class does this automatically, but WebSocket subclasses must do it explicitly. Forgetting this call will hang the processing queue. + +```python +await self.finish_request( + request_id=self.current_request_id, + reason=reason, + error=error, +) +``` + +#### 2. Consolidate Cleanup with `_finalize_request()` + +Without a helper, the same cleanup code (send audio_end, flush recorder, call finish_request, reset state) gets duplicated across every exit path: normal completion, cancellation, errors, empty text. Extract it into a single `_finalize_request()` method as shown in the extension example above. + +#### 3. TTFB Measurement + +For WebSocket mode, you measure TTFB yourself: +- Record `sent_ts = datetime.now()` when sending text to the vendor +- On the first audio chunk, compute `ttfb_ms = (now - sent_ts) * 1000` +- Yield `EVENT_TTS_TTFB_METRIC` from the client, then call `send_tts_ttfb_metrics()` in the extension +- Call `send_tts_audio_start()` at TTFB time (before first audio chunk) + +#### 4. Handle Invalid Text Input + +The guarder `test_invalid_text_handling` sends empty strings, emoji-only, punctuation-only, and whitespace-only text. Your client must handle these gracefully: + +```python +# In the client's get() method — return immediately for empty text +if not text.strip(): + yield None, EVENT_TTS_END + return +``` + +For text that the vendor silently ignores (producing no audio), set a reasonable `WS_RECV_TIMEOUT` (5-8 seconds) so the extension times out and reports an error rather than hanging indefinitely. + +#### 5. Cancel and Reconnect Strategy + +- **Cancel**: Send a Flush command and drain responses until the vendor acknowledges it (e.g., `Flushed`). This leaves the connection in a clean state for the next request. +- **Reconnect flag**: Set `_needs_reconnect = True` on errors and timeouts. Check it at the start of `get()` and reconnect before sending. +- **Eager reconnect**: After connection errors, reconnect immediately rather than waiting for the next request. This reduces latency on the next call. +- **Typed auth failures**: Distinguish authentication / HTTP status failures (for example 401) from generic transport errors so invalid-credential tests map to the correct TEN fatal error behavior. + +#### 6. Ensure `tts_audio_start` Before `tts_audio_end` + +The TEN protocol requires `tts_audio_start` before `tts_audio_end`. Track whether it was sent with a flag (`_audio_start_sent`), and send it in `_finalize_request()` if it was not sent yet (e.g., on early errors). + ## Extension Structure and Supporting Files The overall directory structure can be referenced from [Project Structure Overview](#project-structure-overview). @@ -550,10 +971,20 @@ tests/ ├── test_params.py # Parameter config tests, mainly validating parameter checks ├── test_robustness.py # Robustness tests for exceptional situations ├── test_metrics.py # Metrics tests +├── test_state_machine.py # Sequential/interleaved request lifecycle tests └── configs/ # Test config files - ├── test_config.json # Test config - ├── invalid_config.json # Invalid config test - └── mock_config.json # Mock config + ├── property.json # Default valid config + ├── property_basic_audio_setting1.json + ├── property_basic_audio_setting2.json + ├── property_dump.json + ├── property_invalid.json + └── property_miss_required.json +``` + +If your vendor supports subtitle or word-timestamp alignment, also add: + +```text +tests/configs/property_subtitle_alignment.json ``` ### Unit Test Best Practices @@ -841,6 +1272,86 @@ Guarder integration tests include the following checks to ensure the extension c 6. **Respect test order when needed**: Some tests may have dependencies. 7. **Clean up resources**: Remove temporary files and release resources after testing. +## 🔌 Wiring into a Voice Assistant Graph + +After unit tests and guarder tests pass, wire your extension into a real agent graph to validate it end-to-end. + +### 1. Add the Dependency + +In the voice-assistant example's `manifest.json`, add your extension as a local dependency: + +```json title="examples/voice-assistant/tenapp/manifest.json" +{ + "dependencies": [ + ...existing dependencies..., + { + "type": "extension", + "name": "my_tts_extension", + "version": "0.1.0", + "path": "../../../ten_packages/extension/my_tts_extension" + } + ] +} +``` + +### 2. Add a Graph Variant + +In the voice-assistant `property.json`, add a new graph that uses your TTS extension. Copy an existing graph (e.g., `voice_assistant`) and change the TTS extension name: + +```json title="examples/voice-assistant/tenapp/property.json (excerpt)" +{ + "name": "voice_assistant_my_tts", + "auto_start": false, + "graph": { + "nodes": [ + { + "type": "extension", + "name": "tts", + "addon": "my_tts_extension", + "extension_group": "tts", + "property": { + "params": { + "api_key": "${env:MY_TTS_API_KEY}", + "model": "default-model", + "encoding": "linear16", + "sample_rate": 24000 + } + } + } + // ...other nodes (agora_rtc, main, stt, llm, etc.) + ], + "connections": [ + // Copy the full connections block from a known-good + // voice_assistant graph and change only the TTS addon/node. + // Do not assume routing is implicit unless you have verified + // that exact graph implementation. + ] + } +} +``` + +For best results, copy a confirmed working graph variant and modify only: +- the TTS dependency in `manifest.json` +- the TTS node's `addon` +- the TTS node's `property.params` + +Avoid hand-reconstructing the connections block from memory. A working voice +pipeline must still route ASR results into the main controller, text from LLM +back through the main controller, TTS input into the TTS node, and PCM audio +back to `agora_rtc`. + +### 3. Test Live + +Start the agent with your new graph: +- In the playground, select `voice_assistant_my_tts` from the graph dropdown +- Or set `auto_start: true` on your graph and restart the agent + +Validate: +- Audio plays back clearly +- Multi-turn conversation works +- Interruptions (flush) work correctly +- TTFB feels responsive + ## 🌐 End-to-End Testing After development is complete, you can quickly replace the TTS node in a TEN Agent graph with TMan Designer to validate it in a real conversation scenario. @@ -1999,21 +2510,76 @@ If you have additional custom parameters that are not part of the vendor's offic Create a flexible config class that supports required parameters as well as optional passthrough parameters: ```python title="config.py" -from pydantic import BaseModel -from typing import Dict, Optional +from pydantic import BaseModel, Field +from typing import Any class MyTTSConfig(BaseModel): # All vendor parameters live in params, including required and optional ones - params: Dict[str, Optional[str]] = {} - - # Non-vendor parameters used only by this TTS extension - extra_params: Dict[str, Optional[str]] = {} + params: dict[str, Any] = Field(default_factory=dict) # Standard dump configuration shared by TTS extensions dump: bool = False - dump_path: Optional[str] = None + dump_path: str | None = None ``` +#### Params Passthrough Pattern + +Every TTS extension in the repo follows the same convention: `config.params` is a flat dict containing both well-known keys (like `api_key`, `model`) and arbitrary vendor-specific keys. The `update_params()` method extracts the well-known keys onto named fields and leaves the rest in `params` for forwarding to the vendor API. + +```python title="config.py" +from pydantic import BaseModel, Field +from typing import Any +import copy +from ten_ai_base import utils + + +class VendorTTSConfig(BaseModel): + api_key: str = "" + base_url: str = "wss://api.vendor.com/v1/speak" + model: str = "default-model" + encoding: str = "linear16" + sample_rate: int = 24000 + + dump: bool = False + dump_path: str = "/tmp" + params: dict[str, Any] = Field(default_factory=dict) + + def update_params(self) -> None: + """Extract well-known keys from params onto named fields. + + After this call, self.params contains only the extra + vendor-specific keys that should be forwarded to the API. + """ + params = self.params if isinstance(self.params, dict) else {} + self.params = params + + # Each known key is extracted and removed from params + for attr in ("api_key", "base_url", "model", "encoding", "sample_rate"): + if attr in params: + setattr(self, attr, params.pop(attr)) + + def to_str(self, sensitive_handling: bool = True) -> str: + if not sensitive_handling: + return f"{self}" + config = copy.deepcopy(self) + if config.api_key: + config.api_key = utils.encrypt(config.api_key) + return f"{config}" +``` + +The client then forwards remaining `params` to the vendor. For WebSocket APIs, this means query string parameters: + +```python +# In the client's _build_ws_url() +for key, value in self.config.params.items(): + if key not in {"api_key", "base_url"} and value is not None: + query_params[key] = value +``` + +For HTTP APIs, these would be merged into the request JSON body. For SDK-based vendors, they become keyword arguments. + +This pattern is consistent across all TTS extensions: Cartesia, ElevenLabs, Minimax, OpenAI, Cosy, Fish Audio, Bytedance, and Deepgram all use it. + #### Read Extension Config Load and initialize config during `on_init`: @@ -2061,12 +2627,14 @@ async def on_init(self, ten_env: AsyncTenEnv) -> None: Add a masking helper to the config class so sensitive information is protected in logs: ```python title="config.py" +from pydantic import BaseModel, Field +from typing import Any from ten_ai_base.utils import encrypt class MyTTSConfig(BaseModel): - params: Dict[str, Optional[str]] = {} + params: dict[str, Any] = Field(default_factory=dict) dump: bool = False - dump_path: Optional[str] = None + dump_path: str | None = None def to_json(self, sensitive_handling: bool = False) -> str: """ From e50d738cc8b05fdd93b6f6fd119b3ab21d40f80c Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Wed, 8 Apr 2026 11:25:24 +0000 Subject: [PATCH 2/5] fix: use typed connection exception and correct test config layout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit replace generic RuntimeError with VendorTTSConnectionException that preserves HTTP status code, enabling the extension to distinguish 401 (FATAL_ERROR) from transient failures (NON_FATAL_ERROR). add connection error handling in request_tts() showing the mapping. remove fictional property.json from test configs tree — the guarder tests use property_basic_audio_setting1.json as their default config. add inline descriptions showing which guarder tests use each file. --- .../extension_dev/create_tts_extension.mdx | 64 +++++++++++++++---- 1 file changed, 51 insertions(+), 13 deletions(-) diff --git a/content/docs/ten_agent_examples/extension_dev/create_tts_extension.mdx b/content/docs/ten_agent_examples/extension_dev/create_tts_extension.mdx index a518a85..fd24751 100644 --- a/content/docs/ten_agent_examples/extension_dev/create_tts_extension.mdx +++ b/content/docs/ten_agent_examples/extension_dev/create_tts_extension.mdx @@ -553,6 +553,18 @@ EVENT_TTS_TTFB_METRIC = 5 # Time-to-first-byte measurement WS_RECV_TIMEOUT = 8.0 # Seconds before giving up on a response +class VendorTTSConnectionException(Exception): + """Typed connection failure that preserves the HTTP status code. + + The extension uses status_code to distinguish fatal errors (e.g. 401) + from transient ones, mapping them to the correct TEN error code. + """ + def __init__(self, status_code: int, body: str): + self.status_code = status_code + self.body = body + super().__init__(f"Connection failed (code: {status_code}): {body}") + + class VendorTTSClient: """Serial WebSocket TTS client. @@ -685,11 +697,11 @@ class VendorTTSClient: self._ws_url, additional_headers=headers ) except InvalidStatus as e: - # Surface auth / HTTP status failures explicitly so the - # extension can map them to the correct TEN error code. - status_code = e.response.status_code - raise RuntimeError( - f"WebSocket connection failed with status {status_code}" + # Preserve the HTTP status code so the extension can + # distinguish 401 (FATAL_ERROR) from transient failures + raise VendorTTSConnectionException( + status_code=e.response.status_code, + body=str(e), ) from e async def _ensure_connection(self) -> None: @@ -730,7 +742,8 @@ from ten_ai_base.struct import TTSTextInput from ten_ai_base.const import LOG_CATEGORY_VENDOR, LOG_CATEGORY_KEY_POINT from .config import VendorTTSConfig from .vendor_tts import ( - VendorTTSClient, EVENT_TTS_RESPONSE, EVENT_TTS_END, + VendorTTSClient, VendorTTSConnectionException, + EVENT_TTS_RESPONSE, EVENT_TTS_END, EVENT_TTS_TTFB_METRIC, EVENT_TTS_ERROR, ) @@ -781,6 +794,32 @@ class VendorTTSExtension(AsyncTTS2BaseExtension): async def request_tts(self, t: TTSTextInput) -> None: """Called by the base class for each queued text input.""" + try: + await self._do_request_tts(t) + except VendorTTSConnectionException as e: + # Map HTTP status to TEN error code: + # 401 → FATAL_ERROR (bad credentials, won't recover) + # anything else → NON_FATAL_ERROR (transient) + code = ( + ModuleErrorCode.FATAL_ERROR + if e.status_code == 401 + else ModuleErrorCode.NON_FATAL_ERROR + ) + error = ModuleError( + message=e.body, + module=ModuleType.TTS, + code=code, + vendor_info=ModuleErrorVendorInfo( + vendor=self.vendor(), + code=str(e.status_code), + message=e.body, + ), + ) + await self._finalize_request(TTSAudioEndReason.ERROR, error=error) + + async def _do_request_tts(self, t: TTSTextInput) -> None: + """Core request logic, separated so connection errors + can be caught and mapped in request_tts().""" # --- New request setup --- if t.request_id != self.current_request_id: if self.client: @@ -972,13 +1011,12 @@ tests/ ├── test_robustness.py # Robustness tests for exceptional situations ├── test_metrics.py # Metrics tests ├── test_state_machine.py # Sequential/interleaved request lifecycle tests -└── configs/ # Test config files - ├── property.json # Default valid config - ├── property_basic_audio_setting1.json - ├── property_basic_audio_setting2.json - ├── property_dump.json - ├── property_invalid.json - └── property_miss_required.json +└── configs/ # Guarder test config files + ├── property_basic_audio_setting1.json # Basic audio + boundary + metrics + invalid text tests + ├── property_basic_audio_setting2.json # Second sample-rate variant for comparison + ├── property_dump.json # Dump, flush, and per-request-ID dump tests + ├── property_invalid.json # Invalid required params (e.g. bad API key) + └── property_miss_required.json # Missing required params (e.g. empty API key) ``` If your vendor supports subtitle or word-timestamp alignment, also add: From f8be898e624af080c14ef5b86f8bcc7bbaea85ca Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Fri, 24 Apr 2026 08:49:19 +0000 Subject: [PATCH 3/5] docs(extension): tighten asr and tts guides --- .../extension_dev/create_asr_extension.mdx | 41 ++++++++++++ .../extension_dev/create_tts_extension.mdx | 62 +++++++++++++++++++ 2 files changed, 103 insertions(+) diff --git a/content/docs/ten_agent_examples/extension_dev/create_asr_extension.mdx b/content/docs/ten_agent_examples/extension_dev/create_asr_extension.mdx index 6059a71..9a832af 100644 --- a/content/docs/ten_agent_examples/extension_dev/create_asr_extension.mdx +++ b/content/docs/ten_agent_examples/extension_dev/create_asr_extension.mdx @@ -21,6 +21,20 @@ Choose the appropriate section to read based on your needs. - Have the `tman` command-line tool installed and be familiar with its basic usage. - Have an API key from an ASR vendor ready (for testing). +## Definition of Done + +Before calling a new ASR extension complete, verify all of these: + +- Vendor wire contract is confirmed: endpoint, event names, payload encoding, finalize behavior. +- Correct base class is used and all required methods are implemented. +- Config is validated and secrets are redacted in logs. +- `config:` log output is present. +- Fatal vs non-fatal errors are classified intentionally. +- Reconnect is actually wired if reconnect support is claimed. +- Standalone tests are present and green. +- Guarder tests are green. +- README and example configuration are accurate. + ## Table of Contents ### Part 1: Basic - Implement Basic Functionality @@ -179,6 +193,21 @@ You only need to focus on integrating with the specific ASR vendor. ## 4. Implement Core Functionality +### 4.0 Verify the Vendor Wire Contract First + +Before writing code, confirm the provider contract you will implement: + +- Exact WebSocket or HTTP endpoint +- Startup or ready event name +- Partial and final event names +- Whether audio is sent as raw binary or encoded JSON +- Whether output payloads are raw text, JSON, base64, or another format +- How the provider signals end-of-input or finalize +- Whether the connection stays open after finalize or closes per utterance + +Write the confirmed message schema into your implementation notes before coding. +Do not rely on guessed event names. + ### 4.1 Configuration Management #### Config Model Design @@ -235,6 +264,18 @@ language = self.config.params.get("language", "en-US") # With default value **Note**: The `property.json` file generated by the template is empty `{}`. You need to manually add configurations. +### 4.1.1 Error Classification Rules + +Use these rules consistently: + +- Invalid config or missing required config: `FATAL_ERROR` +- Auth failure such as `401`, `403`, invalid API key: `FATAL_ERROR` +- Transient disconnect or timeout: `NON_FATAL_ERROR` +- Retry ceiling reached: `FATAL_ERROR` + +When the vendor supplies machine-readable error details, include +`ModuleErrorVendorInfo`. + ### 4.2 Read Configuration ```python diff --git a/content/docs/ten_agent_examples/extension_dev/create_tts_extension.mdx b/content/docs/ten_agent_examples/extension_dev/create_tts_extension.mdx index fd24751..5bc20a4 100644 --- a/content/docs/ten_agent_examples/extension_dev/create_tts_extension.mdx +++ b/content/docs/ten_agent_examples/extension_dev/create_tts_extension.mdx @@ -7,6 +7,20 @@ description: Build, develop, test, and publish a complete TTS extension from scr This guide walks you through building a production-grade TTS (Text-to-Speech) Extension from scratch, covering the full workflow from project setup and core development to testing, validation, and publishing. +## Definition of Done + +Before calling a new TTS extension complete, verify all of these: + +- Vendor wire contract is confirmed: endpoint, event names, payload encoding, output framing. +- Correct base class is used for the transport mode. +- Config is validated and secrets are redacted in logs. +- `config:` log output is present. +- Fatal vs non-fatal errors are classified intentionally. +- Reconnect behavior is bounded and auth failures are fatal. +- Standalone tests are present and green. +- Guarder tests are green. +- README and example configuration are accurate. + ## What Is a TTS Extension The TTS Extension is a **standard extension building block** in the TEN Framework ecosystem, designed specifically for text-to-speech functionality. @@ -193,6 +207,54 @@ HTTP mode uses standard HTTP streaming requests. It is suitable for traditional HTTP mode is built on `AsyncTTS2HttpExtension`, which already provides full request handling, TTFB reporting, and audio data handling. Developers only need to implement the client interface and the config class. +## WebSocket Mode + +For streaming TTS, verify the provider contract before implementation: + +- Exact WebSocket endpoint +- Outbound event names for text chunks and end-of-input +- Inbound event names for audio, end-of-audio, and errors +- Whether audio payloads are raw bytes or base64-encoded fields +- Whether the connection is long-lived or should be reconnected per request + +Do not guess the wire protocol from examples of other vendors. + +### Connection Lifecycle Expectations + +Production-ready WebSocket TTS extensions should define: + +- Preheat or initial connect behavior +- Cancel behavior +- Reconnect behavior after transient failures +- Exponential backoff and retry ceiling +- Fatal auth/config failure handling +- Lifecycle logs emitted at `info` level if automation depends on them + +### Input Validation Notes + +Validate edge cases explicitly: + +- Empty or whitespace-only input should follow the actual guarder expectation +- Punctuation-only input should be tested explicitly +- Very long input should be bounded intentionally + +Guarder behavior should win over assumptions about what seems cleaner. + +### Minimum Required Standalone Tests + +Production-ready TTS extensions should include at least: + +- `test_basic_audio` +- `test_flush` +- `test_invalid_api_key` +- `test_metrics` +- `test_state_machine` +- `test_robustness` +- `test_dump` + +If subtitle or word alignment is unsupported, document that explicitly rather +than faking support. + #### 1. Implement the HTTP Client Interface **The client must implement the `AsyncTTS2HttpClient` interface:** From 908bb4bf9c08b0f9c0bc2cce205349ca8763647d Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Fri, 24 Apr 2026 15:13:52 +0000 Subject: [PATCH 4/5] docs(tts): trim extension guide examples (#314) --- .../extension_dev/create_tts_extension.mdx | 555 ++---------------- 1 file changed, 48 insertions(+), 507 deletions(-) diff --git a/content/docs/ten_agent_examples/extension_dev/create_tts_extension.mdx b/content/docs/ten_agent_examples/extension_dev/create_tts_extension.mdx index 5bc20a4..46e5b20 100644 --- a/content/docs/ten_agent_examples/extension_dev/create_tts_extension.mdx +++ b/content/docs/ten_agent_examples/extension_dev/create_tts_extension.mdx @@ -207,54 +207,6 @@ HTTP mode uses standard HTTP streaming requests. It is suitable for traditional HTTP mode is built on `AsyncTTS2HttpExtension`, which already provides full request handling, TTFB reporting, and audio data handling. Developers only need to implement the client interface and the config class. -## WebSocket Mode - -For streaming TTS, verify the provider contract before implementation: - -- Exact WebSocket endpoint -- Outbound event names for text chunks and end-of-input -- Inbound event names for audio, end-of-audio, and errors -- Whether audio payloads are raw bytes or base64-encoded fields -- Whether the connection is long-lived or should be reconnected per request - -Do not guess the wire protocol from examples of other vendors. - -### Connection Lifecycle Expectations - -Production-ready WebSocket TTS extensions should define: - -- Preheat or initial connect behavior -- Cancel behavior -- Reconnect behavior after transient failures -- Exponential backoff and retry ceiling -- Fatal auth/config failure handling -- Lifecycle logs emitted at `info` level if automation depends on them - -### Input Validation Notes - -Validate edge cases explicitly: - -- Empty or whitespace-only input should follow the actual guarder expectation -- Punctuation-only input should be tested explicitly -- Very long input should be bounded intentionally - -Guarder behavior should win over assumptions about what seems cleaner. - -### Minimum Required Standalone Tests - -Production-ready TTS extensions should include at least: - -- `test_basic_audio` -- `test_flush` -- `test_invalid_api_key` -- `test_metrics` -- `test_state_machine` -- `test_robustness` -- `test_dump` - -If subtitle or word alignment is unsupported, document that explicitly rather -than faking support. - #### 1. Implement the HTTP Client Interface **The client must implement the `AsyncTTS2HttpClient` interface:** @@ -431,6 +383,26 @@ WebSocket mode supports bidirectional WebSocket communication and allows streami ### Implementation Rules +Before implementing a WebSocket TTS extension, verify the provider wire +contract: + +- Exact WebSocket endpoint +- Outbound event names for text chunks and end-of-input +- Inbound event names for audio, end-of-audio, and errors +- Whether audio payloads are raw bytes or base64-encoded fields +- Whether the connection is long-lived or should be reconnected per request + +Do not infer the wire protocol from another vendor's example. + +Define the connection lifecycle up front: + +- Preheat or initial connect behavior +- Cancel behavior +- Reconnect behavior after transient failures +- Exponential backoff and retry ceiling +- Fatal auth/config failure handling +- Lifecycle logs at `info` level when automation depends on them + #### 1. Connection Management Strategy **Connection lifecycle management:** @@ -587,465 +559,24 @@ def _clear_queues(self) -> None: 6. **Set timeouts**: Avoid long blocking periods. 7. **Log properly**: Include vendor errors, state changes, request flow, and response flow. See [Logging Specifications](#logging-specifications). -### Simple Serial WebSocket Pattern - -Not all WebSocket TTS vendors need the full duplex send/receive loop shown above. Many vendors (such as Deepgram) support a simpler serial pattern: send text, receive audio chunks, repeat. This pattern is easier to implement and debug. Choose this path when the vendor API processes one utterance at a time over a persistent connection. - -#### Complete Client Example - -```python title="{vendor}_tts.py" -import asyncio -import json -from datetime import datetime -from typing import AsyncIterator -from urllib.parse import urlencode - -import websockets -from websockets.asyncio.client import ClientConnection -from websockets.exceptions import InvalidStatus - -from .config import VendorTTSConfig - -# Event types yielded back to the extension -EVENT_TTS_RESPONSE = 1 # Audio bytes -EVENT_TTS_END = 2 # Vendor finished this utterance -EVENT_TTS_ERROR = 3 # Vendor error -EVENT_TTS_TTFB_METRIC = 5 # Time-to-first-byte measurement - -WS_RECV_TIMEOUT = 8.0 # Seconds before giving up on a response - - -class VendorTTSConnectionException(Exception): - """Typed connection failure that preserves the HTTP status code. - - The extension uses status_code to distinguish fatal errors (e.g. 401) - from transient ones, mapping them to the correct TEN error code. - """ - def __init__(self, status_code: int, body: str): - self.status_code = status_code - self.body = body - super().__init__(f"Connection failed (code: {status_code}): {body}") - - -class VendorTTSClient: - """Serial WebSocket TTS client. - - Each get() call sends a text request and streams audio until done. - The connection is reused across calls and reconnected when needed. - """ - - def __init__(self, config: VendorTTSConfig, ten_env): - self.config = config - self.ten_env = ten_env - self._ws: ClientConnection | None = None - self._is_cancelled = False - self._needs_reconnect = False - self._sent_ts: datetime | None = None - self._ttfb_sent: bool = False - self._ws_url = self._build_ws_url() - - def _build_ws_url(self) -> str: - """Build the WebSocket URL with query parameters. - - Known config keys become fixed query params. Any extra keys - in config.params are forwarded as-is — see the params - passthrough pattern in Configuration Management Design. - """ - query_params: dict[str, str | int | float | bool] = { - "model": self.config.model, - "encoding": self.config.encoding, - "sample_rate": self.config.sample_rate, - } - for key, value in self.config.params.items(): - if key not in {"api_key", "base_url"} and value is not None: - query_params[key] = value - return f"{self.config.base_url}?{urlencode(query_params)}" - - async def start(self) -> None: - """Preheat: establish initial connection during on_init.""" - await self._connect() - - async def stop(self) -> None: - """Tear down the connection during on_stop.""" - if self._ws: - try: - await self._ws.close() - except Exception: - pass - self._ws = None - - async def cancel(self) -> None: - """Cancel current TTS. Sends Flush and drains until Flushed - so the connection is left in a clean state for the next request.""" - self._is_cancelled = True - if self._ws: - try: - await self._ws.send(json.dumps({"type": "Flush"})) - await asyncio.wait_for(self._drain_until_flushed(), timeout=3.0) - except Exception: - self._needs_reconnect = True - - def reset_ttfb(self) -> None: - """Reset TTFB tracking for a new request.""" - self._sent_ts = None - self._ttfb_sent = False - - async def get(self, text: str) -> AsyncIterator[tuple[bytes | int | None, int]]: - """Send text and yield (data, event_type) tuples. - - Callers iterate this to receive audio chunks, TTFB metrics, - end-of-stream, or error events. - """ - if not text.strip(): - yield None, EVENT_TTS_END - return - - if self._needs_reconnect: - await self._reconnect() - await self._ensure_connection() - - if not self._ttfb_sent: - self._sent_ts = datetime.now() - - # Clear cancel flag just before sending — avoids a race - # where a concurrent cancel() call gets overwritten - self._is_cancelled = False - - # Send Speak + Flush to trigger audio generation - await self._ws.send(json.dumps({"type": "Speak", "text": text})) - await self._ws.send(json.dumps({"type": "Flush"})) - - # Stream audio until Flushed or error - try: - while not self._is_cancelled: - try: - message = await asyncio.wait_for( - self._ws.recv(), timeout=WS_RECV_TIMEOUT - ) - except asyncio.TimeoutError: - self._needs_reconnect = True - yield b"Timeout waiting for audio", EVENT_TTS_ERROR - break - - if isinstance(message, bytes): - if self._is_cancelled: - break - # TTFB on first audio chunk - if self._sent_ts and not self._ttfb_sent: - ttfb_ms = int( - (datetime.now() - self._sent_ts).total_seconds() * 1000 - ) - yield ttfb_ms, EVENT_TTS_TTFB_METRIC - self._ttfb_sent = True - yield message, EVENT_TTS_RESPONSE - else: - data = json.loads(message) - msg_type = data.get("type", "") - if msg_type == "Flushed": - yield None, EVENT_TTS_END - break - elif msg_type == "Error": - self._needs_reconnect = True - yield data.get("err_msg", "").encode(), EVENT_TTS_ERROR - break - except Exception as e: - self._needs_reconnect = True - yield str(e).encode(), EVENT_TTS_ERROR - - async def _connect(self) -> None: - headers = {"Authorization": f"Token {self.config.api_key}"} - try: - self._ws = await websockets.connect( - self._ws_url, additional_headers=headers - ) - except InvalidStatus as e: - # Preserve the HTTP status code so the extension can - # distinguish 401 (FATAL_ERROR) from transient failures - raise VendorTTSConnectionException( - status_code=e.response.status_code, - body=str(e), - ) from e - - async def _ensure_connection(self) -> None: - if not self._ws: - await self._connect() - - async def _reconnect(self) -> None: - await self.stop() - await self._connect() - self._needs_reconnect = False - - async def _drain_until_flushed(self) -> None: - """Read and discard messages until Flushed.""" - while self._ws: - msg = await self._ws.recv() - if isinstance(msg, str): - if json.loads(msg).get("type") == "Flushed": - return -``` - -#### Complete Extension Example - -The extension below is a complete, working implementation that handles the full lifecycle: initialization, request processing, TTFB metrics, cancellation, error handling, and cleanup. - -**Key patterns to note:** -- **`_finalize_request()`** consolidates all cleanup into one place (audio end, finish request, state reset) -- **`finish_request()`** is called to signal the base class queue that the current request is done — **forgetting this will hang the queue** -- **`_audio_start_sent`** ensures `tts_audio_start` is always sent before `tts_audio_end`, even on errors - -```python title="extension.py" -from datetime import datetime -from ten_ai_base.tts2 import AsyncTTS2BaseExtension -from ten_ai_base.message import ( - ModuleError, ModuleErrorCode, ModuleType, - ModuleErrorVendorInfo, TTSAudioEndReason, -) -from ten_ai_base.struct import TTSTextInput -from ten_ai_base.const import LOG_CATEGORY_VENDOR, LOG_CATEGORY_KEY_POINT -from .config import VendorTTSConfig -from .vendor_tts import ( - VendorTTSClient, VendorTTSConnectionException, - EVENT_TTS_RESPONSE, EVENT_TTS_END, - EVENT_TTS_TTFB_METRIC, EVENT_TTS_ERROR, -) - - -class VendorTTSExtension(AsyncTTS2BaseExtension): - def __init__(self, name: str) -> None: - super().__init__(name) - self.config: VendorTTSConfig | None = None - self.client: VendorTTSClient | None = None - self.current_request_id: str | None = None - self.current_request_finished: bool = False - self.total_audio_bytes: int = 0 - self.sent_ts: datetime | None = None - self._audio_start_sent: bool = False - - async def on_init(self, ten_env) -> None: - await super().on_init(ten_env) - config_json, _ = await self.ten_env.get_property_to_json("") - self.config = VendorTTSConfig.model_validate_json(config_json) - self.config.update_params() - ten_env.log_info( - self.config.to_str(sensitive_handling=True), - category=LOG_CATEGORY_KEY_POINT, - ) - # Await start() directly — do not use asyncio.create_task() - # so preheat failures surface during initialization - self.client = VendorTTSClient(config=self.config, ten_env=ten_env) - await self.client.start() - - async def on_stop(self, ten_env) -> None: - if self.client: - await self.client.stop() - await super().on_stop(ten_env) - - def vendor(self) -> str: - return "vendor_name" - - def synthesize_audio_sample_rate(self) -> int: - return self.config.sample_rate if self.config else 24000 - - async def cancel_tts(self) -> None: - """Called by the base class on flush. Cancel in-flight audio - and finalize the request as INTERRUPTED.""" - self.current_request_finished = True - if self.current_request_id and self.client: - await self.client.cancel() - await self._finalize_request(TTSAudioEndReason.INTERRUPTED) - - async def request_tts(self, t: TTSTextInput) -> None: - """Called by the base class for each queued text input.""" - try: - await self._do_request_tts(t) - except VendorTTSConnectionException as e: - # Map HTTP status to TEN error code: - # 401 → FATAL_ERROR (bad credentials, won't recover) - # anything else → NON_FATAL_ERROR (transient) - code = ( - ModuleErrorCode.FATAL_ERROR - if e.status_code == 401 - else ModuleErrorCode.NON_FATAL_ERROR - ) - error = ModuleError( - message=e.body, - module=ModuleType.TTS, - code=code, - vendor_info=ModuleErrorVendorInfo( - vendor=self.vendor(), - code=str(e.status_code), - message=e.body, - ), - ) - await self._finalize_request(TTSAudioEndReason.ERROR, error=error) - - async def _do_request_tts(self, t: TTSTextInput) -> None: - """Core request logic, separated so connection errors - can be caught and mapped in request_tts().""" - # --- New request setup --- - if t.request_id != self.current_request_id: - if self.client: - self.client.reset_ttfb() - self.current_request_id = t.request_id - self.current_request_finished = False - self.total_audio_bytes = 0 - self.sent_ts = None - self._audio_start_sent = False - - if t.text_input_end: - self.current_request_finished = True - - text = t.text.strip() - if not text: - # Empty text with text_input_end: finalize immediately - if t.text_input_end: - await self._finalize_request(TTSAudioEndReason.REQUEST_END) - return - - # --- Stream audio from vendor --- - self.ten_env.log_debug( - f"send_text_to_tts_server: {text} of request_id: {t.request_id}", - category=LOG_CATEGORY_VENDOR, - ) - if self.sent_ts is None: - self.sent_ts = datetime.now() - - async for data, event in self.client.get(text): - if event == EVENT_TTS_RESPONSE and isinstance(data, bytes): - self.total_audio_bytes += len(data) - await self.send_tts_audio_data(data) - - elif event == EVENT_TTS_TTFB_METRIC and isinstance(data, int): - self.sent_ts = datetime.now() - await self.send_tts_audio_start( - request_id=self.current_request_id, - ) - self._audio_start_sent = True - await self.send_tts_ttfb_metrics( - request_id=self.current_request_id, - ttfb_ms=data, - extra_metadata={"model": self.config.model}, - ) - - elif event == EVENT_TTS_END: - if t.text_input_end: - await self._finalize_request(TTSAudioEndReason.REQUEST_END) - break - - elif event == EVENT_TTS_ERROR: - error = ModuleError( - message=str(data), - module=ModuleType.TTS, - code=ModuleErrorCode.NON_FATAL_ERROR, - vendor_info=ModuleErrorVendorInfo(vendor=self.vendor()), - ) - if t.text_input_end: - await self._finalize_request( - TTSAudioEndReason.ERROR, error=error - ) - break - - async def _finalize_request( - self, reason: TTSAudioEndReason, error: ModuleError | None = None - ) -> None: - """Consolidate all request cleanup into one place. - - This pattern avoids duplicate finalization code scattered - across multiple branches. Every exit path calls this method. - """ - # Ensure audio_start was sent (protocol requires it before audio_end) - if not self._audio_start_sent: - await self.send_tts_audio_start( - request_id=self.current_request_id, - ) - self._audio_start_sent = True - - interval_ms = self._current_request_interval_ms() - duration_ms = self._calculate_audio_duration_ms() - - await self.send_tts_audio_end( - request_id=self.current_request_id, - request_event_interval_ms=interval_ms, - request_total_audio_duration_ms=duration_ms, - reason=reason, - ) - - # CRITICAL: Signal the base class queue that this request is done. - # Forgetting this call will hang the queue permanently. - await self.finish_request( - request_id=self.current_request_id, - reason=reason, - error=error, - ) - self.sent_ts = None - - def _current_request_interval_ms(self) -> int: - if not self.sent_ts: - return 0 - return int((datetime.now() - self.sent_ts).total_seconds() * 1000) - - def _calculate_audio_duration_ms(self) -> int: - if not self.config: - return 0 - bytes_per_sample = 2 # 16-bit PCM - return int( - self.total_audio_bytes - / (self.synthesize_audio_sample_rate() * bytes_per_sample) - * 1000 - ) -``` - -### Critical WebSocket Implementation Patterns - -These patterns are essential for passing all guarder tests. They apply to both the duplex and serial WebSocket patterns. - -#### 1. Always Call `finish_request()` - -When inheriting from `AsyncTTS2BaseExtension` directly (WebSocket mode), **you must call `finish_request()`** to signal the base class queue that the current request is complete. The HTTP base class does this automatically, but WebSocket subclasses must do it explicitly. Forgetting this call will hang the processing queue. - -```python -await self.finish_request( - request_id=self.current_request_id, - reason=reason, - error=error, -) -``` - -#### 2. Consolidate Cleanup with `_finalize_request()` - -Without a helper, the same cleanup code (send audio_end, flush recorder, call finish_request, reset state) gets duplicated across every exit path: normal completion, cancellation, errors, empty text. Extract it into a single `_finalize_request()` method as shown in the extension example above. - -#### 3. TTFB Measurement - -For WebSocket mode, you measure TTFB yourself: -- Record `sent_ts = datetime.now()` when sending text to the vendor -- On the first audio chunk, compute `ttfb_ms = (now - sent_ts) * 1000` -- Yield `EVENT_TTS_TTFB_METRIC` from the client, then call `send_tts_ttfb_metrics()` in the extension -- Call `send_tts_audio_start()` at TTFB time (before first audio chunk) - -#### 4. Handle Invalid Text Input - -The guarder `test_invalid_text_handling` sends empty strings, emoji-only, punctuation-only, and whitespace-only text. Your client must handle these gracefully: - -```python -# In the client's get() method — return immediately for empty text -if not text.strip(): - yield None, EVENT_TTS_END - return -``` - -For text that the vendor silently ignores (producing no audio), set a reasonable `WS_RECV_TIMEOUT` (5-8 seconds) so the extension times out and reports an error rather than hanging indefinitely. - -#### 5. Cancel and Reconnect Strategy - -- **Cancel**: Send a Flush command and drain responses until the vendor acknowledges it (e.g., `Flushed`). This leaves the connection in a clean state for the next request. -- **Reconnect flag**: Set `_needs_reconnect = True` on errors and timeouts. Check it at the start of `get()` and reconnect before sending. -- **Eager reconnect**: After connection errors, reconnect immediately rather than waiting for the next request. This reduces latency on the next call. -- **Typed auth failures**: Distinguish authentication / HTTP status failures (for example 401) from generic transport errors so invalid-credential tests map to the correct TEN fatal error behavior. - -#### 6. Ensure `tts_audio_start` Before `tts_audio_end` - -The TEN protocol requires `tts_audio_start` before `tts_audio_end`. Track whether it was sent with a flag (`_audio_start_sent`), and send it in `_finalize_request()` if it was not sent yet (e.g., on early errors). +### Additional WebSocket Requirements + +The full duplex example above is only one pattern. Some vendors support a +simpler serial model over a persistent connection. Keep the guide-level +requirements the same regardless of which shape you implement: + +- Call `finish_request()` on every exit path when inheriting from + `AsyncTTS2BaseExtension` directly. +- Use one cleanup/finalization path so `tts_audio_end`, queue completion, and + state reset stay consistent. +- Measure TTFB explicitly in WebSocket mode and report it through + `send_tts_ttfb_metrics()`. +- Make invalid-input behavior intentional and aligned with the guarder + expectations for empty, whitespace-only, punctuation-only, and oversized text. +- Treat auth/config failures as fatal and transient transport failures as + recoverable until the retry ceiling is reached. +- Ensure `tts_audio_start` is emitted before `tts_audio_end`, including early + error paths. ## Extension Structure and Supporting Files @@ -1087,6 +618,16 @@ If your vendor supports subtitle or word-timestamp alignment, also add: tests/configs/property_subtitle_alignment.json ``` +At minimum, the standalone suite should cover: + +- `test_basic_audio` +- `test_flush` +- `test_invalid_api_key` +- `test_metrics` +- `test_state_machine` +- `test_robustness` +- `test_dump` + ### Unit Test Best Practices 1. **Cover all major flows**: Make sure every primary path is tested. From 778033810f6762e12e83ce4ccfa18333df4ffb71 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Thu, 30 Apr 2026 10:06:01 +0000 Subject: [PATCH 5/5] docs(tts): clarify websocket chunk reuse and invalid input (#314) --- .../extension_dev/create_tts_extension.mdx | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/content/docs/ten_agent_examples/extension_dev/create_tts_extension.mdx b/content/docs/ten_agent_examples/extension_dev/create_tts_extension.mdx index 46e5b20..d941615 100644 --- a/content/docs/ten_agent_examples/extension_dev/create_tts_extension.mdx +++ b/content/docs/ten_agent_examples/extension_dev/create_tts_extension.mdx @@ -402,6 +402,8 @@ Define the connection lifecycle up front: - Exponential backoff and retry ceiling - Fatal auth/config failure handling - Lifecycle logs at `info` level when automation depends on them +- Reuse the same WebSocket connection across multiple `tts_text_input` chunks + that share a `request_id`; do not open and close the connection per chunk #### 1. Connection Management Strategy @@ -571,8 +573,11 @@ requirements the same regardless of which shape you implement: state reset stay consistent. - Measure TTFB explicitly in WebSocket mode and report it through `send_tts_ttfb_metrics()`. -- Make invalid-input behavior intentional and aligned with the guarder - expectations for empty, whitespace-only, punctuation-only, and oversized text. +- If the vendor does not reject invalid input cleanly, validate it before + sending or add timeout/error handling in the client. In all cases, keep + behavior aligned with the guarder expectations for empty, whitespace-only, + punctuation-only, and oversized text, and ensure + `test_invalid_text_handling` passes. - Treat auth/config failures as fatal and transient transport failures as recoverable until the retry ceiling is reached. - Ensure `tts_audio_start` is emitted before `tts_audio_end`, including early