"""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 json import os import queue import socket import sqlite3 import ssl import struct import threading import time import uuid 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 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" 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 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"), 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.sequence = 0 self.timestamp = 0 self.ssrc = uuid.uuid4().int & 0xFFFFFFFF 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 (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 payload = self._payload(packet) if payload is None: return None try: self.peer = (str(peer[0]), int(peer[1])) except (IndexError, TypeError, ValueError) as exc: raise CellCallError( "RTP_PEER_INVALID", "RTP peer address is invalid" ) from exc 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 inbound RTP peer has been observed" ) payload = pcm16_to_alaw(raw, input_rate_hz, 8000) sent = 0 with self._send_lock: for offset in range(0, len(payload), 160): frame = payload[offset : offset + 160] if not frame: continue 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) time.sleep(0.02) return sent @staticmethod def _payload(packet: bytes) -> bytes | None: if len(packet) < 12 or packet[0] >> 6 != 2: return None 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 return packet[offset:end] or 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 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), } class RealCellCall: """Execute one authorized call using a fixed Cell-local trunk.""" 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) -> CellCallResult: if not callee.isdigit() or not 3 <= len(callee) <= 32: raise CellCallError( "CALLEE_INVALID", "callee must be the original digit string" ) self._reset_call() self.engine.history.clear() result = CellCallResult( self.call_id, "failed", "CELL_EXECUTOR_EXCEPTION", started_at=time.monotonic(), ) 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._add_when_ready(self.external_channel_id) target = self.ari.request( "POST", "channels", { "endpoint": f"PJSIP/{'7089' + callee}@{self.config.trunk_id}", "app": self.config.ari_app, "appArgs": self.call_id, "callerId": self.config.caller_id, "timeout": 30000, "channelId": f"target_{self.call_id}", }, ) self.target_channel_id = self._channel_id(target) self._add_when_ready(self.target_channel_id) self.ari.request( "POST", f"bridges/{self.bridge_id}/record", { "name": self.recording_name, "format": "wav", "ifExists": "fail", "beepEnabled": False, }, ) self._wait_answer(result) result.connected = True 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 while time.monotonic() < deadline: try: event = self.events.get(timeout=min(0.5, deadline - time.monotonic())) except queue.Empty: continue event_type = event.get("type") channel = event.get("channel") or {} channel_id = channel.get("id") if event_type == "StasisStart" and isinstance(channel_id, str): self._add_when_ready(channel_id) if ( event_type == "ChannelStateChange" and channel_id == self.target_channel_id and channel.get("state") == "Up" ): return 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" ) 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 _add_when_ready(self, channel_id: str | None) -> None: if not channel_id or channel_id in self._known_channels: return self._known_channels.add(channel_id) with contextlib.suppress(CellCallError): self.ari.request( "POST", f"bridges/{self.bridge_id}/addChannel", {"channel": channel_id}, ) @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: with contextlib.suppress(CellCallError): self.ari.request("DELETE", f"recordings/live/{self.recording_name}") try: raw = self.ari.request( "GET", f"recordings/stored/{self.recording_name}/file" ) except CellCallError: return None if not isinstance(raw, bytes) or not raw: 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, ) -> 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.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 = self._command(body) except CellCallError: self.broker.reject(message) return {"status": "rejected", "reason_code": "COMMAND_INVALID"} 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).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 _command(self, body: Any) -> tuple[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"), ) 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]) @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", "RTPMedia", "RealCellCall", "RealCellWorker", "alaw_to_pcm16", "voice_level", ]