""" FIXED Call Session Management for Concurrent Calls Each call gets completely isolated services and state with proper database handling """ import asyncio import json import time from datetime import datetime from typing import Optional, Dict, Any, List from fastapi import WebSocket from fastapi.websockets import WebSocketDisconnect from config import Config from services.log_utils import Log from services.connection_manager import WebSocketConnectionManager from services.audio_service import AudioService from services.bot_configuration_service import BotConfigurationService class CallSession: """ Completely isolated session for each concurrent call. Each call gets its own instances of all services with proper database isolation. """ def __init__(self, session_id: str, websocket: WebSocket): self.session_id = session_id self.websocket = websocket self.bot_configuration_service = BotConfigurationService() self.connection_manager = WebSocketConnectionManager(websocket) self.audio_service = AudioService() self.service_type = 4 self.elevenlabs_websocket_service = None self.elevenlabs_receiver_task = None self.elevenlabs_last_reconnect_time = 0 self.elevenlabs_reconnect_cooldown = 2 # Reduced from 5s to 2s for faster recovery self.elevenlabs_reconnect_attempts = 0 self.elevenlabs_audio_queue = asyncio.Queue(maxsize=100) # Queue to buffer audio during reconnection self.elevenlabs_audio_queue_task = None # Task to process queued audio self.elevenlabs_connection_retry_task = None # Task to retry connection when limit reached self.elevenlabs_conversation_id: Optional[str] = None # Store conversation_id in CallSession for safety (prevents loss if service is recreated) self.elevenlabs_initializing = False # Flag to prevent connection checks during initialization self.elevenlabs_audio_playing = False self.last_interrupt_time = 0 self.interrupt_debounce_seconds = 0.3 self.vad_score_history = [] self.vad_history_duration = 0.25 self.min_sustained_score = 0.45 self.min_sustained_duration = 0.15 self.business_id: Optional[str] = None self.call_id: Optional[str] = None self.stream_id: Optional[str] = None self.agent_id: Optional[str] = None self.phone_number: Optional[str] = None self.did: Optional[str] = None self.customer_name: Optional[str] = None self.bot_config: Optional[Dict[str, Any]] = None self.bot_name: Optional[str] = None self.bot_id: Optional[int] = None self.company_name: Optional[str] = None self.call_start_time: Optional[datetime] = None self.is_active = True self.call_ended = False # Flag to track if call has legitimately ended (end_call tool or voicemail) self.pending_end_call_tool_id = None # Track pending end_call tool to detect abandonment self.pending_transfer_tool_id = None # Track pending transfer tool to detect abandonment self.pending_transfer_target = None # Track transfer target for logging self._last_connection_warning_time = 0 # Rate limit connection state warnings self._last_call_ended_log_time = 0 # Rate limit "call already ended" logs self.conversation_count = 0 self.user_messages: List[Dict[str, Any]] = [] self.bot_messages: List[Dict[str, Any]] = [] self.max_messages = 100 # Limit message history to prevent memory issues Log.info(f"🆕 Created isolated call session: {session_id}") async def initialize(self) -> bool: """Initialize the call session with all required connections.""" try: Log.info(f"✅ Session {self.session_id} initialized (service_type 4 - ElevenLabs WebSocket)") return True except Exception as e: Log.error(f"❌ Failed to initialize session {self.session_id}: {e}") return False # async def _fetch_customer_data(self, phone_number: str) -> dict: # """Fetch customer data from external API with connection pooling.""" # if not phone_number or phone_number == 'Not provided': # return {} # try: # Log.info(f"🔍 Fetching customer data for {phone_number}...") # # Use shared connector with connection pooling to avoid connection exhaustion # connector = aiohttp.TCPConnector(limit=100, limit_per_host=20) # timeout = aiohttp.ClientTimeout(total=5.0) # async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session: # payload = {"caller_id": phone_number} # Changed from phone_number to caller_id # async with session.post("https://devapp.syntheon.in/contacts_api.php", json=payload) as response: # if response.status == 200: # data = await response.json() # Log.info(f"✅ Fetched customer data: {data}") # return data.get("dynamic_variables", {}) # else: # Log.warning(f"⚠️ Failed to fetch customer data: Status {response.status}") # return {} # except Exception as e: # Log.error(f"❌ Error fetching customer data: {e}") # return {} async def handle_call_start(self, data: dict) -> None: """Handle call start event with business context - OPTIMIZED for concurrent calls.""" try: start_info = data.get('start', {}) self.stream_id = start_info.get('streamId') self.call_id = start_info.get('callId') extra_headers = data.get('extra_headers', {}) self.business_id = extra_headers.get('X-BID', f"BID_{self.session_id}") self.agent_id = extra_headers.get('X-Agent-id', 'Not provided') self.phone_number = extra_headers.get('X-CustNumber', 'Not provided') self.customer_name = extra_headers.get('X-CustName', 'Valued Customer') self.did = extra_headers.get('X-DID', None) # OPTIMIZATION: Run independent operations in parallel to reduce latency # customer_task = asyncio.create_task(self._fetch_customer_data(self.phone_number)) # api_data = await customer_task # if api_data: # self.customer_name = api_data.get('customer_name', self.customer_name) await self._load_bot_configuration() Log.info(f"🎙️ Initializing ElevenLabs WebSocket service for session {self.session_id}") await self._initialize_elevenlabs_websocket() self.call_start_time = datetime.now() except Exception as e: Log.error(f"Error handling call start for session {self.session_id}: {e}") async def _load_bot_configuration(self) -> None: """Load bot configuration for this call (only agent_id, company_name, bot_name needed for ElevenLabs).""" try: if not self.did: Log.warning(f"⚠️ No DID provided for session {self.session_id}, using default configuration") self.bot_config = None self.bot_name = 'Default Assistant' self.bot_id = None self.company_name = Config.COMPANY_NAME if hasattr(Config, 'COMPANY_NAME') else 'MCube' return config_result = await self.bot_configuration_service.get_bot_configuration_by_did(self.did) self.bot_config = config_result.get('bot_config') self.bot_name = config_result.get('bot_name', 'Assistant') self.bot_id = config_result.get('bot_id') self.company_name = config_result.get('company_name', Config.COMPANY_NAME if hasattr(Config, 'COMPANY_NAME') else 'MCube') except Exception as e: Log.error(f"❌ Error loading bot configuration for session {self.session_id}: {e}") self.bot_config = None self.bot_name = 'Default Assistant' self.bot_id = None self.company_name = Config.COMPANY_NAME if hasattr(Config, 'COMPANY_NAME') else 'MCube' async def handle_media_event(self, data: dict) -> None: """Handle incoming media data for this call.""" try: if not self.elevenlabs_websocket_service: # Queue audio chunk and trigger connection retry if not already retrying try: self.elevenlabs_audio_queue.put_nowait(data) # Service not initialized, audio queued # Trigger connection retry if not already retrying if not self.elevenlabs_connection_retry_task or self.elevenlabs_connection_retry_task.done(): self.elevenlabs_connection_retry_task = asyncio.create_task( self._retry_elevenlabs_connection() ) except asyncio.QueueFull: Log.warning(f"⚠️ Audio queue full for session {self.session_id}, dropping chunk") return # Skip connection check if still initializing - prevents false disconnection detection if self.elevenlabs_initializing: # ElevenLabs WebSocket still initializing, skipping audio chunk return import time current_time = time.time() # Check connection state more reliably - prioritize actual websocket state is_connected = False if self.elevenlabs_websocket_service: # First check actual websocket state (most reliable) if self.elevenlabs_websocket_service.websocket: try: if hasattr(self.elevenlabs_websocket_service.websocket, 'state'): if hasattr(self.elevenlabs_websocket_service.websocket.state, 'name'): # Trust actual websocket state over internal flag if self.elevenlabs_websocket_service.websocket.state.name == 'OPEN': is_connected = True # Sync internal flag with actual state if not self.elevenlabs_websocket_service.connected: self.elevenlabs_websocket_service.connected = True except Exception: pass # Fallback to service's is_connected() method if not is_connected: is_connected = self.elevenlabs_websocket_service.is_connected() if not is_connected: # Only log if we're not in the middle of initialization or reconnection # This reduces false positive warnings during normal connection transitions if not self.elevenlabs_initializing: websocket_state = 'unknown' if self.elevenlabs_websocket_service.websocket: if hasattr(self.elevenlabs_websocket_service.websocket, 'state'): if hasattr(self.elevenlabs_websocket_service.websocket.state, 'name'): websocket_state = self.elevenlabs_websocket_service.websocket.state.name else: websocket_state = str(self.elevenlabs_websocket_service.websocket.state) else: websocket_state = 'no state attribute' else: websocket_state = 'no websocket' # Only log if state is actually problematic (not OPEN or CONNECTING) # Rate limit: log at most once per 5 seconds to prevent log spam if websocket_state not in ['OPEN', 'CONNECTING']: current_time = time.time() if current_time - self._last_connection_warning_time >= 5.0: Log.warning(f"⚠️ Connection state check failed for session {self.session_id} - WebSocket state: {websocket_state}") self._last_connection_warning_time = current_time if not is_connected: # CRITICAL: Don't reconnect if call has legitimately ended # This prevents unwanted reconnections when: # 1. Call ended via end_call tool (Client ended call) # 2. Voicemail was detected (Voicemail detected) # 3. MCube WebSocket is disconnected (call is over) # 4. Session is no longer active # 5. WebSocket is CLOSING (might be closing for terminal reason - wait for close reason) # Check if call has been ended legitimately # Rate limit: log at most once per 5 seconds to prevent log spam when handle_media_event is called repeatedly if self.call_ended: current_time = time.time() if current_time - self._last_call_ended_log_time >= 5.0: Log.info(f"🛑 Call already ended (end_call or voicemail) - skipping reconnection for session {self.session_id}") self._last_call_ended_log_time = current_time return # CRITICAL: Check if MCube WebSocket is still connected FIRST (if not, call is over) # This is especially important for "Client ended call" where ElevenLabs sends generic "Connection closed cleanly" # If MCube is disconnected, the user hung up and we should NOT reconnect ElevenLabs # Check this BEFORE waiting for close reason to avoid race conditions mcube_disconnected = False if self.websocket: try: if hasattr(self.websocket, 'client_state'): mcube_state = self.websocket.client_state.name if hasattr(self.websocket.client_state, 'name') else str(self.websocket.client_state) if mcube_state not in ['CONNECTED', 'CONNECTING']: mcube_disconnected = True Log.info(f"🛑 MCube WebSocket disconnected ({mcube_state}) - user hung up, marking call as ended (session {self.session_id})") else: # If we can't check state, try to detect if WebSocket is closed by checking if it's None or has no state # This is a fallback for edge cases if not hasattr(self.websocket, 'client_state'): Log.debug(f"⚠️ MCube WebSocket state unavailable - assuming connected for session {self.session_id}") except Exception as e: Log.debug(f"Could not check MCube WebSocket state in handle_media_event: {e}") if mcube_disconnected: # Mark call as ended to prevent future reconnection attempts self.call_ended = True return # Skip ElevenLabs reconnection if MCube is gone # Check if WebSocket is in CLOSING state - this might indicate a terminal close reason # Wait a moment for the close reason to be processed before attempting reconnection if self.elevenlabs_websocket_service and self.elevenlabs_websocket_service.websocket: try: if hasattr(self.elevenlabs_websocket_service.websocket, 'state'): if hasattr(self.elevenlabs_websocket_service.websocket.state, 'name'): ws_state = self.elevenlabs_websocket_service.websocket.state.name else: ws_state = str(self.elevenlabs_websocket_service.websocket.state) # If WebSocket is CLOSING or CLOSED, it might be closing for a terminal reason (Agent ended call, etc.) # Give the close handler a chance to process the close reason and set call_ended if ws_state in ['CLOSING', 'CLOSED']: Log.info(f"⏳ WebSocket is {ws_state} - waiting for close reason before reconnection (session {self.session_id})") # Wait longer (2 seconds) to ensure the ConnectionClosed exception is caught and processed # This gives the receive_messages task time to catch the exception and call on_connection_closed await asyncio.sleep(2.0) # Check again if call_ended was set by the connection_closed callback if self.call_ended: Log.info(f"🛑 Call ended legitimately (close reason processed) - skipping reconnection for session {self.session_id}") return else: # If call_ended wasn't set but WebSocket is CLOSED, check MCube again # Sometimes the callback runs but MCube check happens too early if self.websocket: try: if hasattr(self.websocket, 'client_state'): mcube_state = self.websocket.client_state.name if hasattr(self.websocket.client_state, 'name') else str(self.websocket.client_state) if mcube_state not in ['CONNECTED', 'CONNECTING']: self.call_ended = True Log.info(f"🛑 MCube WebSocket disconnected ({mcube_state}) after wait - marking call as ended (session {self.session_id})") return except Exception: pass # If call_ended still False and MCube is connected, log for debugging Log.debug(f"⚠️ WebSocket is {ws_state} but call_ended is False - may have missed close reason for session {self.session_id}") # Note: WebSocket close_code/reason are typically only available in ConnectionClosed exception # So we rely on the connection_closed callback to set call_ended flag # The wait above (2.0s) should give it time to process except Exception as e: # If we can't check state, continue with normal reconnection logic Log.debug(f"Could not check WebSocket state: {e}") # Check if session is still active if not self.is_active: Log.info(f"🛑 Session no longer active - skipping reconnection for session {self.session_id}") return time_since_last_reconnect = current_time - self.elevenlabs_websocket_service.last_reconnect_time # Reduced cooldown: 2s base, max 10s (was 5s base, max 60s) cooldown = min(self.elevenlabs_reconnect_cooldown * (2 ** min(self.elevenlabs_reconnect_attempts, 2)), 10) if time_since_last_reconnect < cooldown: # Queue audio instead of dropping it try: self.elevenlabs_audio_queue.put_nowait(data) # Reconnection cooldown active, audio queued except asyncio.QueueFull: Log.warning(f"⚠️ Audio queue full for session {self.session_id}, dropping chunk") return if self.elevenlabs_reconnect_attempts >= 10: Log.error(f"❌ Max reconnection attempts ({self.elevenlabs_reconnect_attempts}) reached for session {self.session_id}. Stopping reconnection attempts.") return Log.warning(f"⚠️ ElevenLabs WebSocket not connected for session {self.session_id}, attempting to reconnect... (attempt {self.elevenlabs_reconnect_attempts + 1})") # CRITICAL: Restore conversation_id to service before reconnecting # This ensures get_websocket_url() includes conversation_id for resume # Always restore from CallSession (source of truth) even if service already has one if self.elevenlabs_conversation_id: self.elevenlabs_websocket_service.conversation_id = self.elevenlabs_conversation_id Log.info(f"🔄 Restored conversation_id to service before reconnect: {self.elevenlabs_conversation_id[:20]}...") elif self.elevenlabs_websocket_service.conversation_id: # Service has conversation_id but CallSession doesn't - sync it back self.elevenlabs_conversation_id = self.elevenlabs_websocket_service.conversation_id Log.info(f"🔄 Synced conversation_id from service to CallSession: {self.elevenlabs_conversation_id[:20]}...") # CRITICAL: Guard against reconnecting without conversation_id if we've already established one # This prevents creating duplicate conversations if reconnect happens before metadata arrives # Note: We allow first connection without conversation_id (it will be received in metadata) if self.elevenlabs_reconnect_attempts > 0 and not self.elevenlabs_conversation_id: Log.warning(f"⚠️ Reconnect blocked: conversation_id not yet established (attempt {self.elevenlabs_reconnect_attempts + 1}) - waiting for metadata") return try: await self.elevenlabs_websocket_service.connect(use_signed_url=False) Log.info(f"✅ Reconnected to ElevenLabs WebSocket for session {self.session_id}") self.elevenlabs_reconnect_attempts += 1 self.elevenlabs_websocket_service.last_reconnect_time = current_time # Start processing queued audio chunks after reconnection if not self.elevenlabs_audio_queue_task or self.elevenlabs_audio_queue_task.done(): self.elevenlabs_audio_queue_task = asyncio.create_task(self._process_audio_queue()) async def reset_reconnect_attempts_after_delay(): await asyncio.sleep(30) if self.elevenlabs_websocket_service and self.elevenlabs_websocket_service.is_connected(): self.elevenlabs_reconnect_attempts = 0 asyncio.create_task(reset_reconnect_attempts_after_delay()) if self.elevenlabs_receiver_task and not self.elevenlabs_receiver_task.done(): self.elevenlabs_receiver_task.cancel() try: await self.elevenlabs_receiver_task except asyncio.CancelledError: pass self.elevenlabs_websocket_service.on_audio_received = self._handle_elevenlabs_audio self.elevenlabs_websocket_service.on_interruption_received = self._handle_elevenlabs_interruption self.elevenlabs_websocket_service.on_response_received = self._handle_elevenlabs_response self.elevenlabs_websocket_service.on_tool_request_received = self._handle_elevenlabs_tool_request self.elevenlabs_websocket_service.on_transcript_received = self._handle_elevenlabs_transcript self.elevenlabs_websocket_service.on_vad_score_received = self._handle_elevenlabs_vad_score self.elevenlabs_websocket_service.on_connection_closed = self._handle_elevenlabs_connection_closed self.elevenlabs_websocket_service.on_automated_greeting_detected = self._handle_automated_greeting_detected self.elevenlabs_websocket_service.on_tool_abandoned = self._handle_tool_abandoned async def message_handler_wrapper(message: dict): """Wrapper to handle messages from ElevenLabs.""" try: await self.elevenlabs_websocket_service.handle_message(message) except Exception as e: Log.error(f"❌ Error in message handler for session {self.session_id}: {e}") import traceback traceback.print_exc() self.elevenlabs_receiver_task = asyncio.create_task( self.elevenlabs_websocket_service.receive_messages(message_handler_wrapper) ) except Exception as reconnect_error: Log.error(f"❌ Failed to reconnect to ElevenLabs WebSocket: {reconnect_error}") return audio_payload = data.get('media', {}).get('payload', '') if audio_payload: try: import base64 import audioop mulaw_bytes = base64.b64decode(audio_payload) if not self.elevenlabs_websocket_service or not self.elevenlabs_websocket_service.is_connected(): return pcm_bytes = audioop.ulaw2lin(mulaw_bytes, 2) if len(pcm_bytes) > 0: try: await self.elevenlabs_websocket_service.send_audio_chunk(pcm_bytes) except Exception as audio_send_error: if "ConnectionClosed" not in type(audio_send_error).__name__: Log.warning(f"⚠️ Failed to send audio chunk: {audio_send_error}") except Exception as send_error: if "ConnectionClosed" not in type(send_error).__name__: Log.error(f"❌ Failed to send audio to ElevenLabs: {send_error}") return except Exception as e: Log.error(f"Error handling media for session {self.session_id}: {e}") async def handle_played_stream_event(self, name: str) -> None: """Handle playedStream event from MCube for audio completion.""" try: self.audio_service.handle_played_stream_event(name) if self.elevenlabs_audio_playing and "elevenlabs_audio" in name: self.elevenlabs_audio_playing = False if self.websocket and hasattr(self.websocket, 'client_state'): if self.websocket.client_state.name == 'CONNECTED': played_stream_message = { "event": "playedStream", "streamId": self.connection_manager.state.stream_id, "name": name } await self.websocket.send_text(json.dumps(played_stream_message)) except Exception as e: import traceback Log.error(f"❌ Error handling playedStream event: {e}") Log.error(f"Full traceback: {traceback.format_exc()}") async def _initialize_elevenlabs_websocket(self) -> None: """Initialize ElevenLabs WebSocket service for service_type 4.""" try: self.elevenlabs_initializing = True # Set flag to prevent connection checks during init from services.elevenlabs_websocket_service import ElevenLabsWebSocketService from services.elevenlabs_connection_limiter import elevenlabs_limiter # Acquire connection slot before connecting if not await elevenlabs_limiter.acquire(self.session_id, timeout=10.0): Log.warning( f"⚠️ ElevenLabs connection limit reached for session {self.session_id}. " f"Active: {elevenlabs_limiter.get_active_count()}/{elevenlabs_limiter.max_connections}. " f"Audio chunks will be queued and connection will be retried." ) # Don't raise exception - let retry mechanism handle it # Audio chunks will be queued in handle_media_event self.elevenlabs_initializing = False return try: # Get agent_id from bot_config (fetched from database) # Database returns 'agent_id', not 'elevenlabs_agent_id' agent_id = self.bot_config.get('agent_id') if self.bot_config else None if not agent_id: # Fallback to config if not found in database agent_id = Config.ELEVENLABS_AGENT_ID if hasattr(Config, 'ELEVENLABS_AGENT_ID') else None if not agent_id: raise ValueError("ELEVENLABS_AGENT_ID is required. Set it in bot config or Config.") self.elevenlabs_websocket_service = ElevenLabsWebSocketService(agent_id=agent_id) # CRITICAL: Restore conversation_id to service if we have it stored (from previous connection) # This ensures conversation_id is available in the service for get_websocket_url() if self.elevenlabs_conversation_id: self.elevenlabs_websocket_service.conversation_id = self.elevenlabs_conversation_id Log.info(f"🔄 Restored conversation_id to service on initialization: {self.elevenlabs_conversation_id[:20]}...") await self.elevenlabs_websocket_service.connect(use_signed_url=False) except Exception as e: # Release slot if connection fails await elevenlabs_limiter.release(self.session_id) raise self.elevenlabs_websocket_service.on_audio_received = self._handle_elevenlabs_audio self.elevenlabs_websocket_service.on_transcript_received = self._handle_elevenlabs_transcript self.elevenlabs_websocket_service.on_response_received = self._handle_elevenlabs_response self.elevenlabs_websocket_service.on_vad_score_received = self._handle_elevenlabs_vad_score self.elevenlabs_websocket_service.on_interruption_received = self._handle_elevenlabs_interruption self.elevenlabs_websocket_service.on_tool_request_received = self._handle_elevenlabs_tool_request self.elevenlabs_websocket_service.on_connection_closed = self._handle_elevenlabs_connection_closed self.elevenlabs_websocket_service.on_automated_greeting_detected = self._handle_automated_greeting_detected self.elevenlabs_websocket_service.on_tool_abandoned = self._handle_tool_abandoned self.elevenlabs_websocket_service.on_conversation_id_received = self._handle_conversation_id_received # CRITICAL: Restore conversation_id to service if we have it stored (from previous connection) # This ensures conversation_id is available in the service for get_websocket_url() if self.elevenlabs_conversation_id and not self.elevenlabs_websocket_service.conversation_id: self.elevenlabs_websocket_service.conversation_id = self.elevenlabs_conversation_id Log.info(f"🔄 Restored conversation_id to service: {self.elevenlabs_conversation_id[:20]}...") # CRITICAL: Only send conversation_initiation if conversation_id doesn't exist # If conversation_id exists, we're resuming an existing conversation and should NOT re-initiate # This prevents creating duplicate conversations on reconnection # Use CallSession's stored conversation_id as source of truth if self.elevenlabs_conversation_id is None: await self.elevenlabs_websocket_service.send_conversation_initiation( text_only=False, # Use audio mode dynamic_variables={ "bot_name": self.company_name or "MCube", "phone_number": getattr(self, 'phone_number', 'Unknown'), "call_id": self.call_id or "Unknown", "mcube_business_id": self.business_id or "Unknown" } ) else: Log.info(f"🔄 Skipping conversation_initiation - resuming existing conversation: {self.elevenlabs_conversation_id[:20]}...") async def message_handler_wrapper(message: dict): """Wrapper to handle messages from ElevenLabs.""" try: await self.elevenlabs_websocket_service.handle_message(message) except Exception as e: Log.error(f"❌ Error in message handler for session {self.session_id}: {e}") import traceback traceback.print_exc() self.elevenlabs_receiver_task = asyncio.create_task( self.elevenlabs_websocket_service.receive_messages(message_handler_wrapper) ) self.elevenlabs_initializing = False # Clear flag after successful initialization # Start processing queued audio chunks if any if not self.elevenlabs_audio_queue_task or self.elevenlabs_audio_queue_task.done(): self.elevenlabs_audio_queue_task = asyncio.create_task(self._process_audio_queue()) except Exception as e: Log.error(f"❌ Error initializing ElevenLabs WebSocket for session {self.session_id}: {e}") self.elevenlabs_websocket_service = None self.elevenlabs_initializing = False # Clear flag on error async def _retry_elevenlabs_connection(self) -> None: """Retry ElevenLabs connection initialization periodically when connection limit is reached.""" max_retries = 30 # Retry for 5 minutes (30 * 10s) retry_interval = 10 # Retry every 10 seconds for attempt in range(max_retries): try: # Check if service is already initialized (another retry might have succeeded) if self.elevenlabs_websocket_service and self.elevenlabs_websocket_service.is_connected(): return # Try to acquire connection slot from services.elevenlabs_connection_limiter import elevenlabs_limiter if await elevenlabs_limiter.acquire(self.session_id, timeout=1.0): # Slot acquired - initialize service try: await self._initialize_elevenlabs_websocket() # Process queued chunks if not self.elevenlabs_audio_queue_task or self.elevenlabs_audio_queue_task.done(): self.elevenlabs_audio_queue_task = asyncio.create_task( self._process_audio_queue() ) return except Exception as init_error: # Release slot if initialization fails await elevenlabs_limiter.release(self.session_id) Log.warning(f"⚠️ Connection initialization failed after acquiring slot: {init_error}") # Continue retrying except Exception: # Retry attempt failed, will retry again pass # Wait before next retry await asyncio.sleep(retry_interval) Log.error(f"❌ Failed to connect ElevenLabs for {self.session_id} after {max_retries} retries") async def _process_audio_queue(self) -> None: """Process queued audio chunks after reconnection.""" try: processed_count = 0 while not self.elevenlabs_audio_queue.empty(): if not self.elevenlabs_websocket_service or not self.elevenlabs_websocket_service.is_connected(): Log.warning(f"⚠️ Connection lost while processing audio queue for session {self.session_id}") break try: # Get queued audio data (with timeout to prevent blocking) queued_data = await asyncio.wait_for(self.elevenlabs_audio_queue.get(), timeout=0.1) # Process the queued audio chunk audio_payload = queued_data.get('media', {}).get('payload', '') if audio_payload: import base64 import audioop mulaw_bytes = base64.b64decode(audio_payload) pcm_bytes = audioop.ulaw2lin(mulaw_bytes, 2) if len(pcm_bytes) > 0: await self.elevenlabs_websocket_service.send_audio_chunk(pcm_bytes) processed_count += 1 self.elevenlabs_audio_queue.task_done() except asyncio.TimeoutError: break # No more items in queue except Exception as e: Log.warning(f"⚠️ Error processing queued audio chunk: {e}") break # Processed queued audio chunks except Exception as e: Log.error(f"❌ Error in audio queue processing for session {self.session_id}: {e}") async def _handle_elevenlabs_audio(self, audio_bytes: bytes) -> None: """Handle audio received from ElevenLabs and send to Mcube.""" try: if not audio_bytes or len(audio_bytes) == 0: return if not self.connection_manager.state.stream_id: return # Early exit if Mcube WebSocket is disconnected to prevent spam if self.websocket.client_state.name != 'CONNECTED': return import base64 import audioop # Get the actual audio format from ElevenLabs (e.g., "pcm_16000", "mulaw_8000", "ulaw_8000") elevenlabs_format = getattr(self.elevenlabs_websocket_service, 'agent_audio_format', 'pcm_16000') target_sample_rate = Config.MCUBE_SAMPLE_RATE # 8000 Hz (configurable) mulaw_bytes = None if 'ulaw_8000' in str(elevenlabs_format) or 'mulaw_8000' in str(elevenlabs_format): mulaw_bytes = audio_bytes else: # Fallback conversion logic is_mulaw = isinstance(elevenlabs_format, str) and ('mulaw' in elevenlabs_format.lower() or 'ulaw' in elevenlabs_format.lower()) source_sample_rate = 16000 if isinstance(elevenlabs_format, str): if '16000' in elevenlabs_format or '16k' in elevenlabs_format.lower(): source_sample_rate = 16000 elif '8000' in elevenlabs_format or '8k' in elevenlabs_format.lower(): source_sample_rate = 8000 else: source_sample_rate = 8000 if is_mulaw else 16000 else: source_sample_rate = 16000 if is_mulaw: mulaw_bytes = audio_bytes # Already in μ-law format if source_sample_rate != target_sample_rate: pcm_bytes = audioop.ulaw2lin(mulaw_bytes, 2) resampled_pcm, _ = audioop.ratecv( pcm_bytes, 2, 1, source_sample_rate, target_sample_rate, None # No state ) # Convert back to μ-law mulaw_bytes = audioop.lin2ulaw(resampled_pcm, 2) else: if source_sample_rate != target_sample_rate: resampled_bytes, _ = audioop.ratecv( audio_bytes, 2, 1, source_sample_rate, target_sample_rate, None # No state ) audio_bytes = resampled_bytes mulaw_bytes = audioop.lin2ulaw(audio_bytes, 2) chunk_size = 640 total_chunks = (len(mulaw_bytes) + chunk_size - 1) // chunk_size self.elevenlabs_audio_playing = True for i in range(0, len(mulaw_bytes), chunk_size): # Check connection state before each chunk to avoid spam if self.websocket.client_state.name != 'CONNECTED': self.elevenlabs_audio_playing = False return chunk = mulaw_bytes[i:i + chunk_size] audio_base64 = base64.b64encode(chunk).decode('utf-8') audio_message = { "event": "playAudio", "media": { "contentType": Config.MCUBE_AUDIO_FORMAT, "sampleRate": Config.MCUBE_SAMPLE_RATE, "payload": audio_base64, "name": f"elevenlabs_audio_{int(datetime.now().timestamp())}_{i // chunk_size}" } } try: await self.connection_manager.send_to_mcube(audio_message) except (WebSocketDisconnect, RuntimeError) as send_error: if "websocket.send" in str(send_error) or "websocket.close" in str(send_error) or "already completed" in str(send_error): self.elevenlabs_audio_playing = False return raise # Re-raise if it's a different error try: last_chunk_name = f"elevenlabs_audio_{int(datetime.now().timestamp())}_{total_chunks - 1}" checkpoint_message = self.audio_service.create_checkpoint_message( self.connection_manager.state.stream_id, last_chunk_name ) await self.connection_manager.send_to_mcube(checkpoint_message) except Exception: pass except Exception as e: Log.error(f"❌ Error handling ElevenLabs audio: {e}") async def _handle_elevenlabs_transcript(self, transcript: dict) -> None: """Handle transcript received from ElevenLabs.""" try: # Extract transcript text from dict format # Transcript comes as: {'user_transcript': '...', 'event_id': ...} transcript_text = transcript.get('user_transcript', '') if isinstance(transcript, dict) else str(transcript) Log.info(f"🎤 [ElevenLabs User Transcript]: {transcript}") # Check for voicemail patterns in transcript if transcript_text and not self.call_ended: transcript_lower = transcript_text.lower() voicemail_patterns = [ 'voicemail', 'leave a message', 'record your message', 'not available', 'at the tone', 'please record', 'after the tone', 'finished recording', 'person you are trying to reach', 'please record your message' ] if any(pattern in transcript_lower for pattern in voicemail_patterns): Log.info(f"🛑 Voicemail detected from transcript - marking call as ended (session {self.session_id})") self.call_ended = True # Don't process this as a normal user message return # Process as normal message if not voicemail self.user_messages.append({ 'content': transcript_text, 'timestamp': datetime.now().isoformat(), 'type': 'user' }) # Limit message history to prevent memory issues if len(self.user_messages) > self.max_messages: self.user_messages = self.user_messages[-self.max_messages:] except Exception as e: Log.error(f"❌ Error handling ElevenLabs transcript: {e}") async def _handle_elevenlabs_response(self, response: str) -> None: """Handle text response received from ElevenLabs.""" try: Log.info(f"🤖 [ElevenLabs Agent Response]: {response}") self.bot_messages.append({ 'content': response, 'timestamp': datetime.now().isoformat(), 'type': 'bot' }) # Limit message history to prevent memory issues if len(self.bot_messages) > self.max_messages: self.bot_messages = self.bot_messages[-self.max_messages:] except Exception as e: Log.error(f"❌ Error handling ElevenLabs response: {e}") async def _handle_elevenlabs_vad_score(self, score: float) -> None: """Handle VAD score received from ElevenLabs for interruption detection with noise filtering.""" try: import time current_time = time.time() if not self.elevenlabs_audio_playing: return self.vad_score_history.append((current_time, score)) cutoff_time = current_time - self.vad_history_duration self.vad_score_history = [(t, s) for t, s in self.vad_score_history if t > cutoff_time] if len(self.vad_score_history) < 2: return recent_scores = [s for _, s in self.vad_score_history] avg_score = sum(recent_scores) / len(recent_scores) max_score = max(recent_scores) VAD_THRESHOLD = 0.45 high_scores = [s for s in recent_scores if s > self.min_sustained_score] sustained_duration = len(high_scores) * 0.05 if score > VAD_THRESHOLD and avg_score > VAD_THRESHOLD and sustained_duration >= self.min_sustained_duration: time_since_last_interrupt = current_time - self.last_interrupt_time if time_since_last_interrupt < self.interrupt_debounce_seconds: return self.last_interrupt_time = current_time Log.info(f"🎙️ ElevenLabs VAD detected sustained user speech (score: {score:.2f}, avg: {avg_score:.2f}, sustained: {sustained_duration:.2f}s) - triggering interruption") self.elevenlabs_audio_playing = False self.vad_score_history.clear() if self.elevenlabs_websocket_service and self.elevenlabs_websocket_service.is_connected(): await self.elevenlabs_websocket_service.send_interruption() if self.connection_manager.state.stream_id: try: clear_audio_message = { "event": "clearAudio", "streamId": self.connection_manager.state.stream_id } await self.connection_manager.send_to_mcube(clear_audio_message) except Exception as e: Log.error(f"❌ Error sending clearAudio: {e}") except Exception as e: Log.error(f"❌ Error handling ElevenLabs VAD score: {e}") async def _handle_elevenlabs_connection_closed(self, close_code: Any, close_reason: str, is_terminal_reason: bool) -> None: """ Handle ElevenLabs WebSocket connection closed event. Args: close_code: WebSocket close code close_reason: Close reason string from ElevenLabs is_terminal_reason: True if this is a terminal call end reason (should not reconnect) """ try: if is_terminal_reason: # Mark call as ended to prevent unwanted reconnections self.call_ended = True Log.info(f"🛑 Call marked as ended (ElevenLabs: {close_reason}) - reconnections disabled for session {self.session_id}") else: # Non-terminal reason (e.g., network error) - reconnection may be allowed Log.info(f"⚠️ ElevenLabs connection closed (Code: {close_code}, Reason: {close_reason}) - may allow reconnection") # CRITICAL: Check MCube WebSocket state for ANY non-terminal close reason # If MCube is disconnected, the user hung up and we should mark call as ended # This prevents unwanted reconnections when the user hangs up, regardless of the close reason # This is especially important for "Client ended call" where ElevenLabs sends "Connection closed cleanly" mcube_disconnected = False if self.websocket: try: if hasattr(self.websocket, 'client_state'): mcube_state = self.websocket.client_state.name if hasattr(self.websocket.client_state, 'name') else str(self.websocket.client_state) if mcube_state not in ['CONNECTED', 'CONNECTING']: mcube_disconnected = True Log.info(f"🛑 MCube WebSocket disconnected ({mcube_state}) - user hung up, marking call as ended (session {self.session_id})") else: # Try alternative method to check if WebSocket is closed try: # Try to access WebSocket to see if it's still valid if not hasattr(self.websocket, 'client_state'): # If client_state doesn't exist, WebSocket might be closed # But we can't be sure, so log and continue Log.debug(f"⚠️ MCube WebSocket state unavailable in connection_closed handler for session {self.session_id}") except Exception: # If accessing websocket fails, it's likely closed mcube_disconnected = True Log.info(f"🛑 MCube WebSocket appears closed (access failed) - marking call as ended (session {self.session_id})") except Exception as e: Log.debug(f"Could not check MCube WebSocket state in connection_closed handler: {e}") if mcube_disconnected: self.call_ended = True Log.info(f"🛑 Call marked as ended (MCube disconnected + ElevenLabs: {close_reason}) - reconnections disabled (session {self.session_id})") except Exception as e: Log.error(f"❌ Error handling connection closed event: {e}") async def _handle_conversation_id_received(self, conversation_id: str) -> None: """ Handle conversation_id received from ElevenLabs. Store it in CallSession for safety (prevents loss if service is recreated). Args: conversation_id: The conversation ID from ElevenLabs """ try: if not self.elevenlabs_conversation_id: self.elevenlabs_conversation_id = conversation_id Log.info(f"💾 Stored conversation_id in CallSession: {conversation_id[:20]}... (session {self.session_id})") except Exception as e: Log.error(f"❌ Error storing conversation_id: {e}") async def _handle_automated_greeting_detected(self, greeting_text: str, is_voicemail: bool = False) -> None: """ Handle automated greeting detection from ElevenLabs (voicemail detection). Args: greeting_text: The detected greeting text is_voicemail: True if this is a voicemail greeting """ try: if is_voicemail: Log.info(f"🛑 Voicemail detected via automated greeting - marking call as ended (session {self.session_id})") self.call_ended = True else: Log.info(f"📞 Automated greeting detected (not voicemail): {greeting_text[:100]}...") except Exception as e: Log.error(f"❌ Error handling automated greeting detection: {e}") async def _handle_tool_abandoned(self, tool_call_id: str, result: str) -> None: """ Handle tool execution abandonment (e.g., end_call or transfer_to_number tool abandoned due to user input). Args: tool_call_id: The tool call ID that was abandoned result: The abandonment message from ElevenLabs """ try: # Check if this was an end_call tool that was abandoned if self.pending_end_call_tool_id == tool_call_id: # Reset call_ended flag since the call is still active self.call_ended = False self.pending_end_call_tool_id = None Log.warning(f"⚠️ end_call tool execution abandoned (tool_call_id: {tool_call_id}) - call is still active, resetting call_ended flag (session {self.session_id})") Log.warning(f" Abandonment reason: {result[:200] if result else 'No reason provided'}") # Check if this was a transfer tool that was abandoned elif self.pending_transfer_tool_id == tool_call_id: transfer_target = self.pending_transfer_target or "unknown target" Log.warning(f"⚠️ transfer_to_number tool execution abandoned (tool_call_id: {tool_call_id}, target: {transfer_target}) - user interrupted transfer (session {self.session_id})") Log.warning(f" Abandonment reason: {result[:200] if result else 'No reason provided'}") # Clear tracking self.pending_transfer_tool_id = None self.pending_transfer_target = None # Note: Transfer abandonment doesn't require resetting call_ended flag # The call continues normally after transfer is abandoned else: # Not a tracked tool, or tool_call_id doesn't match - log for debugging Log.info(f"🔧 Tool execution abandoned (tool_call_id: {tool_call_id}): {result[:200] if result else 'No result'}") except Exception as e: Log.error(f"❌ Error handling tool abandonment event: {e}") async def _handle_elevenlabs_interruption(self, event_id: Any) -> None: """Handle interruption event received from ElevenLabs.""" try: Log.info(f"⚠️ ElevenLabs interruption detected (event_id: {event_id}) - sending clearAudio to Mcube") self.elevenlabs_audio_playing = False self.vad_score_history.clear() if self.connection_manager.state.stream_id: try: clear_audio_message = { "event": "clearAudio", "streamId": self.connection_manager.state.stream_id } await self.connection_manager.send_to_mcube(clear_audio_message) except Exception as e: Log.error(f"❌ Error sending clearAudio on interruption: {e}") else: Log.warning(f"⚠️ No stream_id available, cannot send clearAudio") except Exception as e: Log.error(f"❌ Error handling ElevenLabs interruption: {e}") async def _handle_elevenlabs_tool_request(self, tool_name: str, tool_call_id: str, tool_request: dict) -> None: """Handle tool request from ElevenLabs (e.g., end_call).""" try: Log.info(f"🔧 Received tool request: {tool_name} (call_id: {tool_call_id})") if tool_name == "end_call": # Track this tool call ID to detect if it gets abandoned self.pending_end_call_tool_id = tool_call_id # Mark call as ended to prevent unwanted reconnections # NOTE: If tool execution is abandoned, we'll reset this flag self.call_ended = True Log.info(f"🛑 Call marked as ended (end_call tool triggered, tool_call_id: {tool_call_id}) - reconnections disabled for session {self.session_id}") if self.connection_manager.state.stream_id: try: from services.mcube_service import McubeService terminate_message = McubeService.create_terminate_message( self.connection_manager.state.stream_id ) await self.connection_manager.send_to_mcube(terminate_message) except Exception as e: import traceback Log.error(f"❌ Error sending terminate message: {e}") Log.error(f"Full traceback: {traceback.format_exc()}") else: Log.warning(f"⚠️ No stream_id available, cannot send terminate message") if self.elevenlabs_websocket_service and self.elevenlabs_websocket_service.is_connected(): try: await self.elevenlabs_websocket_service.send_client_tool_result( tool_call_id=tool_call_id, result="Call terminated successfully", is_error=False ) except Exception as e: import traceback Log.error(f"❌ Error sending tool result: {e}") Log.error(f"Full traceback: {traceback.format_exc()}") elif tool_name == "voicemail_detection": # Handle voicemail detection tool request # The voicemail is typically detected via transcript, so we check recent transcripts # or the tool request parameters for voicemail indicators voicemail_detected = False voicemail_message = "" # Check if voicemail indicators are in the tool request parameters tool_params = tool_request.get("parameters", {}) if tool_params: # Check for common voicemail keywords in parameters voicemail_keywords = ["voicemail", "not available", "leave a message", "at the tone"] params_str = str(tool_params).lower() if any(keyword in params_str for keyword in voicemail_keywords): voicemail_detected = True voicemail_message = "Voicemail detected from tool parameters" # Also check recent user transcripts for voicemail indicators if not voicemail_detected and self.user_messages: recent_messages = self.user_messages[-5:] # Check last 5 messages voicemail_keywords = ["voicemail", "not available", "leave a message", "at the tone", "record your message"] for msg in recent_messages: msg_text = str(msg.get("content", "")).lower() if any(keyword in msg_text for keyword in voicemail_keywords): voicemail_detected = True voicemail_message = f"Voicemail detected in transcript: {msg.get('content', '')[:100]}" break # If voicemail is detected, mark call as ended to prevent unwanted reconnections if voicemail_detected: self.call_ended = True Log.info(f"🛑 Call marked as ended (voicemail detected) - reconnections disabled for session {self.session_id}") # Send tool result back to ElevenLabs if self.elevenlabs_websocket_service and self.elevenlabs_websocket_service.is_connected(): try: result = { "voicemail_detected": voicemail_detected, "message": voicemail_message if voicemail_detected else "No voicemail detected" } await self.elevenlabs_websocket_service.send_client_tool_result( tool_call_id=tool_call_id, result=json.dumps(result), is_error=False ) Log.info(f"✅ Voicemail detection result sent: {result}") except Exception as e: import traceback Log.error(f"❌ Error sending voicemail detection result: {e}") Log.error(f"Full traceback: {traceback.format_exc()}") else: Log.warning(f"⚠️ ElevenLabs WebSocket not connected, cannot send voicemail detection result") elif tool_name == "detect_language" or tool_name == "language_detection": # ------------------------------------------------------------------------- # detect_language: works for ANY language ElevenLabs supports # ------------------------------------------------------------------------- # Path 1 (primary): ElevenLabs sends language in tool params. # - We accept: language, language_code, detected_language, lang # - Any value (kn, ml, mr, bn, gu, pa, or, as, fr, es, ...) is PASSED # THROUGH as-is. No code change needed when ElevenLabs adds languages. # Path 2 (fallback): Params are empty. We use simple word heuristics for # a few languages only. For others, we return "unknown"; ElevenLabs # typically sends params when they support the language. # To add a new language: (1) Params: no change. (2) Heuristics: add # trigger words in the fallback block below (optional). # ------------------------------------------------------------------------- detected_language = "unknown" confidence = 0.0 tool_params = tool_request.get("parameters") or {} # ElevenLabs may send language_id, language, language_code, detected_language, or lang raw_lang = ( tool_params.get("language_id") or tool_params.get("language") or tool_params.get("language_code") or tool_params.get("detected_language") or tool_params.get("lang") ) if raw_lang is not None and str(raw_lang).strip(): raw_lang = str(raw_lang).strip() _norm = raw_lang.lower() # Map ElevenLabs language_id / name / locale to stable code. # Covers: ElevenLabs supported languages + Indian language variants. # ElevenLabs list: en, ja, zh, de, hi, fr, ko, pt, it, es, ru, id, nl, tr, # fil, pl, sv, bg, ro, ar, cs, el, fi, hr, ms, sk, da, ta, uk, vi, no, hu. # Unlisted values are passed through (len <= 10 to support "fil", "pt-BR", etc.). _NAME_TO_CODE = { # ElevenLabs languages (language_id + name) "en": "en", "en-in": "en", "en-us": "en", "english": "en", "ja": "ja", "japanese": "ja", "zh": "zh", "chinese": "zh", "zh-cn": "zh", "zh-tw": "zh", "de": "de", "german": "de", "hi": "hi", "hi-in": "hi", "hindi": "hi", "fr": "fr", "french": "fr", "ko": "ko", "korean": "ko", "pt": "pt", "portuguese": "pt", "pt-br": "pt", "pt-pt": "pt", "it": "it", "italian": "it", "es": "es", "spanish": "es", "ru": "ru", "russian": "ru", "id": "id", "indonesian": "id", "nl": "nl", "dutch": "nl", "tr": "tr", "turkish": "tr", "fil": "fil", "filipino": "fil", "tagalog": "fil", "pl": "pl", "polish": "pl", "sv": "sv", "swedish": "sv", "bg": "bg", "bulgarian": "bg", "ro": "ro", "romanian": "ro", "ar": "ar", "arabic": "ar", "cs": "cs", "czech": "cs", "el": "el", "greek": "el", "fi": "fi", "finnish": "fi", "hr": "hr", "croatian": "hr", "ms": "ms", "malay": "ms", "sk": "sk", "slovak": "sk", "da": "da", "danish": "da", "ta": "ta", "ta-in": "ta", "tamil": "ta", "in flag tamil": "ta", "uk": "uk", "ukrainian": "uk", "vi": "vi", "vietnamese": "vi", "no": "no", "norwegian": "no", "nb": "no", "nn": "no", "hu": "hu", "hungarian": "hu", # Indian languages (extra) "te": "te", "te-in": "te", "telugu": "te", "kn": "kn", "kn-in": "kn", "kannada": "kn", "ml": "ml", "ml-in": "ml", "malayalam": "ml", "mr": "mr", "mr-in": "mr", "marathi": "mr", "bn": "bn", "bn-in": "bn", "bengali": "bn", "gu": "gu", "gu-in": "gu", "gujarati": "gu", "pa": "pa", "pa-in": "pa", "punjabi": "pa", "or": "or", "or-in": "or", "odia": "or", "orya": "or", "as": "as", "as-in": "as", "assamese": "as", } detected_language = _NAME_TO_CODE.get(_norm) if detected_language is None: # Unknown code/name from ElevenLabs → pass through (e.g. new lang they add later) detected_language = raw_lang if len(raw_lang) <= 10 else raw_lang[:10] confidence = float(tool_params.get("confidence") or 0.9) # Fallback: heuristics only when params gave nothing (best-effort for a few langs) # CRITICAL: Handle race condition where tool request and transcript arrive simultaneously # When params are empty and user_messages is empty, wait briefly (150ms) for transcript to arrive # This prevents returning 'unknown' when the transcript is arriving at the same time tool_event_id = tool_request.get("event_id") if detected_language == "unknown": # If no user messages yet, wait briefly for transcript (same event_id) to be processed if not self.user_messages and tool_event_id: await asyncio.sleep(0.15) # 150ms wait for transcript to arrive and be added to user_messages # Check user_messages (may have been updated during wait) if self.user_messages: recent_messages = self.user_messages[-3:] for msg in recent_messages: msg_text = str(msg.get("content", "")).lower() if any(w in msg_text for w in ["hello", "yes", "no", "thank", "please", "ok", "okay", "it's clean", "come on"]): detected_language, confidence = "en", 0.7 break if any(w in msg_text for w in ["namaste", "haan", "nahi", "kaise", "kya"]): detected_language, confidence = "hi", 0.7 break # Tamil: common words/scripts (including "சொல்லுங்கள்" from logs) if any(w in msg_text for w in ["ஆமா", "சரி", "உம்", "வணக்கம்", "எனக்கு", "புரியுது", "கண்டிப்பா", "சொல்லுங்கள்", "பேசுகிறீர்களா", "amma", "sari", "puriyuthu", "solungala"]): detected_language, confidence = "ta", 0.7 break if any(w in msg_text for w in ["అవును", "సరే", "ఎలా", "emu", "sare", "ela"]): detected_language, confidence = "te", 0.7 break if any(w in msg_text for w in ["ಹೌದು", "ಸರಿ", "howdu", "sari"]): detected_language, confidence = "kn", 0.7 break if any(w in msg_text for w in ["അതെ", "ശരി", "athae", "sari"]): detected_language, confidence = "ml", 0.7 break # If still unknown, leave as "unknown"; ElevenLabs usually sends params. result = {"language": detected_language, "confidence": confidence} if self.elevenlabs_websocket_service and self.elevenlabs_websocket_service.is_connected(): try: await self.elevenlabs_websocket_service.send_client_tool_result( tool_call_id=tool_call_id, result=json.dumps(result), is_error=False ) Log.info(f"✅ Language detection result sent: {result}") except Exception as e: Log.error(f"❌ Error sending language detection result: {e}") else: Log.warning(f"⚠️ ElevenLabs WebSocket not connected, cannot send language detection result") elif tool_name == "skip_turn": # Handle skip turn tool request # Skip the current turn/response if self.elevenlabs_websocket_service and self.elevenlabs_websocket_service.is_connected(): try: await self.elevenlabs_websocket_service.send_client_tool_result( tool_call_id=tool_call_id, result="Turn skipped successfully", is_error=False ) Log.info(f"✅ Skip turn executed for session {self.session_id}") except Exception as e: import traceback Log.error(f"❌ Error sending skip turn result: {e}") Log.error(f"Full traceback: {traceback.format_exc()}") else: Log.warning(f"⚠️ ElevenLabs WebSocket not connected, cannot send skip turn result") elif tool_name == "transfer_to_agent": # Handle transfer to agent tool request tool_params = tool_request.get("parameters", {}) agent_number = tool_params.get("agent_number") if tool_params else None # Use default agent number from config if not provided if not agent_number: agent_number = Config.AGENT_PHONE_NUMBER if hasattr(Config, 'AGENT_PHONE_NUMBER') else None if self.connection_manager.state.stream_id and agent_number: try: from services.mcube_service import McubeService transfer_message = McubeService.create_transfer_message( self.connection_manager.state.stream_id, agent_number, show_original_caller_id=True ) await self.connection_manager.send_to_mcube(transfer_message) Log.info(f"✅ Transfer to agent initiated: {agent_number}") # Send tool result back to ElevenLabs if self.elevenlabs_websocket_service and self.elevenlabs_websocket_service.is_connected(): await self.elevenlabs_websocket_service.send_client_tool_result( tool_call_id=tool_call_id, result=f"Call transferred to agent: {agent_number}", is_error=False ) except Exception as e: import traceback Log.error(f"❌ Error transferring to agent: {e}") Log.error(f"Full traceback: {traceback.format_exc()}") # Send error result to ElevenLabs if self.elevenlabs_websocket_service and self.elevenlabs_websocket_service.is_connected(): await self.elevenlabs_websocket_service.send_client_tool_result( tool_call_id=tool_call_id, result=f"Error transferring to agent: {e}", is_error=True ) else: error_msg = "No stream_id or agent_number available for transfer" Log.warning(f"⚠️ {error_msg}") if self.elevenlabs_websocket_service and self.elevenlabs_websocket_service.is_connected(): await self.elevenlabs_websocket_service.send_client_tool_result( tool_call_id=tool_call_id, result=error_msg, is_error=True ) elif tool_name == "transfer_to_number": # Handle transfer to number tool request # Track this tool call ID to handle abandonment self.pending_transfer_tool_id = tool_call_id tool_params = tool_request.get("parameters", {}) phone_number = tool_params.get("phone_number") if tool_params else None sip_uri = tool_params.get("sip_uri") if tool_params else None transfer_number = tool_params.get("transfer_number") if tool_params else None # Try multiple parameter names (different ElevenLabs configurations use different names) transfer_target = phone_number or sip_uri or transfer_number # FALLBACK: If ElevenLabs doesn't send parameters (known issue with SIP REFER config), # use the SIP URI from Human Transfer Rules configuration if not transfer_target: # Get bot-specific transfer number from configuration # For now using the configured SIP URI - in future can be stored per-bot in database transfer_target = "sip:9873216540@1.6.192.216" Log.info(f" ⚠️ ElevenLabs sent empty parameters, using configured SIP URI as fallback") # EXTRACT phone number from SIP URI for MCube (MCube only accepts E.164 format) # MCube documentation: "transferTo": "phone number to transfer call to in E.164 format" if transfer_target and transfer_target.startswith("sip:"): # Extract phone number: sip:918062911257@1.6.192.216 -> 918062911257 phone_number_only = transfer_target.split(":")[1].split("@")[0] Log.info(f" 🔄 Extracted phone number from SIP URI: {phone_number_only}") transfer_target = phone_number_only # Validate transfer target - check for "Unknown" or invalid values if not transfer_target or transfer_target.lower() in ["unknown", "unknown number", "none", ""]: error_msg = f"Invalid transfer target: '{transfer_target}' - cannot transfer to unknown number" Log.warning(f"⚠️ {error_msg} (tool_call_id: {tool_call_id})") self.pending_transfer_target = None if self.elevenlabs_websocket_service and self.elevenlabs_websocket_service.is_connected(): await self.elevenlabs_websocket_service.send_client_tool_result( tool_call_id=tool_call_id, result=error_msg, is_error=True ) # Clear tracking since we've handled the error self.pending_transfer_tool_id = None return # Store transfer target for logging if abandonment occurs self.pending_transfer_target = transfer_target Log.info(f" 📞 Transfer target (E.164 format): {transfer_target}") Log.info(f" 🌐 Stream ID: {self.connection_manager.state.stream_id}") if self.connection_manager.state.stream_id and transfer_target: try: from services.mcube_service import McubeService transfer_message = McubeService.create_transfer_message( self.connection_manager.state.stream_id, transfer_target, show_original_caller_id=True ) await self.connection_manager.send_to_mcube(transfer_message) Log.info(f"✅ Transfer initiated to: {transfer_target}") # Send tool result back to ElevenLabs if self.elevenlabs_websocket_service and self.elevenlabs_websocket_service.is_connected(): await self.elevenlabs_websocket_service.send_client_tool_result( tool_call_id=tool_call_id, result=f"Call transferred to: {transfer_target}", is_error=False ) # Clear tracking on success self.pending_transfer_tool_id = None self.pending_transfer_target = None except Exception as e: import traceback Log.error(f"❌ Error transferring to number: {e}") Log.error(f"Full traceback: {traceback.format_exc()}") # Send error result to ElevenLabs if self.elevenlabs_websocket_service and self.elevenlabs_websocket_service.is_connected(): await self.elevenlabs_websocket_service.send_client_tool_result( tool_call_id=tool_call_id, result=f"Error transferring to number: {e}", is_error=True ) # Clear tracking on error self.pending_transfer_tool_id = None self.pending_transfer_target = None else: error_msg = f"No stream_id or transfer target available for transfer (stream_id: {self.connection_manager.state.stream_id}, phone_number: {phone_number}, sip_uri: {sip_uri}, transfer_number: {transfer_number})" Log.warning(f"⚠️ {error_msg}") if self.elevenlabs_websocket_service and self.elevenlabs_websocket_service.is_connected(): await self.elevenlabs_websocket_service.send_client_tool_result( tool_call_id=tool_call_id, result=error_msg, is_error=True ) # Clear tracking on error self.pending_transfer_tool_id = None self.pending_transfer_target = None elif tool_name == "play_keypad_touch_tone": # Handle play keypad touch tone tool request tool_params = tool_request.get("parameters", {}) tone = tool_params.get("tone") if tool_params else None if self.connection_manager.state.stream_id and tone: try: # Create DTMF tone message for Mcube # Note: Mcube may need a specific format for DTMF tones dtmf_message = { "event": "dtmf", "streamId": self.connection_manager.state.stream_id, "tone": tone } await self.connection_manager.send_to_mcube(dtmf_message) Log.info(f"✅ Keypad touch tone played: {tone}") # Send tool result back to ElevenLabs if self.elevenlabs_websocket_service and self.elevenlabs_websocket_service.is_connected(): await self.elevenlabs_websocket_service.send_client_tool_result( tool_call_id=tool_call_id, result=f"Keypad tone {tone} played successfully", is_error=False ) except Exception as e: import traceback Log.error(f"❌ Error playing keypad tone: {e}") Log.error(f"Full traceback: {traceback.format_exc()}") # Send error result to ElevenLabs if self.elevenlabs_websocket_service and self.elevenlabs_websocket_service.is_connected(): await self.elevenlabs_websocket_service.send_client_tool_result( tool_call_id=tool_call_id, result=f"Error playing keypad tone: {e}", is_error=True ) else: error_msg = "No stream_id or tone available for keypad tone" Log.warning(f"⚠️ {error_msg}") if self.elevenlabs_websocket_service and self.elevenlabs_websocket_service.is_connected(): await self.elevenlabs_websocket_service.send_client_tool_result( tool_call_id=tool_call_id, result=error_msg, is_error=True ) else: Log.info(f"⚠️ Unknown tool request: {tool_name} - ignoring") except Exception as e: Log.error(f"❌ Error handling tool request: {e}") async def cleanup(self) -> None: """Cleanup call session and release resources.""" try: Log.info(f"🧹 Starting cleanup for session {self.session_id}") # Release ElevenLabs connection slot from services.elevenlabs_connection_limiter import elevenlabs_limiter await elevenlabs_limiter.release(self.session_id) # Cancel audio queue processing task if self.elevenlabs_audio_queue_task and not self.elevenlabs_audio_queue_task.done(): self.elevenlabs_audio_queue_task.cancel() try: await self.elevenlabs_audio_queue_task except asyncio.CancelledError: pass # Cancel connection retry task if self.elevenlabs_connection_retry_task and not self.elevenlabs_connection_retry_task.done(): self.elevenlabs_connection_retry_task.cancel() try: await self.elevenlabs_connection_retry_task except asyncio.CancelledError: pass # Clear audio queue while not self.elevenlabs_audio_queue.empty(): try: self.elevenlabs_audio_queue.get_nowait() self.elevenlabs_audio_queue.task_done() except asyncio.QueueEmpty: break if self.elevenlabs_websocket_service: try: await self.elevenlabs_websocket_service.close() except Exception as el_error: Log.error(f"Error closing ElevenLabs WebSocket: {el_error}") if self.elevenlabs_receiver_task: self.elevenlabs_receiver_task.cancel() try: await self.elevenlabs_receiver_task except asyncio.CancelledError: pass self.is_active = False except Exception as e: Log.error(f"❌ Error cleaning up: {e}") def get_status(self) -> Dict[str, Any]: """Get current status of this call session.""" return { "session_id": self.session_id, "business_id": self.business_id, "call_id": self.call_id, "is_active": self.is_active, "conversation_count": self.conversation_count, }