"""Real Cell ARI/RTP call runner. This is intentionally a small Cell-side adapter: RabbitMQ owns the business command, while this module owns one already-authorized call's ARI channels, RTP media, AI turns, and recording. No provider credential or SIP endpoint is accepted from the command payload. """ from __future__ import annotations import base64 import contextlib import http.client import io import ipaddress import json import os import queue import re import socket import sqlite3 import ssl import struct import threading import time import uuid import wave from collections.abc import Mapping from dataclasses import dataclass, field from pathlib import Path from typing import Any from urllib.parse import urlencode, urlsplit from .ai_runtime import ( AIProviderError, ConversationEngine, pcm16_to_alaw, pcm_to_wav, ) from .core import BrokerError class CellCallError(RuntimeError): """A real Cell call failed without permitting a silent fallback.""" def __init__(self, code: str, detail: str, retryable: bool = False) -> None: super().__init__(detail) self.code = code self.detail = detail self.retryable = retryable @dataclass(frozen=True) class CellRoute: """A trusted local route binding; call payloads cannot define its values.""" route_policy_id: str caller_profile_id: str trunk_id: str caller_id: str dial_prefix: str = "" def __post_init__(self) -> None: for field_name, value in ( ("route_policy_id", self.route_policy_id), ("caller_profile_id", self.caller_profile_id), ): if not isinstance(value, str) or not re.fullmatch( r"[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}", value ): raise CellCallError("ROUTE_CONFIG_INVALID", f"{field_name} is invalid") if not isinstance(self.trunk_id, str) or not re.fullmatch( r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}", self.trunk_id ): raise CellCallError("ROUTE_CONFIG_INVALID", "trunk_id is invalid") if not isinstance(self.caller_id, str) or not re.fullmatch( r"[A-Za-z0-9_.+\- ]+(?:<[A-Za-z0-9_.+\- ]+>)?", self.caller_id ): raise CellCallError("ROUTE_CONFIG_INVALID", "caller_id is invalid") if ( not isinstance(self.dial_prefix, str) or len(self.dial_prefix) > 32 or ( self.dial_prefix and not re.fullmatch(r"[A-Za-z0-9]+", self.dial_prefix) ) ): raise CellCallError("ROUTE_CONFIG_INVALID", "dial_prefix is invalid") def load_cell_routes(raw: str | None) -> dict[str, CellRoute]: if not raw or not raw.strip(): return {} try: data = json.loads(raw) except (TypeError, json.JSONDecodeError) as exc: raise CellCallError( "ROUTE_CONFIG_INVALID", "CELL_ROUTE_MAP_JSON is invalid" ) from exc if not isinstance(data, Mapping): raise CellCallError( "ROUTE_CONFIG_INVALID", "CELL_ROUTE_MAP_JSON must be an object" ) routes: dict[str, CellRoute] = {} for route_policy_id, value in data.items(): if not isinstance(value, Mapping): raise CellCallError( "ROUTE_CONFIG_INVALID", f"route {route_policy_id!r} must be an object" ) caller_profile_id = value.get("caller_profile_id") trunk_id = value.get("trunk_id") caller_id = value.get("caller_id") dial_prefix = value.get("dial_prefix", "") if not isinstance(caller_profile_id, str): raise CellCallError( "ROUTE_CONFIG_INVALID", f"route {route_policy_id!r} has invalid caller profile", ) if not isinstance(trunk_id, str): raise CellCallError( "ROUTE_CONFIG_INVALID", f"route {route_policy_id!r} has invalid trunk" ) if not isinstance(caller_id, str): raise CellCallError( "ROUTE_CONFIG_INVALID", f"route {route_policy_id!r} has invalid caller" ) if not isinstance(dial_prefix, str): raise CellCallError( "ROUTE_CONFIG_INVALID", f"route {route_policy_id!r} has invalid prefix" ) route = CellRoute( route_policy_id=str(route_policy_id), caller_profile_id=caller_profile_id, trunk_id=trunk_id, caller_id=caller_id, dial_prefix=dial_prefix, ) routes[route.route_policy_id] = route return routes @dataclass(frozen=True) class CellCallConfig: """Trusted, Cell-local settings; none are read from ``call.execute``.""" ari_url: str ari_username: str ari_password: str ari_app: str = "agent-call" trunk_id: str = "provider-primary" caller_id: str = "BD93205882" dial_prefix: str = "7089" rtp_bind_host: str = "127.0.0.1" rtp_bind_port: int = 0 rtp_format: str = "alaw" event_timeout_s: float = 45.0 first_speech_timeout_s: float = 10.0 max_turn_s: float = 12.0 end_silence_s: float = 0.9 voice_threshold: int = 350 recording_dir: str = "/data/recordings" def __post_init__(self) -> None: parsed = urlsplit(self.ari_url) if parsed.scheme not in {"http", "https"} or not parsed.hostname: raise CellCallError( "ARI_URL_INVALID", "Cell ARI URL must be an HTTP(S) URL" ) if not self.ari_username or not self.ari_password: raise CellCallError( "ARI_CREDENTIALS_MISSING", "Cell ARI credentials are required" ) if not self.ari_app or not self.trunk_id or not self.caller_id: raise CellCallError( "CELL_CONFIG_INVALID", "Cell ARI/trunk identity is incomplete" ) if len(self.dial_prefix) > 32 or ( self.dial_prefix and not re.fullmatch(r"[A-Za-z0-9]+", self.dial_prefix) ): raise CellCallError( "DIAL_PREFIX_INVALID", "Cell dial prefix must be an ASCII alphanumeric token", ) if self.rtp_format != "alaw": raise CellCallError( "RTP_FORMAT_UNSUPPORTED", "real Cell currently requires PCMA/alaw" ) if not 0 <= self.rtp_bind_port <= 65535: raise CellCallError("RTP_PORT_INVALID", "RTP bind port is invalid") if self.event_timeout_s <= 0 or self.first_speech_timeout_s <= 0: raise CellCallError( "CELL_TIMEOUT_INVALID", "Cell timeouts must be positive" ) @classmethod def from_env(cls) -> CellCallConfig: try: rtp_bind_port = int(os.environ.get("CELL_RTP_BIND_PORT", "0")) except (TypeError, ValueError) as exc: raise CellCallError( "RTP_PORT_INVALID", "CELL_RTP_BIND_PORT is invalid" ) from exc return cls( ari_url=os.environ.get("CELL_ARI_URL", "http://127.0.0.1:8088"), ari_username=os.environ.get("CELL_ARI_USERNAME", "outbound"), ari_password=os.environ.get("CELL_ARI_PASSWORD", ""), ari_app=os.environ.get("CELL_ARI_APP", "agent-call"), trunk_id=os.environ.get("CELL_SIP_TRUNK_ID", "provider-primary"), caller_id=os.environ.get("CELL_CALLER_ID", "BD93205882"), dial_prefix=os.environ.get("CELL_DIAL_PREFIX", "7089"), rtp_bind_host=os.environ.get("CELL_RTP_BIND_HOST", "127.0.0.1"), rtp_bind_port=rtp_bind_port, recording_dir=os.environ.get("CELL_RECORDING_DIR", "/data/recordings"), ) class ARIClient: def __init__(self, config: CellCallConfig) -> None: parsed = urlsplit(config.ari_url) self.scheme = parsed.scheme self.host = parsed.hostname or "" self.port = parsed.port or (443 if self.scheme == "https" else 80) self.prefix = parsed.path.rstrip("/") or "/ari" self.username = config.ari_username self.password = config.ari_password self.timeout_s = config.event_timeout_s def request( self, method: str, resource: str, params: Mapping[str, Any] | None = None, body: bytes | None = None, content_type: str = "application/json", ) -> dict[str, Any] | bytes: path = self.prefix + "/" + resource.lstrip("/") if params: encoded = urlencode( [ (key, str(value).lower() if isinstance(value, bool) else str(value)) for key, value in params.items() ] ) path += "?" + encoded auth = base64.b64encode(f"{self.username}:{self.password}".encode()).decode( "ascii" ) connection: http.client.HTTPConnection | http.client.HTTPSConnection if self.scheme == "https": connection = http.client.HTTPSConnection( self.host, self.port, timeout=self.timeout_s, context=ssl.create_default_context(), ) else: connection = http.client.HTTPConnection( self.host, self.port, timeout=self.timeout_s ) try: connection.request( method, path, body=body, headers={ "Authorization": f"Basic {auth}", "Accept": "application/json, audio/wav", **({"Content-Type": content_type} if body is not None else {}), }, ) response = connection.getresponse() data = response.read() if response.status >= 400: raise CellCallError( f"ARI_HTTP_{response.status}", f"ARI returned HTTP {response.status}", retryable=response.status == 409 or response.status >= 500, ) if not data: return {} if "json" in (response.getheader("Content-Type") or ""): try: parsed = json.loads(data.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise CellCallError( "ARI_RESPONSE_INVALID", "ARI returned invalid JSON" ) from exc if not isinstance(parsed, dict): raise CellCallError( "ARI_RESPONSE_INVALID", "ARI returned a non-object response" ) return parsed return data except CellCallError: raise except (TimeoutError, OSError, http.client.HTTPException) as exc: raise CellCallError( "ARI_CONNECTION_FAILED", "ARI request failed", True ) from exc finally: connection.close() def events(self, app: str) -> Any: try: import websocket except ImportError as exc: # pragma: no cover - dependency gate raise CellCallError( "WEBSOCKET_CLIENT_MISSING", "websocket-client is required for ARI" ) from exc parsed_scheme = "wss" if self.scheme == "https" else "ws" url = f"{parsed_scheme}://{self.host}:{self.port}{self.prefix}/events?{urlencode({'app': app})}" auth = base64.b64encode(f"{self.username}:{self.password}".encode()).decode( "ascii" ) try: return websocket.create_connection( url, header=[f"Authorization: Basic {auth}"], timeout=self.timeout_s, http_no_proxy=[self.host, "localhost", "127.0.0.1"], ) except Exception as exc: raise CellCallError( "ARI_EVENTS_FAILED", "ARI event connection failed", True ) from exc class RTPMedia: """Minimal RTP/PCMA endpoint for an Asterisk externalMedia channel.""" def __init__(self, host: str, port: int) -> None: self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) try: self.socket.bind((host, port)) except OSError: self.socket.close() raise self.peer: tuple[str, int] | None = None self.recv_ssrc: int | None = None self.sequence = 0 self.timestamp = 0 self.ssrc = uuid.uuid4().int & 0xFFFFFFFF self._next_send_at: float | None = None self._send_lock = threading.Lock() @property def address(self) -> tuple[str, int]: value = self.socket.getsockname() try: return str(value[0]), int(value[1]) except (IndexError, TypeError, ValueError) as exc: raise CellCallError( "RTP_ADDRESS_INVALID", "RTP socket address is invalid" ) from exc def close(self) -> None: with contextlib.suppress(OSError): self.socket.close() def receive(self, timeout_s: float) -> bytes | None: self.socket.settimeout(max(0.0, timeout_s)) try: packet, peer = self.socket.recvfrom(2048) except (BlockingIOError, TimeoutError, ConnectionRefusedError): return None except OSError as exc: if self.socket.fileno() < 0: return None raise CellCallError( "RTP_RECEIVE_FAILED", "RTP receive failed", True ) from exc parsed = self._packet_info(packet) if parsed is None: return None payload, payload_type, ssrc = parsed if payload_type != 8: return None try: candidate = (str(peer[0]), int(peer[1])) except (IndexError, TypeError, ValueError) as exc: raise CellCallError( "RTP_PEER_INVALID", "RTP peer address is invalid" ) from exc if self.peer is not None and candidate != self.peer: return None if self.recv_ssrc is not None and ssrc != self.recv_ssrc: return None if self.peer is None: self.peer = candidate if self.recv_ssrc is None: self.recv_ssrc = ssrc return payload def send_pcm16(self, raw: bytes, input_rate_hz: int = 24000) -> int: if not raw or len(raw) % 2: return 0 if self.peer is None: raise CellCallError( "RTP_PEER_MISSING", "no Asterisk externalMedia RTP peer is configured" ) payload = pcm16_to_alaw(raw, input_rate_hz, 8000) sent = 0 packet_interval_s = 0.02 with self._send_lock: now = time.monotonic() if self._next_send_at is None or self._next_send_at < now - packet_interval_s: self._next_send_at = now for offset in range(0, len(payload), 160): frame = payload[offset : offset + 160] if not frame: continue delay = self._next_send_at - time.monotonic() if delay > 0: time.sleep(delay) header = struct.pack( "!BBHII", 0x80, 0x80 | 8 if offset == 0 else 8, self.sequence & 0xFFFF, self.timestamp & 0xFFFFFFFF, self.ssrc, ) try: self.socket.sendto(header + frame, self.peer) except OSError as exc: raise CellCallError( "RTP_SEND_FAILED", "RTP send failed", True ) from exc self.sequence += 1 self.timestamp += 160 sent += len(frame) self._next_send_at = max( self._next_send_at + packet_interval_s, time.monotonic() ) return sent @staticmethod def _packet_info(packet: bytes) -> tuple[bytes, int, int] | None: if len(packet) < 12 or packet[0] >> 6 != 2: return None payload_type = packet[1] & 0x7F ssrc = struct.unpack("!I", packet[8:12])[0] cc = packet[0] & 0x0F offset = 12 + cc * 4 if len(packet) < offset: return None if packet[0] & 0x10: if len(packet) < offset + 4: return None extension_words = struct.unpack("!H", packet[offset + 2 : offset + 4])[0] offset += 4 + extension_words * 4 if len(packet) < offset: return None end = len(packet) if packet[0] & 0x20: padding = packet[-1] if not padding or padding > end - offset: return None end -= padding payload = packet[offset:end] return (payload, payload_type, ssrc) if payload else None @classmethod def _payload(cls, packet: bytes) -> bytes | None: parsed = cls._packet_info(packet) return parsed[0] if parsed is not None else None def alaw_to_pcm16(payload: bytes) -> bytes: output = bytearray() for encoded in payload: value = encoded ^ 0x55 magnitude = ((value & 0x0F) << 4) + 8 exponent = (value & 0x70) >> 4 if exponent: magnitude = (magnitude + 0x100) << (exponent - 1) sample = -magnitude if value & 0x80 else magnitude try: output.extend(int(sample).to_bytes(2, "little", signed=True)) except OverflowError as exc: raise CellCallError( "RTP_AUDIO_INVALID", "decoded PCMA sample is out of range" ) from exc return bytes(output) def voice_level(raw_pcm16: bytes) -> int: if len(raw_pcm16) < 2: return 0 samples = [ int.from_bytes(raw_pcm16[index : index + 2], "little", signed=True) for index in range(0, len(raw_pcm16) - 1, 2) ] return max(abs(sample) for sample in samples) if samples else 0 @dataclass class CellCallResult: call_id: str status: str reason_code: str turns: list[dict[str, Any]] = field(default_factory=list) recording_path: str | None = None connected: bool = False media_packets_in: int = 0 media_packets_out: int = 0 started_at: float = 0.0 ended_at: float = 0.0 route_policy_id: str | None = None caller_profile_id: str | None = None trunk_id: str | None = None effective_target: str | None = None def as_dict(self) -> dict[str, Any]: return { "call_id": self.call_id, "status": self.status, "reason_code": self.reason_code, "turns": self.turns, "recording_path": self.recording_path, "connected": self.connected, "media_packets_in": self.media_packets_in, "media_packets_out": self.media_packets_out, "duration_ms": max(0.0, (self.ended_at - self.started_at) * 1000), "route_policy_id": self.route_policy_id, "caller_profile_id": self.caller_profile_id, "trunk_id": self.trunk_id, "effective_target": self.effective_target, } class RealCellCall: """Execute one authorized call using a trusted Cell-local route binding.""" def __init__( self, config: CellCallConfig, engine: ConversationEngine, asr: Any ) -> None: self.config = config self.engine = engine self.asr = asr self.ari = ARIClient(config) self.call_id = f"call_{uuid.uuid4().hex}" self.events: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=256) self.event_stop = threading.Event() self.event_thread: threading.Thread | None = None self.ws: Any | None = None self.media: RTPMedia | None = None self.bridge_id = f"bridge_{self.call_id}" self.external_channel_id: str | None = None self.target_channel_id: str | None = None self.recording_name = f"agent-call-{self.call_id}" self._known_channels: set[str] = set() self._prebuffer: list[bytes] = [] self._call_ended = False self._in_packets = 0 self._out_bytes = 0 def _reset_call(self) -> None: self.call_id = f"call_{uuid.uuid4().hex}" self.event_stop = threading.Event() self.event_thread = None self.ws = None self.media = None self.bridge_id = f"bridge_{self.call_id}" self.external_channel_id = None self.target_channel_id = None self.recording_name = f"agent-call-{self.call_id}" self._known_channels.clear() self._prebuffer.clear() self._call_ended = False self._in_packets = 0 self._out_bytes = 0 def start_authorized_call( self, callee: str, route: CellRoute | None = None ) -> CellCallResult: if not callee.isdigit() or not 3 <= len(callee) <= 32: raise CellCallError( "CALLEE_INVALID", "callee must be the original digit string" ) selected_route = route or CellRoute( "legacy", "legacy", self.config.trunk_id, self.config.caller_id, self.config.dial_prefix, ) self._reset_call() self.engine.history.clear() result = CellCallResult( self.call_id, "failed", "CELL_EXECUTOR_EXCEPTION", started_at=time.monotonic(), route_policy_id=selected_route.route_policy_id, caller_profile_id=selected_route.caller_profile_id, trunk_id=selected_route.trunk_id, effective_target=selected_route.dial_prefix + callee, ) try: self.media = RTPMedia(self.config.rtp_bind_host, self.config.rtp_bind_port) self.ws = self.ari.events(self.config.ari_app) self._start_event_reader() self.ari.request("POST", f"bridges/{self.bridge_id}", {"type": "mixing"}) external = self.ari.request( "POST", "channels/externalMedia", { "app": self.config.ari_app, "external_host": f"{self.media.address[0]}:{self.media.address[1]}", "format": self.config.rtp_format, "direction": "both", "channelId": f"external_{self.call_id}", }, ) self.external_channel_id = self._channel_id(external) self._try_set_external_media_peer() self._add_when_ready(self.external_channel_id) target = self.ari.request( "POST", "channels", { "endpoint": f"PJSIP/{selected_route.dial_prefix + callee}@{selected_route.trunk_id}", "app": self.config.ari_app, "appArgs": self.call_id, "callerId": selected_route.caller_id, "timeout": 30, "channelId": f"target_{self.call_id}", }, ) self.target_channel_id = self._channel_id(target) self._add_when_ready(self.target_channel_id) self._wait_answer(result) result.connected = True self.ari.request( "POST", f"bridges/{self.bridge_id}/record", { "name": self.recording_name, "format": "wav", "ifExists": "fail", "beepEnabled": False, }, ) opening = str(self.engine.config["conversation"].get("opening", "")) if opening: opening_result = self.engine.speak(opening, f"opening_{self.call_id}") self._play(opening_result["audio"]) result.media_packets_out = self._out_bytes // 160 max_turns = int(self.engine.config["conversation"]["max_turns"]) for turn_index in range(max_turns): captured = self._capture_turn( self.config.first_speech_timeout_s if turn_index == 0 else self.config.end_silence_s * 2 ) if self._call_ended: result.reason_code = "CUSTOMER_HANGUP" result.status = "completed" if result.turns else "failed" break if not captured: result.reason_code = ( "CUSTOMER_SILENT" if turn_index == 0 else "CONVERSATION_COMPLETE" ) result.status = "completed" if turn_index else "failed" break wav = pcm_to_wav(captured, 8000) segments = list( self.asr.transcribe( wav, self.engine.config, f"turn_{turn_index}_{self.call_id}" ) ) final = next( (item for item in reversed(segments) if item.get("is_final")), None ) if ( not isinstance(final, Mapping) or not isinstance(final.get("text"), str) or not final["text"] ): raise CellCallError( "ASR_NO_FINAL", "ASR did not produce a final telephone turn" ) text = str(final["text"]) response = self.engine.run_text(text) if response.get("status") != "completed": raise CellCallError( str(response.get("reason_code", "AI_GENERATION_FAILED")), "AI turn failed", ) self._play(response.get("audio", b"")) result.media_packets_out = self._out_bytes // 160 result.turns.append( { "turn_index": turn_index, "customer_text": text, "assistant_text": response.get("text", ""), "asr_segments": segments, "llm_first_token_ms": response.get("llm_first_token_ms"), "tts_first_audio_ms": response.get("tts_first_audio_ms"), } ) else: result.status = "completed" result.reason_code = "MAX_TURNS" if ( result.status == "failed" and result.reason_code == "CELL_EXECUTOR_EXCEPTION" ): result.reason_code = ( "CALL_COMPLETED" if result.turns else result.reason_code ) except (CellCallError, AIProviderError) as exc: result.status = "failed" result.reason_code = ( exc.code if isinstance(exc, (CellCallError, AIProviderError)) else "AI_PROVIDER_ERROR" ) except (OSError, RuntimeError, TypeError, ValueError, KeyError) as exc: del exc result.status = "failed" result.reason_code = "CELL_EXECUTOR_EXCEPTION" finally: result.ended_at = time.monotonic() result.media_packets_in = self._in_packets result.media_packets_out = self._out_bytes // 160 result.recording_path = self._finish_recording() self._cleanup() return result def _start_event_reader(self) -> None: self.event_thread = threading.Thread( target=self._read_events, name=f"ari-events-{self.call_id}", daemon=True ) self.event_thread.start() def _read_events(self) -> None: try: import websocket except ImportError: return while not self.event_stop.is_set() and self.ws is not None: try: raw = self.ws.recv() except ( OSError, RuntimeError, TimeoutError, ValueError, websocket.WebSocketException, ): return if not raw: continue try: event = json.loads(raw) except (TypeError, json.JSONDecodeError): continue if not isinstance(event, dict): continue try: self.events.put_nowait(event) except queue.Full: return def _wait_answer(self, result: CellCallResult) -> None: deadline = time.monotonic() + self.config.event_timeout_s answered = False expected_channels = { channel_id for channel_id in (self.external_channel_id, self.target_channel_id) if channel_id } while time.monotonic() < deadline: try: event = self.events.get(timeout=min(0.5, deadline - time.monotonic())) except queue.Empty: event = None if event is not None: event_type = event.get("type") channel = event.get("channel") or {} channel_id = channel.get("id") if ( event_type == "StasisStart" and isinstance(channel_id, str) and channel_id in expected_channels ): if channel_id == self.external_channel_id: self._try_set_external_media_peer() self._add_when_ready(channel_id) if ( event_type == "ChannelStateChange" and channel_id == self.target_channel_id and channel.get("state") == "Up" ): answered = True if ( event_type in {"ChannelDestroyed", "StasisEnd", "ChannelHangupRequest"} and channel_id == self.target_channel_id ): self._call_ended = True raise CellCallError( "CALL_NOT_ANSWERED", "target channel ended before answer" ) if answered: self._try_set_external_media_peer() try: bridge_channels = self._bridge_channels() except CellCallError as exc: if exc.code not in {"ARI_HTTP_404", "ARI_HTTP_409"}: raise else: if ( expected_channels.issubset(bridge_channels) and self.media is not None and self.media.peer is not None ): return if answered: raise CellCallError( "ARI_BRIDGE_NOT_READY", "answered channels did not join the expected bridge", True, ) raise CellCallError( "CALL_ANSWER_TIMEOUT", "ARI did not report an answered channel", True ) def _capture_turn(self, initial_timeout_s: float) -> bytes: if self.media is None: raise CellCallError("RTP_UNAVAILABLE", "RTP media is not initialized") started = bool(self._prebuffer) last_voice = time.monotonic() if started else 0.0 started_at = time.monotonic() deadline = started_at + max(initial_timeout_s, self.config.end_silence_s) frames = self._prebuffer self._prebuffer = [] while time.monotonic() < deadline: if self._drain_terminal_events(): break payload = self.media.receive( min(0.1, max(0.0, deadline - time.monotonic())) ) if payload is None: if self._drain_terminal_events(): break if ( started and time.monotonic() - last_voice >= self.config.end_silence_s ): break continue self._in_packets += 1 pcm = alaw_to_pcm16(payload) level = voice_level(pcm) if level >= self.config.voice_threshold: if not started: started = True deadline = min( started_at + self.config.max_turn_s, time.monotonic() + self.config.max_turn_s, ) last_voice = time.monotonic() if started: frames.append(pcm) if time.monotonic() - last_voice >= self.config.end_silence_s: break if self.media is not None: self.media.socket.settimeout(None) return b"".join(frames) def _drain_terminal_events(self) -> bool: while True: try: event = self.events.get_nowait() except queue.Empty: return self._call_ended event_type = event.get("type") channel_id = (event.get("channel") or {}).get("id") if ( event_type in {"ChannelDestroyed", "StasisEnd", "ChannelHangupRequest"} and channel_id == self.target_channel_id ): self._call_ended = True return True def _play(self, audio: bytes) -> int: if self.media is None or not audio: return 0 sent = 0 for offset in range(0, len(audio), 2 * 24000 // 50): frame = audio[offset : offset + 2 * 24000 // 50] if not frame: continue if self.engine.config["conversation"].get("allow_interrupt"): self.media.socket.settimeout(0.0) with contextlib.suppress(BlockingIOError, socket.timeout): packet = self.media.receive(0.0) if packet: self._in_packets += 1 pcm = alaw_to_pcm16(packet) if voice_level(pcm) >= self.config.voice_threshold: self._prebuffer.append(pcm) break bytes_sent = self.media.send_pcm16(frame, 24000) sent += bytes_sent self._out_bytes += bytes_sent return sent def _try_set_external_media_peer(self) -> bool: if self.external_channel_id is None: return False try: self._set_external_media_peer(self.external_channel_id) except CellCallError as exc: if exc.code in { "ARI_HTTP_400", "ARI_HTTP_404", "ARI_HTTP_409", "RTP_EXTERNAL_PEER_UNAVAILABLE", }: return False raise return True def _set_external_media_peer(self, channel_id: str) -> None: if self.media is None: raise CellCallError("RTP_UNAVAILABLE", "RTP media is not initialized") values: dict[str, str] = {} for variable in ("UNICASTRTP_LOCAL_ADDRESS", "UNICASTRTP_LOCAL_PORT"): response = self.ari.request( "GET", f"channels/{channel_id}/variable", {"variable": variable}, ) value = response.get("value") if isinstance(response, dict) else None if not isinstance(value, str) or not value: raise CellCallError( "RTP_EXTERNAL_PEER_UNAVAILABLE", f"ARI did not provide {variable}", True, ) values[variable] = value try: address = ipaddress.ip_address(values["UNICASTRTP_LOCAL_ADDRESS"]) port = int(values["UNICASTRTP_LOCAL_PORT"]) except (ValueError, TypeError) as exc: raise CellCallError( "RTP_EXTERNAL_PEER_INVALID", "ARI external RTP peer is invalid" ) from exc if address.is_unspecified or address.is_multicast or not 1 <= port <= 65535: raise CellCallError( "RTP_EXTERNAL_PEER_INVALID", "ARI external RTP peer is invalid" ) self.media.peer = (str(address), port) def _bridge_channels(self) -> set[str]: response = self.ari.request("GET", f"bridges/{self.bridge_id}") channels = response.get("channels") if isinstance(response, dict) else None if not isinstance(channels, list) or not all( isinstance(channel_id, str) for channel_id in channels ): raise CellCallError( "ARI_BRIDGE_INVALID", "ARI did not return bridge channels" ) return set(channels) def _add_when_ready(self, channel_id: str | None) -> bool: if not channel_id or channel_id in self._known_channels: return channel_id in self._known_channels for attempt in range(3): try: self.ari.request( "POST", f"bridges/{self.bridge_id}/addChannel", {"channel": channel_id}, ) except CellCallError as exc: if exc.code not in { "ARI_HTTP_400", "ARI_HTTP_404", "ARI_HTTP_409", "ARI_HTTP_422", }: raise if attempt == 2: return False time.sleep(0.05 * (attempt + 1)) else: self._known_channels.add(channel_id) return True return False @staticmethod def _channel_id(value: dict[str, Any] | bytes) -> str: channel_id = value.get("id") if isinstance(value, dict) else None if not isinstance(channel_id, str) or not channel_id: raise CellCallError( "ARI_CHANNEL_INVALID", "ARI did not return a channel ID" ) return channel_id def _finish_recording(self) -> str | None: try: try: self.ari.request("POST", f"recordings/live/{self.recording_name}/stop") except CellCallError as exc: if exc.code != "ARI_HTTP_404": return None deadline = time.monotonic() + min(self.config.event_timeout_s, 5.0) while True: try: raw = self.ari.request( "GET", f"recordings/stored/{self.recording_name}/file" ) break except CellCallError as exc: if exc.code != "ARI_HTTP_404": return None if time.monotonic() >= deadline: return None time.sleep(0.1) except CellCallError: return None if not isinstance(raw, bytes) or not raw: return None try: with wave.open(io.BytesIO(raw), "rb") as source: frames = source.getnframes() except (EOFError, OSError, wave.Error): return None if not frames > 0: return None path = Path(self.config.recording_dir) / f"{self.recording_name}.wav" try: if path.is_symlink(): return None path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(raw) return str(path) except OSError: return None def _cleanup(self) -> None: self.event_stop.set() for channel_id in (self.target_channel_id, self.external_channel_id): if channel_id: with contextlib.suppress(CellCallError): self.ari.request("DELETE", f"channels/{channel_id}") with contextlib.suppress(CellCallError): self.ari.request("DELETE", f"bridges/{self.bridge_id}") if self.ws is not None: with contextlib.suppress(Exception): self.ws.close() if self.event_thread is not None: self.event_thread.join(timeout=1.0) if self.media is not None: self.media.close() class CellExecutionLedger: """Durable claim/result ledger preventing a post-dial redial.""" def __init__(self, path: str | Path) -> None: self.path = Path(path) if self.path.exists() and self.path.is_symlink(): raise CellCallError( "LEDGER_PATH_INVALID", "execution ledger must not be a symlink" ) self.path.parent.mkdir(parents=True, exist_ok=True) with self._connect() as connection: connection.execute( """ CREATE TABLE IF NOT EXISTS cell_executions ( execution_id TEXT PRIMARY KEY, callee TEXT NOT NULL, status TEXT NOT NULL, result_json TEXT, updated_at REAL NOT NULL ) """ ) def _connect(self) -> sqlite3.Connection: connection = sqlite3.connect(self.path, timeout=5) connection.row_factory = sqlite3.Row return connection def claim(self, execution_id: str, callee: str) -> dict[str, Any]: now = time.time() with self._connect() as connection: connection.execute("BEGIN IMMEDIATE") row = connection.execute( "SELECT status, result_json FROM cell_executions WHERE execution_id=?", (execution_id,), ).fetchone() if row is None: connection.execute( "INSERT INTO cell_executions(execution_id, callee, status, updated_at) VALUES (?, ?, 'in_progress', ?)", (execution_id, callee, now), ) return {"claimed": True, "in_doubt": False, "result": None} status = str(row["status"]) if status == "in_progress": return {"claimed": False, "in_doubt": True, "result": None} try: result = json.loads(row["result_json"]) if row["result_json"] else None except (TypeError, json.JSONDecodeError) as exc: raise CellCallError( "LEDGER_CORRUPT", "execution ledger contains invalid JSON" ) from exc return {"claimed": False, "in_doubt": False, "result": result} def complete(self, execution_id: str, result: Mapping[str, Any]) -> None: encoded = json.dumps(dict(result), ensure_ascii=False, sort_keys=True) with self._connect() as connection: connection.execute( "UPDATE cell_executions SET status='completed', result_json=?, updated_at=? WHERE execution_id=?", (encoded, time.time(), execution_id), ) def result(self, execution_id: str) -> dict[str, Any] | None: with self._connect() as connection: row = connection.execute( "SELECT result_json FROM cell_executions WHERE execution_id=?", (execution_id,), ).fetchone() if row is None or not row["result_json"]: return None try: value = json.loads(row["result_json"]) except (TypeError, json.JSONDecodeError) as exc: raise CellCallError( "LEDGER_CORRUPT", "execution ledger contains invalid JSON" ) from exc if not isinstance(value, dict): raise CellCallError( "LEDGER_CORRUPT", "execution ledger result is not an object" ) return value class RealCellWorker: """Consume one tenant's ``call.execute`` queue and run a Cell executor.""" def __init__( self, broker: Any, tenant_key: str, ledger: CellExecutionLedger, executor: RealCellCall, routes: Mapping[str, CellRoute] | None = None, ) -> None: if not tenant_key: raise CellCallError("TENANT_KEY_INVALID", "tenant_key is required") self.broker = broker self.tenant_key = tenant_key self.queue_name = f"agent-call.executor.{tenant_key}.v1" self.ledger = ledger self.executor = executor self.routes = dict(routes or {}) self.broker.declare_tenant(tenant_key) def process_once(self) -> dict[str, Any] | None: message = self.broker.consume(self.queue_name) if message is None: return None body = message.get("body") if isinstance(message, Mapping) else None try: ( execution_id, tenant_id, agent_version_id, callee, route_policy_id, caller_profile_id, ) = self._command(body) except CellCallError: self.broker.reject(message) return {"status": "rejected", "reason_code": "COMMAND_INVALID"} try: route = self._route(route_policy_id, caller_profile_id) except CellCallError as exc: self.broker.reject(message) return {"status": "rejected", "reason_code": exc.code} claim = self.ledger.claim(execution_id, callee) if claim["in_doubt"]: self.broker.ack(message) return {"status": "reconciled", "reason_code": "EXECUTION_IN_DOUBT"} result = claim["result"] if claim["claimed"]: try: if agent_version_id != self.executor.engine.config["agent_version_id"]: raise CellCallError( "AGENT_VERSION_UNAVAILABLE", "agent version is not installed on this Cell", ) result = ( self.executor.start_authorized_call(callee, route) if route is not None else self.executor.start_authorized_call(callee) ).as_dict() except ( CellCallError, AIProviderError, OSError, RuntimeError, TypeError, ValueError, KeyError, ) as exc: result = { "call_id": f"call_{uuid.uuid4().hex}", "status": "failed", "reason_code": getattr(exc, "code", "CELL_EXECUTOR_EXCEPTION"), "connected": False, "turns": [], } self.ledger.complete(execution_id, result) event = self._event(execution_id, tenant_id, callee, result or {}) try: self.broker.publish( "agent-call.events.v1", "agent-call.call.finished", event, message_id=event["event_id"], ) except (BrokerError, OSError, RuntimeError, TypeError, ValueError, KeyError): self.broker.requeue(message) return {"status": "pending", "reason_code": "EVENT_PUBLISH_FAILED"} self.broker.ack(message) return event def _route(self, route_policy_id: str, caller_profile_id: str) -> CellRoute | None: if not route_policy_id and not caller_profile_id: if self.routes: raise CellCallError( "ROUTE_POLICY_REQUIRED", "route policy is required on a multi-trunk Cell", ) return None if not route_policy_id or not caller_profile_id: raise CellCallError( "ROUTE_POLICY_INVALID", "route policy and caller profile must be paired" ) if not self.routes: if (route_policy_id, caller_profile_id) == ( "route_policy_test", "caller_profile_test", ): return None raise CellCallError( "ROUTE_POLICY_UNAVAILABLE", "route policy is not installed on this Cell" ) route = self.routes.get(route_policy_id) if route is None: raise CellCallError( "ROUTE_POLICY_UNAVAILABLE", "route policy is not installed on this Cell" ) if route.caller_profile_id != caller_profile_id: raise CellCallError( "CALLER_PROFILE_MISMATCH", "caller profile is not bound to the route policy", ) return route def _command(self, body: Any) -> tuple[str, str, str, str, str, str]: if not isinstance(body, Mapping) or body.get("command_type") != "call.execute": raise CellCallError("COMMAND_INVALID", "message is not call.execute") if body.get("tenant_key") != self.tenant_key: raise CellCallError( "TENANT_ROUTE_MISMATCH", "message tenant_key does not match queue" ) payload = body.get("payload") if not isinstance(payload, Mapping): raise CellCallError("COMMAND_INVALID", "call.execute payload is required") values = ( body.get("tenant_id"), payload.get("execution_id"), payload.get("agent_version_id"), payload.get("callee"), ) route_policy_id = payload.get("route_policy_id", "") caller_profile_id = payload.get("caller_profile_id", "") if not isinstance(route_policy_id, str) or not isinstance( caller_profile_id, str ): raise CellCallError( "COMMAND_INVALID", "route policy and caller profile must be strings" ) if bool(route_policy_id) != bool(caller_profile_id): raise CellCallError( "COMMAND_INVALID", "route policy and caller profile must be paired" ) if not all(isinstance(value, str) and value for value in values): raise CellCallError( "COMMAND_INVALID", "call.execute identity fields are required" ) return ( str(values[1]), str(values[0]), str(values[2]), str(values[3]), route_policy_id, caller_profile_id, ) @staticmethod def _event( execution_id: str, tenant_id: str, callee: str, result: Mapping[str, Any] ) -> dict[str, Any]: payload = dict(result) payload.pop("recording_path", None) payload["callee"] = callee payload["execution_id"] = execution_id return { "schema_version": "1.0", "event_type": "call.finished", "event_id": f"{execution_id}:call.finished", "tenant_id": tenant_id, "payload": payload, } __all__ = [ "ARIClient", "CellCallConfig", "CellCallError", "CellCallResult", "CellExecutionLedger", "CellRoute", "RTPMedia", "RealCellCall", "RealCellWorker", "alaw_to_pcm16", "load_cell_routes", "voice_level", ]