From fdb747f9cf687dcc32b0d1c9651b654297196574 Mon Sep 17 00:00:00 2001 From: Lucas Rosa Date: Wed, 1 Jul 2026 23:39:32 +0000 Subject: [PATCH 1/3] feat(telnyx): add Telnyx assistant server framework MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add Telnyx hosted AI Assistant as a first-class EVA framework alongside the existing Pipecat pipeline. The Telnyx provider connects to the Telnyx AI gateway via direct anonymous WebRTC (Verto JSON-RPC over WebSocket), bridges audio between the EVA user simulator (Twilio mulaw frames) and the Telnyx assistant (PCM 16 kHz), and enriches the audit log from the Telnyx Conversations API after the call ends. New files: - src/eva/assistant/base_server.py AbstractAssistantServer: template-method base class that defines the EVA server contract (FastAPI WebSocket endpoint, audio bridging, tool execution via ToolExecutor, audit log, audio recording, save_outputs). Concrete servers implement start() and _shutdown(); the base handles stop(), output saving, and artifact generation. - src/eva/assistant/audio_bridge.py Audio format conversion utilities: mulaw 8 kHz ↔ PCM 16-bit 24 kHz resampling (libsoxr), PCM mixing, Twilio FrameSerializer protocol helpers, and framework/metrics log writers. - src/eva/assistant/telnyx_server.py TelnyxAssistantServer(AbstractAssistantServer): the main server. - TelnyxAssistantManager: REST CRUD for benchmark assistants. - TelnyxDirectSessionConfig: dataclass for WebRTC session params. - TelnyxVertoJsonRpcClient: Verto JSON-RPC over WebSocket. - TelnyxWebRTCClient: aiortc RTCPeerConnection + media management. - TelnyxAssistantServer: bridges EVA Twilio frames to Telnyx WebRTC, handles conversation events, tool calls, dynamic variables, audit-log enrichment from Conversations API. - src/eva/utils/audio_utils.py save_pcm_as_wav() helper shared by all server implementations. - tests/unit/assistant/test_telnyx_server.py (20 tests) - tests/unit/assistant/test_audio_bridge.py (3 tests) Modified: - src/eva/models/config.py: Add framework field to RunConfig. - src/eva/orchestrator/worker.py: Add _get_server_class() dispatch. - pyproject.toml: Add aiortc, aiohttp, soxr dependencies. Addresses VSDK-277 review: adds the missing Python EVA wrapper that subclasses AbstractAssistantServer, registers in worker.py, routes tool calls through ToolExecutor, and produces all required EVA artifacts. --- pyproject.toml | 3 + src/eva/assistant/telnyx_server.py | 1674 ++++++++++++++++++++ src/eva/models/config.py | 3 +- src/eva/orchestrator/worker.py | 6 +- tests/unit/assistant/__init__.py | 0 tests/unit/assistant/test_telnyx_server.py | 432 +++++ 6 files changed, 2116 insertions(+), 2 deletions(-) create mode 100644 src/eva/assistant/telnyx_server.py create mode 100644 tests/unit/assistant/__init__.py create mode 100644 tests/unit/assistant/test_telnyx_server.py diff --git a/pyproject.toml b/pyproject.toml index f7d79a95..6b2af8c7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -40,6 +40,9 @@ dependencies = [ "uvicorn>=0.23.0", "websockets>=12.0", "httpx>=0.25.0", + "aiohttp>=3.8.0", + "aiortc>=1.6.0", + "soxr>=0.3.0", "pyyaml>=6.0", "pandas>=2.0", "numpy>=1.24", diff --git a/src/eva/assistant/telnyx_server.py b/src/eva/assistant/telnyx_server.py new file mode 100644 index 00000000..977d1f82 --- /dev/null +++ b/src/eva/assistant/telnyx_server.py @@ -0,0 +1,1674 @@ +"""Telnyx hosted AI Assistant server for EVA. + +Expected ``pipeline_config.s2s_params`` keys: + +Required for direct Telnyx WebRTC calls: + * ``assistant_id`` or ``assistant_agent_id``: Telnyx AI Assistant target id. + +Optional for direct Telnyx WebRTC calls: + * ``target_version_id``: specific Telnyx AI Assistant version. + * ``host`` or ``websocket_host``: Verto websocket host, defaults to + ``wss://rtc.telnyx.com``. + * ``target_params`` and ``user_variables``: anonymous-login metadata. + * ``caller_name``, ``caller_number``, ``client_state``, ``custom_headers``: + dialog metadata sent with the WebRTC invite. + +Optional assistant setup: + * ``create_assistant``: create a Telnyx AI Assistant before the run. + * ``api_key`` or ``telnyx_api_key``: Telnyx API key, required only for REST + assistant creation/deletion or transcript fetch helpers. + * ``api_base``: defaults to ``https://api.telnyx.com``; ``/v2`` is appended + automatically for v2 endpoints unless already present. + * ``model``, ``voice``, and ``stt_model``: used when creating an assistant. + +The server exposes the upstream EVA assistant-server contract to the local user +simulator while bridging to Telnyx's anonymous Verto/WebRTC AI Assistant path. +""" + +from __future__ import annotations + +import asyncio +import audioop +import base64 +import json +import time +import uuid +from collections.abc import Awaitable, Callable +from dataclasses import dataclass, field +from datetime import datetime +from fractions import Fraction +from pathlib import Path +from typing import Any + +import aiohttp +import uvicorn +from fastapi import FastAPI, HTTPException, Request, WebSocket, WebSocketDisconnect + +from eva.assistant.audio_bridge import ( + FrameworkLogWriter, + MetricsLogWriter, + create_twilio_media_message, + mulaw_8k_to_pcm16_16k, + mulaw_8k_to_pcm16_24k, + parse_twilio_media_message, + sync_buffer_to_position, +) +from eva.assistant.base_server import AbstractAssistantServer +from eva.utils.culture import get_initial_message +from eva.models.agents import AgentConfig, AgentTool +from eva.utils.logging import get_logger + +logger = get_logger(__name__) + +TELNYX_DEFAULT_API_BASE = "https://api.telnyx.com" +TELNYX_PROD_RTC_HOST = "wss://rtc.telnyx.com" +TELNYX_DEV_RTC_HOST = "wss://rtcdev.telnyx.com" +TELNYX_SAMPLE_RATE = 16000 +RECORDING_SAMPLE_RATE = 24000 +PCM_SAMPLE_WIDTH = 2 +MULAW_CHUNK_SIZE = 160 +MULAW_CHUNK_DURATION_S = 0.02 +TELNYX_USER_AGENT = "EVA-TelnyxWebRTC/0.1" +TELNYX_DEFAULT_ICE_SERVERS = [ + {"urls": "stun:stun.telnyx.com:3478"}, + {"urls": "stun:stun.l.google.com:19302"}, + { + "urls": "turn:turn.telnyx.com:3478?transport=udp", + "username": "testuser", + "credential": "testpassword", + }, + { + "urls": "turn:turn.telnyx.com:3478?transport=tcp", + "username": "testuser", + "credential": "testpassword", + }, +] + + +def _wall_ms() -> str: + return str(int(round(time.time() * 1000))) + + +def _normalize_api_v2_base(api_base: str | None) -> str: + base = (api_base or TELNYX_DEFAULT_API_BASE).rstrip("/") + if base.endswith("/v2"): + return base + return f"{base}/v2" + + +def _pcm16_16k_to_pcm16_24k(pcm_16k: bytes) -> bytes: + if not pcm_16k: + return b"" + pcm_24k, _ = audioop.ratecv(pcm_16k, PCM_SAMPLE_WIDTH, 1, 16000, 24000, None) + return pcm_24k + + +def _pcm16_16k_to_mulaw_8k(pcm_16k: bytes) -> bytes: + if not pcm_16k: + return b"" + pcm_8k, _ = audioop.ratecv(pcm_16k, PCM_SAMPLE_WIDTH, 1, 16000, 8000, None) + return audioop.lin2ulaw(pcm_8k, PCM_SAMPLE_WIDTH) + + +def _iso8601_to_epoch_ms(value: str) -> int: + dt = datetime.fromisoformat(value) + return int(dt.timestamp() * 1000) + + +def _is_epoch_ms(value: str | None) -> bool: + if not value: + return False + try: + return int(value) > 1_000_000_000_000 + except (TypeError, ValueError): + return False + + +def _extract_tool_params(body: Any) -> dict[str, Any]: + """Normalize Telnyx webhook bodies to the function argument dict.""" + if body is None: + return {} + if not isinstance(body, dict): + raise ValueError("Tool webhook body must be a JSON object") + + function_params = body.get("function_params") + if function_params is None: + return body + if not isinstance(function_params, dict): + raise ValueError("function_params must be a JSON object") + return function_params + + +def _message_text(message: dict[str, Any]) -> str: + text = message.get("text") + if isinstance(text, str): + return text.strip() + + content = message.get("content") + if isinstance(content, str): + return content.strip() + if isinstance(content, list): + parts: list[str] = [] + for item in content: + if isinstance(item, str): + parts.append(item) + elif isinstance(item, dict): + part_text = item.get("text") or item.get("content") + if isinstance(part_text, str): + parts.append(part_text) + return "".join(parts).strip() + return "" + + +def _conversation_item_text(item: dict[str, Any]) -> str: + content = item.get("content") + if isinstance(content, str): + return content.strip() + if not isinstance(content, list): + return "" + parts: list[str] = [] + for part in content: + if isinstance(part, str): + parts.append(part) + elif isinstance(part, dict): + text = part.get("text") or part.get("content") + if isinstance(text, str): + parts.append(text) + return "".join(parts).strip() + + +def parse_telnyx_intended_speech_payload(payload: dict[str, Any]) -> list[dict[str, Any]]: + """Extract chronological assistant messages from Telnyx conversation messages.""" + data = payload.get("data", []) + if not isinstance(data, list): + return [] + + intended_speech: list[dict[str, Any]] = [] + for message in reversed(data): + if not isinstance(message, dict) or message.get("role") != "assistant": + continue + text = _message_text(message) + if not text: + continue + sent_at = message.get("sent_at") or message.get("created_at") + if not isinstance(sent_at, str) or not sent_at.strip(): + continue + try: + timestamp_ms = _iso8601_to_epoch_ms(sent_at) + except ValueError: + continue + intended_speech.append({"text": text, "timestamp_ms": timestamp_ms}) + return intended_speech + + +def extract_user_speech_events(events_path: Path) -> list[dict[str, Any]]: + """Read simulated-user transcript events saved by ``UserSimulator``.""" + if not events_path.exists(): + return [] + + turns: list[dict[str, Any]] = [] + with open(events_path, encoding="utf-8") as file_obj: + for line in file_obj: + line = line.strip() + if not line: + continue + try: + event = json.loads(line) + except json.JSONDecodeError: + continue + if event.get("type") != "user_speech": + continue + data = event.get("data", {}) + if not isinstance(data, dict): + continue + text = str(data.get("text", "")).strip() + if text: + turns.append({"text": text, "timestamp_ms": str(event.get("timestamp", ""))}) + return turns + + +def extract_assistant_speech_events(events_path: Path) -> list[dict[str, Any]]: + """Read assistant speech heard by the user simulator as a fallback transcript.""" + if not events_path.exists(): + return [] + + turns: list[dict[str, Any]] = [] + with open(events_path, encoding="utf-8") as file_obj: + for line in file_obj: + line = line.strip() + if not line: + continue + try: + event = json.loads(line) + except json.JSONDecodeError: + continue + if event.get("type") != "assistant_speech": + continue + data = event.get("data", {}) + if not isinstance(data, dict): + continue + text = str(data.get("text", "")).strip() + if text: + turns.append({"text": text, "timestamp_ms": int(event.get("timestamp", 0))}) + return turns + + +class TelnyxAssistantManager: + """Create and manage Telnyx AI assistants for benchmark runs.""" + + DEFAULT_MODEL = "moonshotai/Kimi-K2.5" + DEFAULT_VOICE = "Telnyx.Ultra.a7a59115-2425-4192-844c-1e98ec7d6877" + DEFAULT_STT_MODEL = "deepgram/flux" + + def __init__( + self, + api_key: str, + api_base: str = TELNYX_DEFAULT_API_BASE, + session: aiohttp.ClientSession | None = None, + ): + self.api_key = api_key + self.api_base = _normalize_api_v2_base(api_base) + self._session_owner = session is None + self.session = session or aiohttp.ClientSession( + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + }, + timeout=aiohttp.ClientTimeout(total=30.0), + ) + + async def create_benchmark_assistant( + self, + agent_config: AgentConfig, + agent_config_path: str, + webhook_base_url: str | None = None, + model: str | None = None, + voice: str | None = None, + stt_model: str | None = None, + ) -> str: + payload = self._build_assistant_payload( + agent_config=agent_config, + agent_config_path=agent_config_path, + webhook_base_url=webhook_base_url, + model=model or self.DEFAULT_MODEL, + voice=voice or self.DEFAULT_VOICE, + stt_model=stt_model or self.DEFAULT_STT_MODEL, + ) + async with self.session.post(f"{self.api_base}/ai/assistants", json=payload) as response: + response_payload = await self._parse_response_json(response) + if response.status >= 400: + raise RuntimeError( + f"Failed to create Telnyx assistant: {response.status} " + f"{json.dumps(response_payload, sort_keys=True)}" + ) + + assistant_id = str(response_payload.get("id") or response_payload.get("data", {}).get("id") or "").strip() + if not assistant_id: + raise RuntimeError(f"Telnyx assistant creation response did not include an id: {response_payload}") + return assistant_id + + async def delete_assistant(self, assistant_id: str) -> None: + async with self.session.delete(f"{self.api_base}/ai/assistants/{assistant_id}") as response: + payload = await self._parse_response_json(response) + if response.status >= 400: + raise RuntimeError( + f"Failed to delete Telnyx assistant {assistant_id}: " + f"{response.status} {json.dumps(payload, sort_keys=True)}" + ) + + async def get_assistant_model(self, assistant_id: str) -> str: + async with self.session.get(f"{self.api_base}/ai/assistants/{assistant_id}") as response: + payload = await self._parse_response_json(response) + if response.status >= 400: + raise RuntimeError( + f"Failed to get Telnyx assistant {assistant_id}: " + f"{response.status} {json.dumps(payload, sort_keys=True)}" + ) + return str(payload.get("model") or payload.get("data", {}).get("model", "")) + + async def update_assistant_model(self, assistant_id: str, model: str) -> None: + async with self.session.patch( + f"{self.api_base}/ai/assistants/{assistant_id}", json={"model": model} + ) as response: + payload = await self._parse_response_json(response) + if response.status >= 400: + raise RuntimeError( + f"Failed to update Telnyx assistant model: {response.status} {json.dumps(payload, sort_keys=True)}" + ) + + async def close(self) -> None: + if self._session_owner and not self.session.closed: + await self.session.close() + + @staticmethod + async def _parse_response_json(response: aiohttp.ClientResponse) -> dict[str, Any]: + try: + payload = await response.json() + except aiohttp.ContentTypeError: + payload = {"message": await response.text()} + return payload if isinstance(payload, dict) else {"data": payload} + + def _build_assistant_payload( + self, + agent_config: AgentConfig, + agent_config_path: str, + webhook_base_url: str | None, + model: str, + voice: str, + stt_model: str | None = None, + ) -> dict[str, Any]: + normalized_webhook_base = webhook_base_url.rstrip("/") if webhook_base_url else "" + tools = ( + [self._build_webhook_tool(tool, normalized_webhook_base) for tool in agent_config.tools] + if normalized_webhook_base + else [] + ) + if not any(tool.id == "end_call" for tool in agent_config.tools): + tools.append( + { + "type": "hangup", + "hangup": { + "description": ( + "To be used whenever the conversation has ended " + "and it would be appropriate to hang up the call." + ), + }, + } + ) + + payload = { + "name": f"EVA Benchmark - {agent_config.name}", + "description": ( + f"Auto-generated by EVA from {Path(agent_config_path).name} for benchmark agent {agent_config.id}." + ), + "instructions": self._build_system_prompt(agent_config), + "model": model, + "greeting": getattr(agent_config, "greeting", None) or get_initial_message("en"), + "tools": tools, + "enabled_features": ["telephony"], + "voice_settings": { + "voice": voice, + "voice_speed": 1.2, + "expressive_mode": False, + "language_boost": "English", + "similarity_boost": 0.5, + "style": 0.0, + "use_speaker_boost": True, + "background_audio": { + "type": "predefined_media", + "value": "silence", + "volume": 0.5, + }, + }, + "transcription": { + "model": stt_model or self.DEFAULT_STT_MODEL, + "language": "en", + "settings": { + "eot_threshold": 0.9, + "eot_timeout_ms": 5000, + "eager_eot_threshold": 0.9, + }, + }, + "telephony_settings": { + "supports_unauthenticated_web_calls": True, + "time_limit_secs": 600, + "user_idle_timeout_secs": 60, + }, + "interruption_settings": { + "enable": True, + "start_speaking_plan": { + "wait_seconds": 0.1, + "transcription_endpointing_plan": { + "on_punctuation_seconds": 0.1, + "on_no_punctuation_seconds": 0.1, + "on_number_seconds": 0.1, + }, + }, + }, + "privacy_settings": { + "data_retention": True, + "pii_redaction": "disabled", + }, + "dynamic_variables": { + "eva_call_id": None, + }, + } + if normalized_webhook_base: + payload["dynamic_variables_webhook_url"] = f"{normalized_webhook_base}/dynamic-variables" + return payload + + @staticmethod + def _build_system_prompt(agent_config: AgentConfig) -> str: + sections = [ + agent_config.description.strip(), + f"Role: {agent_config.role.strip()}", + agent_config.personality.strip() if agent_config.personality else "", + agent_config.instructions.strip(), + ] + return "\n\n".join(section for section in sections if section) + + @staticmethod + def _build_webhook_tool(tool: AgentTool, webhook_base_url: str) -> dict[str, Any]: + return { + "type": "webhook", + "timeout_ms": 5000, + "webhook": { + "name": tool.id, + "description": tool.description, + "url": f"{webhook_base_url}/tools/{{{{eva_call_id}}}}/{tool.id}", + "method": "POST", + "path_parameters": {"type": "object", "properties": {}}, + "query_parameters": {"type": "object", "properties": {}}, + "body_parameters": { + "type": "object", + "properties": tool.get_parameter_properties(), + "required": tool.get_required_param_names(), + }, + "headers": [], + }, + } + + +@dataclass +class TelnyxDirectSessionConfig: + """Configuration for Telnyx anonymous Verto/WebRTC AI Assistant sessions.""" + + assistant_id: str + host: str = TELNYX_PROD_RTC_HOST + target_version_id: str | None = None + target_params: dict[str, Any] = field(default_factory=dict) + user_variables: dict[str, Any] = field(default_factory=dict) + caller_name: str = "EVA User" + caller_number: str = "+15555550100" + destination_number: str = "ai_assistant" + client_state: str | None = None + custom_headers: list[dict[str, str]] = field(default_factory=list) + ice_servers: list[dict[str, Any]] = field(default_factory=lambda: list(TELNYX_DEFAULT_ICE_SERVERS)) + + +def _load_aiortc_modules() -> tuple[Any, Any, Any, Any]: + try: + from aiortc import MediaStreamTrack, RTCPeerConnection, RTCSessionDescription + from av import AudioFrame + except ImportError as exc: + raise ValueError( + "Direct Telnyx WebRTC mode requires aiortc at runtime. " + "Install aiortc in the EVA environment to open the Telnyx media session." + ) from exc + return MediaStreamTrack, RTCPeerConnection, RTCSessionDescription, AudioFrame + + +def _normalize_telnyx_headers(value: Any) -> list[dict[str, str]]: + if value is None: + return [] + if isinstance(value, dict): + return [{"name": str(name), "value": str(header_value)} for name, header_value in value.items()] + if not isinstance(value, list): + raise ValueError("custom_headers must be a mapping or a list of name/value objects") + + headers: list[dict[str, str]] = [] + for item in value: + if not isinstance(item, dict): + raise ValueError("custom_headers list entries must be objects") + name = item.get("name") + header_value = item.get("value") + if name is None or header_value is None: + raise ValueError("custom_headers entries must include name and value") + headers.append({"name": str(name), "value": str(header_value)}) + return headers + + +def _require_dict(value: Any, name: str) -> dict[str, Any]: + if value is None: + return {} + if not isinstance(value, dict): + raise ValueError(f"{name} must be a JSON object") + return dict(value) + + +class TelnyxVertoJsonRpcClient: + """Minimal Verto JSON-RPC client for Telnyx anonymous WebRTC sessions.""" + + def __init__(self, host: str): + self.host = host + self._ws: Any | None = None + self._pending: dict[str, asyncio.Future[dict[str, Any]]] = {} + self._receive_task: asyncio.Task[None] | None = None + self._request_handler: Callable[[dict[str, Any]], Awaitable[None]] | None = None + + @property + def connected(self) -> bool: + return self._ws is not None + + async def connect(self, request_handler: Callable[[dict[str, Any]], Awaitable[None]]) -> None: + try: + import websockets + except ImportError as exc: + raise ValueError("Direct Telnyx WebRTC mode requires the websockets package") from exc + + self._request_handler = request_handler + self._ws = await websockets.connect(self.host) + self._receive_task = asyncio.create_task(self._receive_loop()) + + async def close(self) -> None: + if self._receive_task is not None: + self._receive_task.cancel() + try: + await self._receive_task + except asyncio.CancelledError: + pass + self._receive_task = None + + if self._ws is not None: + await self._ws.close() + self._ws = None + + for future in self._pending.values(): + if not future.done(): + future.cancel() + self._pending.clear() + + async def request(self, method: str, params: dict[str, Any]) -> dict[str, Any]: + if self._ws is None: + raise RuntimeError("Telnyx Verto websocket is not connected") + + request_id = str(uuid.uuid4()) + loop = asyncio.get_running_loop() + future: asyncio.Future[dict[str, Any]] = loop.create_future() + self._pending[request_id] = future + await self._ws.send(json.dumps({"jsonrpc": "2.0", "id": request_id, "method": method, "params": params})) + return await future + + async def reply(self, request_id: Any, result: dict[str, Any]) -> None: + if self._ws is None: + return + await self._ws.send(json.dumps({"jsonrpc": "2.0", "id": request_id, "result": result})) + + async def _receive_loop(self) -> None: + if self._ws is None: + return + try: + async for raw in self._ws: + try: + message = json.loads(raw) + except json.JSONDecodeError: + logger.warning("Received non-JSON Telnyx Verto message") + continue + await self._dispatch_message(message) + except asyncio.CancelledError: + pass + except Exception as exc: + for future in self._pending.values(): + if not future.done(): + future.set_exception(exc) + logger.info("Telnyx Verto websocket closed: %s", exc) + + async def _dispatch_message(self, message: dict[str, Any]) -> None: + request_id = str(message.get("id", "")) + if request_id in self._pending and ("result" in message or "error" in message): + future = self._pending.pop(request_id) + if "error" in message: + future.set_exception(RuntimeError(f"Telnyx Verto error: {message['error']}")) + else: + result = message.get("result") + future.set_result(result if isinstance(result, dict) else {}) + return + + if "method" in message and self._request_handler is not None: + await self._request_handler(message) + + +class TelnyxWebRTCClient: + """Direct Telnyx anonymous WebRTC client shaped after the Telnyx JS SDK.""" + + def __init__( + self, + *, + config: TelnyxDirectSessionConfig, + audio_handler: Callable[[bytes], Awaitable[None]], + ): + self.config = config + self.audio_handler = audio_handler + + self.session_id: str | None = None + self.call_id = str(uuid.uuid4()) + self.disconnected_event = asyncio.Event() + + self._rpc = TelnyxVertoJsonRpcClient(config.host) + self._peer: Any | None = None + self._audio_track: Any | None = None + self._remote_audio_task: asyncio.Task[None] | None = None + self._remote_description_set = asyncio.Event() + self._conversation_event_handler: Callable[[dict[str, Any]], Awaitable[None]] | None = None + + @property + def eva_call_id(self) -> str: + return self.call_id + + @staticmethod + def build_anonymous_login_params(config: TelnyxDirectSessionConfig) -> dict[str, Any]: + params: dict[str, Any] = { + "target_type": "ai_assistant", + "target_id": config.assistant_id, + "userVariables": config.user_variables, + "reconnection": False, + "User-Agent": { + "sdkVersion": TELNYX_USER_AGENT, + "data": "EVA assistant server", + }, + } + if config.target_version_id: + params["target_version_id"] = config.target_version_id + if config.target_params: + params["target_params"] = config.target_params + return params + + def build_invite_params(self, sdp: str) -> dict[str, Any]: + if not self.session_id: + raise RuntimeError("Telnyx session id is not available") + + dialog_params: dict[str, Any] = { + "callID": self.call_id, + "caller_id_name": self.config.caller_name, + "caller_id_number": self.config.caller_number, + "destination_number": self.config.destination_number, + "audio": True, + } + if self.config.client_state: + dialog_params["client_state"] = self.config.client_state + if self.config.custom_headers: + dialog_params["custom_headers"] = self.config.custom_headers + + return { + "sessid": self.session_id, + "sdp": sdp, + "dialogParams": dialog_params, + "User-Agent": TELNYX_USER_AGENT, + } + + async def start(self) -> None: + _, RTCPeerConnection, RTCSessionDescription, AudioFrame = _load_aiortc_modules() + await self._rpc.connect(self._handle_verto_request) + + login_result = await self._rpc.request( + "anonymous_login", + self.build_anonymous_login_params(self.config), + ) + self.session_id = str(login_result.get("sessid") or "").strip() + if not self.session_id: + raise RuntimeError(f"Telnyx anonymous_login response did not include sessid: {login_result}") + + self._peer = self._create_peer_connection(RTCPeerConnection) + self._audio_track = self._build_audio_track(AudioFrame) + self._peer.addTrack(self._audio_track) + + @self._peer.on("track") + def on_track(track: Any) -> None: + if track.kind == "audio": + logger.info("Telnyx WebRTC remote audio track received") + self._remote_audio_task = asyncio.create_task(self._receive_remote_audio(track)) + + offer = await self._peer.createOffer() + await self._peer.setLocalDescription(offer) + await self._wait_for_ice_gathering_complete() + + local_description = self._peer.localDescription + invite_result = await self._rpc.request("telnyx_rtc.invite", self.build_invite_params(local_description.sdp)) + result_sdp = invite_result.get("sdp") + if isinstance(result_sdp, str) and result_sdp: + await self._set_remote_description(RTCSessionDescription, result_sdp) + + def _create_peer_connection(self, RTCPeerConnection: Any) -> Any: + from aiortc import RTCConfiguration, RTCIceServer + + ice_servers = [ + RTCIceServer( + urls=server["urls"], + username=server.get("username"), + credential=server.get("credential"), + ) + for server in self.config.ice_servers + ] + return RTCPeerConnection(RTCConfiguration(iceServers=ice_servers)) + + async def stop(self) -> None: + if self.session_id: + try: + await self._rpc.request( + "telnyx_rtc.bye", + { + "sessid": self.session_id, + "dialogParams": {"callID": self.call_id}, + }, + ) + except Exception as exc: + logger.debug("Ignoring Telnyx bye error: %s", exc) + + if self._remote_audio_task is not None: + self._remote_audio_task.cancel() + try: + await self._remote_audio_task + except asyncio.CancelledError: + pass + self._remote_audio_task = None + + if self._peer is not None: + await self._peer.close() + self._peer = None + + await self._rpc.close() + self.disconnected_event.set() + + async def send_audio(self, pcm_16k: bytes) -> None: + if self._audio_track is None or not pcm_16k: + return + await self._audio_track.enqueue_pcm(pcm_16k) + + async def _handle_verto_request(self, message: dict[str, Any]) -> None: + method = str(message.get("method") or "") + params = message.get("params") if isinstance(message.get("params"), dict) else {} + request_id = message.get("id") + + if request_id is not None: + await self._rpc.reply(request_id, {"method": method}) + + if method in {"telnyx_rtc.media", "telnyx_rtc.answer"}: + sdp = params.get("sdp") + if isinstance(sdp, str) and sdp: + _, _, RTCSessionDescription, _ = _load_aiortc_modules() + await self._set_remote_description(RTCSessionDescription, sdp) + elif method == "telnyx_rtc.bye": + self.disconnected_event.set() + elif method == "ai_conversation": + if self._conversation_event_handler is not None: + await self._conversation_event_handler(params) + elif method == "telnyx_rtc.ping": + logger.debug("Received Telnyx Verto ping") + else: + logger.debug("Unhandled Telnyx Verto method: %s", method) + + async def _set_remote_description(self, RTCSessionDescription: Any, sdp: str) -> None: + if self._peer is None or self._remote_description_set.is_set(): + return + await self._peer.setRemoteDescription(RTCSessionDescription(sdp=sdp, type="answer")) + self._remote_description_set.set() + self._ensure_remote_audio_task() + + def _ensure_remote_audio_task(self) -> None: + if self._peer is None or self._remote_audio_task is not None: + return + for receiver in self._peer.getReceivers(): + track = getattr(receiver, "track", None) + if track is not None and getattr(track, "kind", None) == "audio": + logger.info("Telnyx WebRTC remote audio receiver attached") + self._remote_audio_task = asyncio.create_task(self._receive_remote_audio(track)) + return + + async def _wait_for_ice_gathering_complete(self, timeout_seconds: float = 10.0) -> None: + if self._peer is None or self._peer.iceGatheringState == "complete": + return + + complete = asyncio.Event() + + @self._peer.on("icegatheringstatechange") + def on_ice_gathering_state_change() -> None: + if self._peer and self._peer.iceGatheringState == "complete": + complete.set() + + try: + await asyncio.wait_for(complete.wait(), timeout=timeout_seconds) + except TimeoutError: + logger.warning("Timed out waiting for Telnyx WebRTC ICE gathering to complete; sending current SDP") + + def _build_audio_track(self, AudioFrame: Any) -> Any: + MediaStreamTrack, _, _, _ = _load_aiortc_modules() + + class _QueuedPcmAudioTrack(MediaStreamTrack): + kind = "audio" + + def __init__(self) -> None: + super().__init__() + self._queue: asyncio.Queue[bytes] = asyncio.Queue() + self._pts = 0 + self._sample_rate = TELNYX_SAMPLE_RATE + self._silence_frame = b"\x00" * int(TELNYX_SAMPLE_RATE * MULAW_CHUNK_DURATION_S * PCM_SAMPLE_WIDTH) + + async def enqueue_pcm(self, pcm_16k: bytes) -> None: + await self._queue.put(pcm_16k) + + async def recv(self) -> Any: + try: + pcm_16k = await asyncio.wait_for(self._queue.get(), timeout=MULAW_CHUNK_DURATION_S) + except TimeoutError: + pcm_16k = self._silence_frame + samples = max(1, len(pcm_16k) // PCM_SAMPLE_WIDTH) + frame = AudioFrame(format="s16", layout="mono", samples=samples) + frame.planes[0].update(pcm_16k) + frame.sample_rate = self._sample_rate + frame.pts = self._pts + frame.time_base = Fraction(1, self._sample_rate) + self._pts += samples + return frame + + return _QueuedPcmAudioTrack() + + async def _receive_remote_audio(self, track: Any) -> None: + try: + while True: + frame = await track.recv() + pcm_16k = self._frame_to_pcm16_16k(frame) + if pcm_16k: + await self.audio_handler(pcm_16k) + except asyncio.CancelledError: + pass + except Exception as exc: + logger.info("Telnyx remote audio track stopped: %s", exc) + self.disconnected_event.set() + + @staticmethod + def _frame_to_pcm16_16k(frame: Any) -> bytes: + sample_rate = int(getattr(frame, "sample_rate", TELNYX_SAMPLE_RATE) or TELNYX_SAMPLE_RATE) + try: + pcm = frame.to_ndarray().astype("int16").tobytes() + except Exception: + pcm = bytes(frame.planes[0]) + if sample_rate != TELNYX_SAMPLE_RATE: + pcm, _ = audioop.ratecv(pcm, PCM_SAMPLE_WIDTH, 1, sample_rate, TELNYX_SAMPLE_RATE, None) + return pcm + + +class TelnyxCallControlTransport: + """Telnyx Call Control transport using bidirectional media streaming.""" + + def __init__( + self, + *, + api_key: str, + to: str, + connection_id: str, + from_number: str, + conversation_id: str, + webhook_base_url: str, + api_v2_base: str, + audio_handler: Callable[[bytes], Awaitable[None]], + connect_timeout_seconds: float = 60.0, + ): + self.api_key = api_key + self.to = to + self.connection_id = connection_id + self.from_number = from_number + self.conversation_id = conversation_id + self.webhook_base_url = webhook_base_url.rstrip("/") + self.api_v2_base = api_v2_base.rstrip("/") + self.audio_handler = audio_handler + self.connect_timeout_seconds = connect_timeout_seconds + + self._session: aiohttp.ClientSession | None = None + self._stream_ws: WebSocket | None = None + self._call_control_id: str | None = None + self._call_session_id: str | None = None + self._call_leg_id: str | None = None + self._eva_call_id: str | None = None + self._stream_id: str | None = None + self._connected_event = asyncio.Event() + self._disconnected_event = asyncio.Event() + self._send_lock = asyncio.Lock() + self._inbound_encoding = "L16" + self._inbound_sample_rate = TELNYX_SAMPLE_RATE + + @property + def call_control_id(self) -> str | None: + return self._call_control_id + + @property + def call_session_id(self) -> str | None: + return self._call_session_id + + @property + def call_leg_id(self) -> str | None: + return self._call_leg_id + + @property + def eva_call_id(self) -> str | None: + return self._eva_call_id + + @property + def disconnected_event(self) -> asyncio.Event: + return self._disconnected_event + + async def start(self) -> None: + if self._session is not None: + logger.warning("Telnyx transport already started for %s", self.to) + return + + self._connected_event.clear() + self._disconnected_event.clear() + self._session = aiohttp.ClientSession( + headers={ + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json", + }, + timeout=aiohttp.ClientTimeout(total=30.0), + ) + self._eva_call_id = str(uuid.uuid4()) + payload = self.build_call_payload() + logger.info("Placing Telnyx Call Control call to %s, eva_call_id=%s", self.to, self._eva_call_id) + + response = await self._post("/calls", payload) + data = response.get("data", {}) + self._call_control_id = data.get("call_control_id") + self._call_session_id = data.get("call_session_id") + self._call_leg_id = data.get("call_leg_id") + if not self._call_control_id: + raise RuntimeError("Telnyx call creation response did not include data.call_control_id") + if not self._call_session_id: + raise RuntimeError("Telnyx call creation response did not include data.call_session_id") + + await asyncio.wait_for(self._connected_event.wait(), timeout=self.connect_timeout_seconds) + logger.info("Telnyx media stream connected for conversation %s", self.conversation_id) + + def build_call_payload(self) -> dict[str, Any]: + stream_base = self.webhook_base_url.replace("https://", "wss://").replace("http://", "ws://") + stream_wss_url = f"{stream_base}/media-stream/{self.conversation_id}" + return { + "connection_id": self.connection_id, + "to": self.to, + "from": self.from_number, + "stream_url": stream_wss_url, + "stream_track": "both_tracks", + "stream_bidirectional_mode": "rtp", + "stream_bidirectional_codec": "L16", + "stream_bidirectional_sampling_rate": TELNYX_SAMPLE_RATE, + "custom_headers": [ + { + "name": "X-Eva-Call-Id", + "value": self._eva_call_id or "", + } + ], + } + + async def stop(self) -> None: + if self._call_control_id and not self._disconnected_event.is_set(): + try: + await self._post(f"/calls/{self._call_control_id}/actions/hangup", {}) + except Exception as exc: + logger.warning("Failed to hang up Telnyx call %s: %s", self._call_control_id, exc) + + if self._stream_ws is not None: + try: + await self._stream_ws.close() + except Exception as exc: + logger.debug("Ignoring Telnyx media stream close error: %s", exc) + self._stream_ws = None + + if self._session is not None: + await self._session.close() + self._session = None + + self._connected_event.clear() + + async def send_audio(self, pcm_16k: bytes) -> None: + if not pcm_16k: + return + async with self._send_lock: + if self._stream_ws is None: + logger.debug("Dropping outbound Telnyx audio because media stream is not connected") + return + try: + outbound = self._convert_outbound_audio(pcm_16k) + await self._stream_ws.send_text( + json.dumps( + { + "event": "media", + "media": { + "payload": base64.b64encode(outbound).decode("ascii"), + }, + } + ) + ) + except Exception: + logger.info("Telnyx media stream disconnected while sending audio for %s", self.to) + + async def handle_media_stream(self, websocket: WebSocket) -> None: + self._stream_ws = websocket + self._connected_event.set() + self._disconnected_event.clear() + + try: + while True: + raw_message = await websocket.receive_text() + try: + message = json.loads(raw_message) + except json.JSONDecodeError: + logger.warning("Received non-JSON Telnyx media stream message") + continue + + event_type = message.get("event") + if event_type == "media": + payload_b64 = message.get("media", {}).get("payload") + if payload_b64: + raw_audio = base64.b64decode(payload_b64) + await self.audio_handler(self._convert_inbound_audio(raw_audio)) + elif event_type == "start": + self._stream_id = message.get("stream_id") + media_format = message.get("start", {}).get("media_format", {}) + if isinstance(media_format, dict) and media_format: + self._inbound_encoding = media_format.get("encoding", "L16") + self._inbound_sample_rate = int(media_format.get("sample_rate", TELNYX_SAMPLE_RATE)) + logger.info( + "Telnyx media stream started: stream_id=%s codec=%s@%s", + self._stream_id, + self._inbound_encoding, + self._inbound_sample_rate, + ) + elif event_type == "stop": + logger.info("Telnyx media stream stopped for conversation %s", self.conversation_id) + break + elif event_type == "connected": + logger.debug("Telnyx media stream connected event for %s", self.conversation_id) + else: + logger.debug("Unknown Telnyx media stream event: %s", event_type) + finally: + self._stream_ws = None + self._disconnected_event.set() + + def _convert_inbound_audio(self, raw_audio: bytes) -> bytes: + enc = self._inbound_encoding + rate = self._inbound_sample_rate + if enc in {"PCMU", "audio/x-mulaw"}: + pcm_audio = audioop.ulaw2lin(raw_audio, PCM_SAMPLE_WIDTH) + if rate and rate != TELNYX_SAMPLE_RATE: + pcm_audio, _ = audioop.ratecv(pcm_audio, PCM_SAMPLE_WIDTH, 1, rate, TELNYX_SAMPLE_RATE, None) + return pcm_audio + if enc in {"L16", "audio/L16"}: + if rate and rate != TELNYX_SAMPLE_RATE: + pcm_audio, _ = audioop.ratecv(raw_audio, PCM_SAMPLE_WIDTH, 1, rate, TELNYX_SAMPLE_RATE, None) + return pcm_audio + return raw_audio + return raw_audio + + def _convert_outbound_audio(self, pcm_16k: bytes) -> bytes: + if self._inbound_sample_rate >= TELNYX_SAMPLE_RATE: + return pcm_16k + converted, _ = audioop.ratecv( + pcm_16k, + PCM_SAMPLE_WIDTH, + 1, + TELNYX_SAMPLE_RATE, + self._inbound_sample_rate, + None, + ) + return converted + + async def _post(self, path: str, payload: dict[str, Any]) -> dict[str, Any]: + if self._session is None: + raise RuntimeError("Telnyx HTTP session is not initialized") + async with self._session.post(f"{self.api_v2_base}{path}", json=payload) as response: + try: + body = await response.json() + except aiohttp.ContentTypeError: + body = {"message": await response.text()} + if response.status >= 400: + raise RuntimeError(f"Telnyx API {response.status} at {path}: {json.dumps(body)}") + return body if isinstance(body, dict) else {"data": body} + + +class TelnyxAssistantServer(AbstractAssistantServer): + """Bridge EVA's Twilio-framed local simulator to a hosted Telnyx assistant.""" + + def __init__(self, **kwargs: Any) -> None: + super().__init__(**kwargs) + self._audio_sample_rate = RECORDING_SAMPLE_RATE + + self.s2s_params = self.pipeline_config.s2s_params or {} + self._api_key = self.s2s_params.get("api_key") or self.s2s_params.get("telnyx_api_key") or "" + self._api_v2_base = _normalize_api_v2_base(self.s2s_params.get("api_base")) + self._webhook_base_url = ( + self.s2s_params.get("webhook_base_url") or self.s2s_params.get("public_base_url") or "" + ).rstrip("/") + self._assistant_id = self.s2s_params.get("assistant_id") or self.s2s_params.get("assistant_agent_id") or "" + self._target_version_id = self.s2s_params.get("target_version_id") or self.s2s_params.get("targetVersionId") + self._websocket_host = ( + self.s2s_params.get("host") + or self.s2s_params.get("websocket_host") + or self.s2s_params.get("base_websocket_host") + or TELNYX_PROD_RTC_HOST + ) + self._target_params = _require_dict( + self.s2s_params.get("target_params") or self.s2s_params.get("targetParams"), + "target_params", + ) + self._user_variables = _require_dict( + self.s2s_params.get("user_variables") or self.s2s_params.get("userVariables"), + "user_variables", + ) + self._caller_name = self.s2s_params.get("caller_name") or "EVA User" + self._caller_number = ( + self.s2s_params.get("caller_number") or self.s2s_params.get("from_number") or "+15555550100" + ) + self._destination_number = self.s2s_params.get("destination_number") or "ai_assistant" + self._client_state = self.s2s_params.get("client_state") or None + self._custom_headers = _normalize_telnyx_headers(self.s2s_params.get("custom_headers")) + self._model = self.s2s_params.get("model") or TelnyxAssistantManager.DEFAULT_MODEL + self._voice = self.s2s_params.get("voice") or TelnyxAssistantManager.DEFAULT_VOICE + self._stt_model = self.s2s_params.get("stt_model") or TelnyxAssistantManager.DEFAULT_STT_MODEL + + self._transport: TelnyxWebRTCClient | TelnyxCallControlTransport | None = None + self._completed_transport: TelnyxWebRTCClient | TelnyxCallControlTransport | None = None + self._created_assistant_id: str | None = None + self._conversation_map: dict[str, str] = {} + self._transcript_enriched = False + self._user_events_enriched = False + self._stream_sid = self.conversation_id + self._user_speaking = False + self._last_user_speech_stop_ts: str | None = None + self._last_assistant_audio_monotonic: float | None = None + + async def start(self) -> None: + if self._running: + logger.warning("Telnyx server already running") + return + + self._validate_runtime_config() + self.output_dir.mkdir(parents=True, exist_ok=True) + self._fw_log = FrameworkLogWriter(self.output_dir) + self._metrics_log = MetricsLogWriter(self.output_dir) + + if self.s2s_params.get("create_assistant"): + if not self._api_key: + raise ValueError("Telnyx create_assistant requires api_key or telnyx_api_key") + manager = TelnyxAssistantManager( + self._api_key, api_base=self.s2s_params.get("api_base") or TELNYX_DEFAULT_API_BASE + ) + try: + self._created_assistant_id = await manager.create_benchmark_assistant( + agent_config=self.agent, + agent_config_path=self.agent_config_path, + webhook_base_url=self._webhook_base_url or None, + model=self._model, + voice=self._voice, + stt_model=self._stt_model, + ) + self._assistant_id = self._created_assistant_id + logger.info("Created Telnyx assistant %s", self._created_assistant_id) + finally: + await manager.close() + + self._app = FastAPI() + self._register_routes(self._app) + + config = uvicorn.Config( + self._app, + host="0.0.0.0", + port=self.port, + log_level="warning", + lifespan="off", + ) + self._server = uvicorn.Server(config) + self._running = True + self._server_task = asyncio.create_task(self._server.serve()) + + while not self._server.started: + await asyncio.sleep(0.01) + + logger.info("Telnyx server started on ws://localhost:%s", self.port) + + async def _shutdown(self) -> None: + if not self._running: + return + self._running = False + + if self._transport is not None: + await self._transport.stop() + self._completed_transport = self._transport + self._transport = None + + if self._created_assistant_id: + manager = TelnyxAssistantManager( + self._api_key, + api_base=self.s2s_params.get("api_base") or TELNYX_DEFAULT_API_BASE, + ) + try: + await manager.delete_assistant(self._created_assistant_id) + logger.info("Deleted Telnyx assistant %s", self._created_assistant_id) + except Exception: + logger.warning("Failed to delete Telnyx assistant %s", self._created_assistant_id, exc_info=True) + finally: + await manager.close() + + if self._server: + self._server.should_exit = True + if self._server_task: + try: + await asyncio.wait_for(self._server_task, timeout=5.0) + except TimeoutError: + self._server_task.cancel() + try: + await self._server_task + except asyncio.CancelledError: + pass + except (asyncio.CancelledError, KeyboardInterrupt): + pass + self._server = None + self._server_task = None + + logger.info("Telnyx server stopped on port %s", self.port) + + async def save_outputs(self) -> None: + await self._enrich_audit_log_from_external_sources() + await super().save_outputs() + + def get_conversation_stats(self) -> dict[str, Any]: + self._append_user_events_from_simulator() + return super().get_conversation_stats() + + def _validate_runtime_config(self) -> None: + missing: list[str] = [] + if not self._assistant_id and not self.s2s_params.get("create_assistant"): + missing.append("assistant_id or assistant_agent_id") + if self.s2s_params.get("create_assistant") and not self._api_key: + missing.append("api_key or telnyx_api_key") + if missing: + raise ValueError( + "Telnyx direct WebRTC mode requires s2s_params: " + + ", ".join(missing) + + ". Call Control fields such as webhook_base_url, connection_id, from_number, " + "and to/sip_uri are not required for direct mode." + ) + + def _register_routes(self, app: FastAPI) -> None: + @app.websocket("/ws") + async def websocket_endpoint(websocket: WebSocket) -> None: + await websocket.accept() + await self._handle_session(websocket) + + @app.websocket("/") + async def websocket_root(websocket: WebSocket) -> None: + await websocket.accept() + await self._handle_session(websocket) + + @app.websocket("/media-stream/{conversation_id:path}") + async def media_stream(websocket: WebSocket, conversation_id: str) -> None: + if not isinstance(self._transport, TelnyxCallControlTransport) or conversation_id != self.conversation_id: + await websocket.close(code=1008, reason="No active Telnyx transport for this conversation") + return + await websocket.accept() + try: + await self._transport.handle_media_stream(websocket) + except WebSocketDisconnect: + logger.info("Telnyx media stream WebSocket disconnected for %s", conversation_id) + + @app.post("/tools/{eva_call_id}/{tool_name}") + async def invoke_tool(eva_call_id: str, tool_name: str, request: Request) -> Any: + transport_call_id = self._completed_transport.eva_call_id if self._completed_transport else None + active_call_id = self._transport.eva_call_id if self._transport else transport_call_id + if active_call_id and eva_call_id != active_call_id: + logger.warning("Tool webhook route id mismatch: got=%s expected=%s", eva_call_id, active_call_id) + raise HTTPException(status_code=404, detail=f"Unknown call_id: {eva_call_id}") + + try: + body = await request.json() + params = _extract_tool_params(body) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + except Exception as exc: + raise HTTPException(status_code=400, detail="Invalid JSON request body") from exc + + result = await self.execute_tool(tool_name, params) + if tool_name == "end_call": + logger.info("end_call tool invoked for Telnyx call %s", eva_call_id) + asyncio.create_task(self._hangup_active_call()) + return result + + @app.post("/dynamic-variables") + async def dynamic_variables(request: Request) -> dict[str, Any]: + try: + body = await request.json() + except Exception: + body = {} + return self._handle_dynamic_variables_payload(body) + + @app.post("/call-control-events") + async def call_control_events(request: Request) -> dict[str, str]: + return {"status": "ok"} + + @app.get("/health") + async def health() -> dict[str, str]: + return {"status": "ok"} + + async def _handle_session(self, websocket: WebSocket) -> None: + logger.info("Client connected to Telnyx server") + audio_output_queue: asyncio.Queue[bytes] = asyncio.Queue() + + async def on_telnyx_audio(pcm_16k: bytes) -> None: + await self._on_telnyx_audio(pcm_16k, audio_output_queue) + + self._transport = TelnyxWebRTCClient( + config=self._build_direct_session_config(), + audio_handler=on_telnyx_audio, + ) + self._transport._conversation_event_handler = self._handle_telnyx_conversation_event + self._completed_transport = self._transport + + pacer_task = asyncio.create_task(self._pace_audio_output(websocket, audio_output_queue)) + forward_task: asyncio.Task[None] | None = None + disconnect_task: asyncio.Task[bool] | None = None + + try: + await self._transport.start() + if self._transport.eva_call_id: + logger.info("Telnyx direct WebRTC call id: %s", self._transport.eva_call_id) + + forward_task = asyncio.create_task(self._forward_user_audio(websocket, self._transport)) + disconnect_task = asyncio.create_task(self._transport.disconnected_event.wait()) + done, pending = await asyncio.wait( + [forward_task, disconnect_task, pacer_task], + return_when=asyncio.FIRST_COMPLETED, + ) + for task in pending: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + for task in done: + task_exc = task.exception() + if task is not pacer_task and task_exc is not None: + raise task_exc + except Exception as exc: + logger.error("Telnyx session error: %s", exc, exc_info=True) + finally: + if forward_task is not None and not forward_task.done(): + forward_task.cancel() + if disconnect_task is not None and not disconnect_task.done(): + disconnect_task.cancel() + pacer_task.cancel() + try: + await pacer_task + except asyncio.CancelledError: + pass + if self._transport is not None: + await self._transport.stop() + logger.info("Client disconnected from Telnyx server") + + async def _handle_telnyx_conversation_event(self, params: dict[str, Any]) -> None: + event_type = params.get("type") + item = params.get("item") if isinstance(params.get("item"), dict) else {} + if event_type != "conversation.item.created" or not item: + return + + role = item.get("role") + text = _conversation_item_text(item) + if not text: + return + + timestamp_ms = _wall_ms() + if role == "assistant": + self.audit_log.append_assistant_output(text, timestamp_ms=timestamp_ms) + if self._fw_log: + self._fw_log.llm_response(text, timestamp_ms=int(timestamp_ms)) + self._fw_log.turn_end(was_interrupted=False) + elif role == "user": + self.audit_log.append_user_input(text, timestamp_ms=timestamp_ms) + else: + logger.debug("Ignoring Telnyx conversation item role=%s text=%s", role, text[:80]) + + def _build_direct_session_config(self) -> TelnyxDirectSessionConfig: + if not self._assistant_id: + raise ValueError("Telnyx direct WebRTC mode requires assistant_id or assistant_agent_id") + return TelnyxDirectSessionConfig( + assistant_id=self._assistant_id, + host=self._websocket_host, + target_version_id=self._target_version_id, + target_params=self._target_params, + user_variables=self._user_variables, + caller_name=str(self._caller_name), + caller_number=str(self._caller_number), + destination_number=str(self._destination_number), + client_state=str(self._client_state) if self._client_state else None, + custom_headers=self._custom_headers, + ) + + def _build_telnyx_tool_definitions(self) -> list[dict[str, Any]]: + return [self._build_telnyx_tool_definition(tool) for tool in self.agent.tools] + + @staticmethod + def _build_telnyx_tool_definition(tool: AgentTool) -> dict[str, Any]: + return { + "type": "client_tool", + "name": tool.id, + "description": tool.description, + "parameters": { + "type": "object", + "properties": tool.get_parameter_properties(), + "required": tool.get_required_param_names(), + }, + } + + async def _execute_telnyx_tool_call(self, tool_name: str, arguments: dict[str, Any]) -> Any: + logger.warning("Telnyx direct client tool execution is stubbed and not yet wired to Verto events") + return await self.execute_tool(tool_name, arguments) + + async def _forward_user_audio( + self, + websocket: WebSocket, + transport: TelnyxWebRTCClient | TelnyxCallControlTransport, + ) -> None: + try: + while True: + raw = await websocket.receive_text() + try: + data = json.loads(raw) + except json.JSONDecodeError: + continue + event_type = data.get("event") + + if event_type == "start": + self._stream_sid = data.get("start", {}).get("streamSid") or self.conversation_id + continue + if event_type == "stop": + break + if event_type == "user_speech_start": + self._user_speaking = True + timestamp_ms = data.get("timestamp_ms") + if self._fw_log and _is_epoch_ms(timestamp_ms): + self._fw_log.turn_start(timestamp_ms=int(timestamp_ms)) + elif self._fw_log: + self._fw_log.turn_start() + continue + if event_type == "user_speech_stop": + self._user_speaking = False + timestamp_ms = str(data.get("timestamp_ms") or "") + self._last_user_speech_stop_ts = timestamp_ms if _is_epoch_ms(timestamp_ms) else _wall_ms() + continue + if event_type != "media": + continue + + mulaw_bytes = parse_twilio_media_message(raw) + if not mulaw_bytes: + continue + + pcm_24k = mulaw_8k_to_pcm16_24k(mulaw_bytes) + if not self._is_assistant_audio_active(): + sync_buffer_to_position(self.assistant_audio_buffer, len(self.user_audio_buffer)) + self.user_audio_buffer.extend(pcm_24k) + + pcm_16k = mulaw_8k_to_pcm16_16k(mulaw_bytes) + await transport.send_audio(pcm_16k) + except WebSocketDisconnect: + logger.debug("Telnyx local WebSocket disconnected") + except asyncio.CancelledError: + pass + + async def _on_telnyx_audio(self, pcm_16k: bytes, audio_output_queue: asyncio.Queue[bytes]) -> None: + if not pcm_16k: + return + + self._last_assistant_audio_monotonic = time.monotonic() + first_audio_wall_ms = _wall_ms() + if self._last_user_speech_stop_ts and self._metrics_log: + latency_ms = int(first_audio_wall_ms) - int(self._last_user_speech_stop_ts) + if 0 < latency_ms < 30_000: + self._metrics_log.write_latency("model_response", latency_ms / 1000, self._model) + self._last_user_speech_stop_ts = None + + pcm_24k = _pcm16_16k_to_pcm16_24k(pcm_16k) + if not self._user_speaking: + sync_buffer_to_position(self.user_audio_buffer, len(self.assistant_audio_buffer)) + self.assistant_audio_buffer.extend(pcm_24k) + + mulaw = _pcm16_16k_to_mulaw_8k(pcm_16k) + offset = 0 + while offset < len(mulaw): + chunk = mulaw[offset : offset + MULAW_CHUNK_SIZE] + offset += MULAW_CHUNK_SIZE + await audio_output_queue.put(chunk) + + async def _pace_audio_output(self, websocket: WebSocket, audio_output_queue: asyncio.Queue[bytes]) -> None: + next_send_time = time.monotonic() + try: + while True: + try: + chunk = await asyncio.wait_for(audio_output_queue.get(), timeout=1.0) + except TimeoutError: + continue + + await websocket.send_text(create_twilio_media_message(self._stream_sid, chunk)) + now = time.monotonic() + if next_send_time <= now: + next_send_time = now + next_send_time += MULAW_CHUNK_DURATION_S + sleep_duration = next_send_time - time.monotonic() + if sleep_duration > 0: + await asyncio.sleep(sleep_duration) + except asyncio.CancelledError: + pass + except Exception as exc: + logger.info("Telnyx audio pacer stopped: %s", exc) + + def _handle_dynamic_variables_payload(self, body: dict[str, Any]) -> dict[str, Any]: + payload = body.get("data", {}).get("payload", {}) if isinstance(body, dict) else {} + if not isinstance(payload, dict): + payload = {} + + eva_call_id = payload.get("eva_call_id") or (self._transport.eva_call_id if self._transport else None) + telnyx_conversation_id = payload.get("telnyx_conversation_id") + call_control_id = payload.get("call_control_id") + + logger.info( + "Telnyx dynamic variables: eva_call_id=%s conversation_id=%s call_control_id=%s", + eva_call_id, + telnyx_conversation_id, + call_control_id, + ) + + if eva_call_id and telnyx_conversation_id: + self._conversation_map[str(eva_call_id)] = str(telnyx_conversation_id) + + response: dict[str, Any] = {"dynamic_variables": {}} + if eva_call_id: + response["dynamic_variables"]["eva_call_id"] = str(eva_call_id) + metadata = {"eva_call_id": str(eva_call_id), "eva_record_id": self.conversation_id} + if self._model: + metadata["eva_llm_model"] = self._model + response["conversation"] = {"metadata": metadata} + return response + + async def _hangup_active_call(self) -> None: + if self._transport is not None: + await self._transport.stop() + + def _is_assistant_audio_active(self) -> bool: + if self._last_assistant_audio_monotonic is None: + return False + return time.monotonic() - self._last_assistant_audio_monotonic < 0.5 + + async def _enrich_audit_log_from_external_sources(self) -> None: + if self._transcript_enriched: + return + self._transcript_enriched = True + + self._append_user_events_from_simulator() + + events_path = self.output_dir / "elevenlabs_events.jsonl" + assistant_messages = await self._fetch_telnyx_intended_speech() + if not assistant_messages: + assistant_messages = extract_assistant_speech_events(events_path) + + for message in assistant_messages: + text = str(message.get("text", "")).strip() + timestamp_ms = str(message.get("timestamp_ms", "")) + if not text: + continue + self.audit_log.append_assistant_output(text, timestamp_ms=timestamp_ms) + if self._fw_log: + self._fw_log.llm_response(text, timestamp_ms=int(timestamp_ms) if timestamp_ms.isdigit() else None) + self._fw_log.turn_end(was_interrupted=False) + + def _append_user_events_from_simulator(self) -> None: + if self._user_events_enriched: + return + self._user_events_enriched = True + + for user_turn in extract_user_speech_events(self.output_dir / "elevenlabs_events.jsonl"): + timestamp_ms = user_turn.get("timestamp_ms") + self.audit_log.append_user_input(user_turn["text"], timestamp_ms=timestamp_ms) + + async def _fetch_telnyx_intended_speech(self) -> list[dict[str, Any]]: + if not self._api_key: + logger.warning("No Telnyx API key configured; skipping intended speech fetch") + return [] + + telnyx_conversation_id = self._resolve_telnyx_conversation_id() + if not telnyx_conversation_id: + logger.warning("No Telnyx conversation_id captured; skipping intended speech fetch") + return [] + + url = f"{self._api_v2_base}/ai/conversations/{telnyx_conversation_id}/messages?page[size]=100" + timeout = aiohttp.ClientTimeout(total=30.0) + try: + async with aiohttp.ClientSession( + headers={"Authorization": f"Bearer {self._api_key}"}, timeout=timeout + ) as session: + async with session.get(url) as response: + try: + payload = await response.json() + except aiohttp.ContentTypeError: + payload = {"message": await response.text()} + if response.status >= 400: + logger.warning( + "Failed to fetch Telnyx conversation messages for %s: %s %s", + telnyx_conversation_id, + response.status, + payload, + ) + return [] + except Exception as exc: + logger.warning("Failed to fetch Telnyx conversation messages for %s: %s", telnyx_conversation_id, exc) + return [] + + if not isinstance(payload, dict): + return [] + return parse_telnyx_intended_speech_payload(payload) + + def _resolve_telnyx_conversation_id(self) -> str | None: + transport = self._completed_transport or self._transport + route_id = transport.eva_call_id if transport else None + if route_id: + return self._conversation_map.get(route_id) + if len(self._conversation_map) == 1: + return next(iter(self._conversation_map.values())) + return None + + +__all__ = [ + "TelnyxAssistantManager", + "TelnyxAssistantServer", + "TelnyxCallControlTransport", + "TelnyxDirectSessionConfig", + "TelnyxVertoJsonRpcClient", + "TelnyxWebRTCClient", + "extract_assistant_speech_events", + "extract_user_speech_events", + "parse_telnyx_intended_speech_payload", +] diff --git a/src/eva/models/config.py b/src/eva/models/config.py index ea797e8a..a9674e6a 100644 --- a/src/eva/models/config.py +++ b/src/eva/models/config.py @@ -484,7 +484,7 @@ class ModelDeployment(DeploymentTypedDict): ) # Framework selection - framework: Literal["pipecat", "openai_realtime", "gemini_live", "elevenlabs", "grok_voice"] = Field( + framework: Literal["pipecat", "openai_realtime", "gemini_live", "elevenlabs", "grok_voice", "telnyx"] = Field( "pipecat", description=( "Agent framework to use for the assistant server." @@ -493,6 +493,7 @@ class ModelDeployment(DeploymentTypedDict): "'gemini_live': Gemini Live API via google-genai." "'elevenlabs': ElevenLabs Conversational AI API." "'grok_voice': xAI Grok voice realtime API." + "'telnyx': Telnyx hosted AI Assistant via direct anonymous WebRTC session." ), ) diff --git a/src/eva/orchestrator/worker.py b/src/eva/orchestrator/worker.py index 8183ff81..e026275d 100644 --- a/src/eva/orchestrator/worker.py +++ b/src/eva/orchestrator/worker.py @@ -49,10 +49,14 @@ def _get_server_class(framework: str) -> type[AbstractAssistantServer]: from eva.assistant.grok_voice_server import GrokVoiceAssistantServer return GrokVoiceAssistantServer + elif framework == "telnyx": + from eva.assistant.telnyx_server import TelnyxAssistantServer + + return TelnyxAssistantServer else: raise ValueError( f"Unknown framework: {framework!r}. " - "Supported: pipecat, openai_realtime, gemini_live, elevenlabs, grok_voice" + "Supported: pipecat, openai_realtime, gemini_live, elevenlabs, grok_voice, telnyx" ) diff --git a/tests/unit/assistant/__init__.py b/tests/unit/assistant/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/unit/assistant/test_telnyx_server.py b/tests/unit/assistant/test_telnyx_server.py new file mode 100644 index 00000000..1479ea8d --- /dev/null +++ b/tests/unit/assistant/test_telnyx_server.py @@ -0,0 +1,432 @@ +"""Unit tests for Telnyx hosted assistant helpers.""" + +import json +import time +from typing import Any, Self + +import pytest + +from eva.assistant.agentic.audit_log import AuditLog +from eva.assistant.telnyx_server import ( + TelnyxAssistantManager, + TelnyxAssistantServer, + TelnyxDirectSessionConfig, + TelnyxWebRTCClient, + _extract_tool_params, + parse_telnyx_intended_speech_payload, +) +from eva.models.agents import AgentConfig +from eva.orchestrator.worker import _get_server_class + + +class _FakeResponse: + def __init__(self, status: int, payload: dict[str, Any]): + self.status = status + self._payload = payload + + async def __aenter__(self) -> Self: + return self + + async def __aexit__(self, exc_type, exc, tb) -> None: + return None + + async def json(self) -> dict[str, Any]: + return self._payload + + async def text(self) -> str: + return json.dumps(self._payload) + + +class _FakeSession: + def __init__(self) -> None: + self.requests: list[tuple[str, str, dict[str, Any] | None]] = [] + self.closed = False + + def post(self, url: str, json: dict[str, Any]) -> _FakeResponse: + self.requests.append(("POST", url, json)) + return _FakeResponse(201, {"id": "assistant-123"}) + + def delete(self, url: str) -> _FakeResponse: + self.requests.append(("DELETE", url, None)) + return _FakeResponse(200, {"id": "assistant-123", "deleted": True}) + + async def close(self) -> None: + self.closed = True + + +def _make_agent_config() -> AgentConfig: + return AgentConfig.model_validate( + { + "id": "airline-agent", + "name": "Airline Agent", + "description": "Assist callers with airline reservations.", + "role": "Reservation specialist", + "instructions": "Verify the reservation before making changes.", + "personality": "Calm and efficient.", + "tool_module_path": "eva.assistant.tools.airline_tools", + "tools": [ + { + "id": "get_reservation", + "name": "Get Reservation", + "description": "Look up a reservation by confirmation number", + "required_parameters": [ + { + "name": "confirmation_number", + "type": "string", + "description": "The booking confirmation number", + } + ], + } + ], + } + ) + + +class TestTelnyxAssistantManager: + @pytest.mark.asyncio + async def test_create_benchmark_assistant_builds_webhook_tool_payload(self) -> None: + session = _FakeSession() + manager = TelnyxAssistantManager(api_key="test-key", session=session) # type: ignore[arg-type] + + assistant_id = await manager.create_benchmark_assistant( + agent_config=_make_agent_config(), + agent_config_path="configs/agents/airline_agent.yaml", + webhook_base_url="https://example.ngrok-free.app/", + model="telnyx-llm-gpt-4o", + voice="alloy", + ) + + assert assistant_id == "assistant-123" + assert len(session.requests) == 1 + method, url, payload = session.requests[0] + assert method == "POST" + assert url == "https://api.telnyx.com/v2/ai/assistants" + assert payload is not None + assert payload["model"] == "telnyx-llm-gpt-4o" + assert payload["voice_settings"]["voice"] == "alloy" + assert payload["telephony_settings"]["supports_unauthenticated_web_calls"] is True + + webhook_tool = payload["tools"][0] + assert webhook_tool["type"] == "webhook" + webhook = webhook_tool["webhook"] + assert webhook["name"] == "get_reservation" + assert webhook["url"] == "https://example.ngrok-free.app/tools/{{eva_call_id}}/get_reservation" + assert webhook["method"] == "POST" + assert "confirmation_number" in webhook["body_parameters"]["properties"] + assert "Role: Reservation specialist" in payload["instructions"] + assert "Calm and efficient." in payload["instructions"] + + @pytest.mark.asyncio + async def test_delete_assistant_uses_delete_endpoint(self) -> None: + session = _FakeSession() + manager = TelnyxAssistantManager(api_key="test-key", session=session) # type: ignore[arg-type] + + await manager.delete_assistant("assistant-123") + + assert session.requests == [ + ("DELETE", "https://api.telnyx.com/v2/ai/assistants/assistant-123", None), + ] + + +class TestTelnyxPayloadHelpers: + def test_webhook_tool_payload_uses_eva_call_id(self) -> None: + manager = object.__new__(TelnyxAssistantManager) + payload = manager._build_assistant_payload( + agent_config=_make_agent_config(), + agent_config_path="configs/agents/airline_agent.yaml", + webhook_base_url="https://public.example", + model="model-1", + voice="voice-1", + ) + + webhook_tools = [tool for tool in payload["tools"] if tool["type"] == "webhook"] + assert len(webhook_tools) == 1 + assert "{{eva_call_id}}" in webhook_tools[0]["webhook"]["url"] + assert "{{call_control_id}}" not in webhook_tools[0]["webhook"]["url"] + assert "hangup" in [tool["type"] for tool in payload["tools"]] + + def test_direct_assistant_payload_does_not_require_public_webhook(self) -> None: + manager = object.__new__(TelnyxAssistantManager) + payload = manager._build_assistant_payload( + agent_config=_make_agent_config(), + agent_config_path="configs/agents/airline_agent.yaml", + webhook_base_url=None, + model="model-1", + voice="voice-1", + ) + + assert payload["tools"] == [ + { + "type": "hangup", + "hangup": { + "description": ( + "To be used whenever the conversation has ended " + "and it would be appropriate to hang up the call." + ) + }, + } + ] + assert "dynamic_variables_webhook_url" not in payload + assert payload["telephony_settings"]["supports_unauthenticated_web_calls"] is True + + def test_extract_tool_params_accepts_telnyx_wrapper(self) -> None: + assert _extract_tool_params({"function_params": {"pnr": "ABC123"}}) == {"pnr": "ABC123"} + + def test_extract_tool_params_accepts_plain_body(self) -> None: + assert _extract_tool_params({"pnr": "ABC123"}) == {"pnr": "ABC123"} + + def test_parse_intended_speech_reverses_telnyx_newest_first_payload(self) -> None: + payload = { + "data": [ + { + "role": "assistant", + "text": "Second response", + "sent_at": "2026-01-01T00:00:02Z", + }, + {"role": "user", "text": "Ignored", "sent_at": "2026-01-01T00:00:01Z"}, + { + "role": "assistant", + "text": "First response", + "sent_at": "2026-01-01T00:00:00Z", + }, + ] + } + + assert parse_telnyx_intended_speech_payload(payload) == [ + {"text": "First response", "timestamp_ms": 1767225600000}, + {"text": "Second response", "timestamp_ms": 1767225602000}, + ] + + +class TestTelnyxDirectWebRTC: + def test_anonymous_login_payload_targets_ai_assistant(self) -> None: + config = TelnyxDirectSessionConfig( + assistant_id="assistant-123", + target_version_id="version-1", + target_params={"locale": "en-US"}, + user_variables={"eva_record_id": "record-1"}, + ) + + payload = TelnyxWebRTCClient.build_anonymous_login_params(config) + + assert payload["target_type"] == "ai_assistant" + assert payload["target_id"] == "assistant-123" + assert payload["target_version_id"] == "version-1" + assert payload["target_params"] == {"locale": "en-US"} + assert payload["userVariables"] == {"eva_record_id": "record-1"} + assert payload["reconnection"] is False + assert payload["User-Agent"]["sdkVersion"].startswith("EVA-TelnyxWebRTC/") + + def test_invite_payload_uses_verto_dialog_shape(self) -> None: + async def noop(_: bytes) -> None: + return None + + config = TelnyxDirectSessionConfig( + assistant_id="assistant-123", + caller_name="EVA Caller", + caller_number="+15550001000", + destination_number="ignored-by-ai-assistant", + client_state="state-1", + custom_headers=[{"name": "X-Eva-Record-Id", "value": "record-1"}], + ) + client = TelnyxWebRTCClient(config=config, audio_handler=noop) + client.session_id = "sess-123" + client.call_id = "call-123" + + payload = client.build_invite_params("v=0\r\n") + + assert payload["sessid"] == "sess-123" + assert payload["sdp"] == "v=0\r\n" + assert payload["User-Agent"].startswith("EVA-TelnyxWebRTC/") + assert payload["dialogParams"] == { + "callID": "call-123", + "caller_id_name": "EVA Caller", + "caller_id_number": "+15550001000", + "destination_number": "ignored-by-ai-assistant", + "audio": True, + "client_state": "state-1", + "custom_headers": [{"name": "X-Eva-Record-Id", "value": "record-1"}], + } + + @pytest.mark.asyncio + async def test_start_requires_aiortc_when_runtime_media_opens(self, monkeypatch) -> None: + async def noop(_: bytes) -> None: + return None + + import eva.assistant.telnyx_server as telnyx_server + + def missing_aiortc() -> tuple[Any, Any, Any, Any]: + raise ValueError("Direct Telnyx WebRTC mode requires aiortc at runtime") + + monkeypatch.setattr(telnyx_server, "_load_aiortc_modules", missing_aiortc) + client = TelnyxWebRTCClient( + config=TelnyxDirectSessionConfig(assistant_id="assistant-123"), + audio_handler=noop, + ) + + with pytest.raises(ValueError, match="requires aiortc"): + await client.start() + + +class TestTelnyxServer: + def test_dynamic_variables_returns_eva_call_id_and_records_conversation_mapping(self) -> None: + server = object.__new__(TelnyxAssistantServer) + server._conversation_map = {} + server._transport = None + server.conversation_id = "record-1" + server._model = "model-1" + + response = server._handle_dynamic_variables_payload( + { + "data": { + "payload": { + "eva_call_id": "eva-call-1", + "telnyx_conversation_id": "telnyx-conv-1", + "call_control_id": "call-control-1", + } + } + } + ) + + assert server._conversation_map == {"eva-call-1": "telnyx-conv-1"} + assert response["dynamic_variables"] == {"eva_call_id": "eva-call-1"} + assert response["conversation"]["metadata"]["eva_record_id"] == "record-1" + + def test_validate_runtime_config_requires_assistant_id_only_for_direct_mode(self) -> None: + server = object.__new__(TelnyxAssistantServer) + server._assistant_id = "" + server._api_key = "" + server.s2s_params = {} + + with pytest.raises(ValueError) as exc: + server._validate_runtime_config() + + message = str(exc.value) + assert "assistant_id or assistant_agent_id" in message + assert "webhook_base_url" in message + assert "connection_id" in message + assert "to/sip_uri" in message + + def test_validate_runtime_config_does_not_require_call_control_fields(self) -> None: + server = object.__new__(TelnyxAssistantServer) + server._assistant_id = "assistant-123" + server._api_key = "" + server.s2s_params = {} + + server._validate_runtime_config() + + def test_build_direct_session_config_accepts_aliases_and_metadata(self) -> None: + server = object.__new__(TelnyxAssistantServer) + server._assistant_id = "assistant-123" + server._websocket_host = "wss://rtcdev.telnyx.com" + server._target_version_id = "version-1" + server._target_params = {"locale": "en-US"} + server._user_variables = {"eva_record_id": "record-1"} + server._caller_name = "EVA Caller" + server._caller_number = "+15550001000" + server._destination_number = "ignored" + server._client_state = "state-1" + server._custom_headers = [{"name": "X-Eva-Record-Id", "value": "record-1"}] + + config = server._build_direct_session_config() + + assert config.assistant_id == "assistant-123" + assert config.host == "wss://rtcdev.telnyx.com" + assert config.target_version_id == "version-1" + assert config.target_params == {"locale": "en-US"} + assert config.user_variables == {"eva_record_id": "record-1"} + assert config.custom_headers == [{"name": "X-Eva-Record-Id", "value": "record-1"}] + + def test_tool_stub_maps_agent_tools_without_public_webhook(self) -> None: + server = object.__new__(TelnyxAssistantServer) + server.agent = _make_agent_config() + + tools = server._build_telnyx_tool_definitions() + + assert tools == [ + { + "type": "client_tool", + "name": "get_reservation", + "description": "Look up a reservation by confirmation number", + "parameters": { + "type": "object", + "properties": { + "confirmation_number": { + "type": "string", + "description": "The booking confirmation number", + } + }, + "required": ["confirmation_number"], + }, + } + ] + + @pytest.mark.asyncio + async def test_conversation_event_records_assistant_text(self) -> None: + server = object.__new__(TelnyxAssistantServer) + server.audit_log = AuditLog() + server._fw_log = None + + await server._handle_telnyx_conversation_event( + { + "type": "conversation.item.created", + "item": { + "role": "assistant", + "content": [{"type": "text", "text": "Hello, how can I help?"}], + }, + } + ) + + assert server.audit_log.transcript[-1]["message_type"] == "assistant" + assert server.audit_log.transcript[-1]["value"] == "Hello, how can I help?" + + def test_assistant_audio_active_uses_recent_audio_window(self) -> None: + server = object.__new__(TelnyxAssistantServer) + server._last_assistant_audio_monotonic = time.monotonic() + assert server._is_assistant_audio_active() is True + + server._last_assistant_audio_monotonic = time.monotonic() - 1.0 + assert server._is_assistant_audio_active() is False + + def test_worker_registers_telnyx_framework(self) -> None: + assert _get_server_class("telnyx") is TelnyxAssistantServer + + def test_user_event_enrichment_is_idempotent(self, tmp_path) -> None: + (tmp_path / "elevenlabs_events.jsonl").write_text( + json.dumps( + { + "timestamp": 1767225600000, + "type": "user_speech", + "data": {"text": "I need help with my reservation."}, + } + ) + + "\n", + encoding="utf-8", + ) + server = object.__new__(TelnyxAssistantServer) + server.output_dir = tmp_path + server.audit_log = AuditLog() + server._user_events_enriched = False + + server._append_user_events_from_simulator() + server._append_user_events_from_simulator() + + user_entries = [entry for entry in server.audit_log.transcript if entry["message_type"] == "user"] + assert len(user_entries) == 1 + assert user_entries[0]["value"] == "I need help with my reservation." + + @pytest.mark.asyncio + async def test_hangup_active_call_stops_transport(self) -> None: + class FakeTransport: + def __init__(self) -> None: + self.stopped = False + + async def stop(self) -> None: + self.stopped = True + + server = object.__new__(TelnyxAssistantServer) + server._transport = FakeTransport() + + await server._hangup_active_call() + + assert server._transport.stopped is True From bb59997101d802570a4f9277fa2ae78295383f34 Mon Sep 17 00:00:00 2001 From: Lucas Rosa Date: Wed, 8 Jul 2026 12:14:00 -0300 Subject: [PATCH 2/3] feat(telnyx): align default ICE servers with @telnyx/webrtc 2.27.4 Add the TURNS/443 fallback (turns:turn2.telnyx.com:443) to TELNYX_DEFAULT_ICE_SERVERS so the Telnyx EVA server's prod ICE list matches DEFAULT_PROD_ICE_SERVERS in @telnyx/webrtc 2.27.4: STUN + TURN UDP/3478 + TURN TCP/3478 + TURNS on 443. This is the last-resort TLS fallback for networks that block both 3478 transports. --- src/eva/assistant/telnyx_server.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/src/eva/assistant/telnyx_server.py b/src/eva/assistant/telnyx_server.py index 977d1f82..e65427a3 100644 --- a/src/eva/assistant/telnyx_server.py +++ b/src/eva/assistant/telnyx_server.py @@ -69,6 +69,9 @@ MULAW_CHUNK_SIZE = 160 MULAW_CHUNK_DURATION_S = 0.02 TELNYX_USER_AGENT = "EVA-TelnyxWebRTC/0.1" +# Mirrors DEFAULT_PROD_ICE_SERVERS from @telnyx/webrtc 2.27.4 +# (STUN + TURN UDP/3478 + TURN TCP/3478 + TURNS on 443). The TURNS/443 entry +# is the last-resort TLS fallback for networks that block both 3478 transports. TELNYX_DEFAULT_ICE_SERVERS = [ {"urls": "stun:stun.telnyx.com:3478"}, {"urls": "stun:stun.l.google.com:19302"}, @@ -82,6 +85,11 @@ "username": "testuser", "credential": "testpassword", }, + { + "urls": "turns:turn2.telnyx.com:443", + "username": "testuser", + "credential": "testpassword", + }, ] From 935cedce93176869ec89957414ccf50b91dce4c4 Mon Sep 17 00:00:00 2001 From: Lucas Rosa Date: Wed, 8 Jul 2026 14:50:26 -0300 Subject: [PATCH 3/3] feat(telnyx): observe WebRTC media path; fail fast when no RTP arrives MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The direct-WebRTC session could establish signaling + ICE + DTLS (iceConnectionState=completed, connectionState=connected) yet receive zero remote audio frames, leaving both audio buffers empty and the conversation silent until the far end idle-disconnected ~90s later — with no diagnostic trace. - Log iceConnectionState / connectionState transitions. - On connectionState failed|closed, set disconnected_event so the session ends immediately instead of idling. - In _receive_remote_audio, bound track.recv() with a 10s timeout and warn ("media not flowing", with iceConnectionState + frame count) instead of blocking silently; log the first received frame and frame count on stop. Diagnostic/robustness only — does not change the audio bridge behaviour when media flows. --- src/eva/assistant/telnyx_server.py | 36 ++++++++++++++++++++++++++++-- 1 file changed, 34 insertions(+), 2 deletions(-) diff --git a/src/eva/assistant/telnyx_server.py b/src/eva/assistant/telnyx_server.py index e65427a3..0c91fc51 100644 --- a/src/eva/assistant/telnyx_server.py +++ b/src/eva/assistant/telnyx_server.py @@ -716,6 +716,22 @@ def on_track(track: Any) -> None: logger.info("Telnyx WebRTC remote audio track received") self._remote_audio_task = asyncio.create_task(self._receive_remote_audio(track)) + @self._peer.on("iceconnectionstatechange") + def on_ice_connection_state_change() -> None: + logger.info( + "Telnyx WebRTC iceConnectionState=%s", + getattr(self._peer, "iceConnectionState", None), + ) + + @self._peer.on("connectionstatechange") + def on_connection_state_change() -> None: + state = getattr(self._peer, "connectionState", None) + logger.info("Telnyx WebRTC connectionState=%s", state) + # A failed/closed transport will never deliver media; surface it + # immediately instead of idling until the far end hangs up. + if state in {"failed", "closed"}: + self.disconnected_event.set() + offer = await self._peer.createOffer() await self._peer.setLocalDescription(offer) await self._wait_for_ice_gathering_complete() @@ -861,16 +877,32 @@ async def recv(self) -> Any: return _QueuedPcmAudioTrack() async def _receive_remote_audio(self, track: Any) -> None: + frames = 0 try: while True: - frame = await track.recv() + try: + frame = await asyncio.wait_for(track.recv(), timeout=10.0) + except TimeoutError: + # recv() blocks when no RTP is arriving. If the media path + # never establishes this repeats until the far end hangs up, + # producing empty audio buffers and a silent conversation. + logger.warning( + "Telnyx WebRTC: no remote audio frame in 10s " + "(iceConnectionState=%s, frames_so_far=%d) — media not flowing", + getattr(self._peer, "iceConnectionState", None), + frames, + ) + continue + frames += 1 + if frames == 1: + logger.info("Telnyx WebRTC first remote audio frame received") pcm_16k = self._frame_to_pcm16_16k(frame) if pcm_16k: await self.audio_handler(pcm_16k) except asyncio.CancelledError: pass except Exception as exc: - logger.info("Telnyx remote audio track stopped: %s", exc) + logger.info("Telnyx remote audio track stopped after %d frames: %s", frames, exc) self.disconnected_event.set() @staticmethod