Update app.py

app.py

@@ -1,25 +1,18 @@
 #!/usr/bin/env python3
 #
-#  san_integration_script.py (v5 - Provider Format Fix)
-#  ======================================================
+#  san_integration_script.py (v6 - Simplified Audio Processing)
+#  ============================================================
 #  Description:
 #  - Establishes a real-time, two-way audio bridge between a SAN system
 #    and the Millis AI platform.
-#  - Dynamically detects the audio format from the SAN `start` event.
-#  - Forwards inbound audio to Millis AI at 16kHz for processing.
-#  - Receives the AI's audio response at 16kHz.
-#  - Streams the audio back to the SAN system using the exact format
-#    it originally specified.
+#  - Direct audio forwarding at 16kHz PCM format (no conversion needed).
+#  - Removed buffer logic for direct streaming.
 #
 #  Changes in this version:
-#  - Fixed the `reverse-media` event payload to match the provider's
-#    expected format (simplified JSON, lowercase 'callid').
-#  - Fixed `ImportError` by changing `starlette.websockets.State` to
-#    `starlette.websockets.WebSocketState`.
-#  - Updated the final connection check to use `WebSocketState.DISCONNECTED`.
-#  - Added comprehensive logging for debugging audio flow.
-#  - Fixed agent ID extraction to use platform_agent_id from extraParams.
-#  - Added MongoDB integration for credential management.
+#  - Removed audio format conversion logic (resampling)
+#  - Removed buffer management and chunking
+#  - Simplified audio flow for direct 16kHz PCM streaming
+#  - Removed scipy dependency
 #  -------------------------------------------------------------------
 
 import os
@@ -35,8 +28,6 @@ from dotenv import load_dotenv
 load_dotenv()
 
 # Third-party libraries
-import numpy as np
-from scipy import signal as scipy_signal
 import websockets
 from websockets.connection import State as WsState
 from fastapi import FastAPI, WebSocket, WebSocketDisconnect
@@ -54,7 +45,6 @@ logging.basicConfig(
 logger = logging.getLogger("san-integration-app")
 
 # ---------- Environment & Configuration -----------------------------------
-#AGENT_ID = os.getenv("MILLIS_AGENT_ID")
 PUBLIC_KEY = os.getenv("MILLIS_PUBLIC_KEY")
 MILLIS_WS_URI = "wss://api-west.millis.ai:8080/millis"
 
@@ -66,40 +56,15 @@ MONGODB_COLLECTION = os.getenv("MONGODB_COLLECTION", "call_metadata")
 # ---------- FastAPI Application -------------------------------------------
 app = FastAPI()
 
-# def validate_environment():
-#     """Validate that all required environment variables are set."""
-#     required_vars = {
-#         "MILLIS_AGENT_ID": AGENT_ID,
-#         "MILLIS_PUBLIC_KEY": PUBLIC_KEY,
-#         "MONGODB_CONNECTION_STRING": MONGODB_CONNECTION_STRING,
-#         "MONGODB_DATABASE_NAME": MONGODB_DATABASE_NAME,
-#         "MONGODB_COLLECTION": MONGODB_COLLECTION
-#     }
-    
-#     missing_vars = [var for var, value in required_vars.items() if not value]
-    
-#     if missing_vars:
-#         error_msg = f"Missing required environment variables: {', '.join(missing_vars)}"
-#         logger.error(error_msg)
-#         logger.error("Please set these variables in your .env file or environment")
-#         raise ValueError(error_msg)
-
 @app.on_event("startup")
 async def startup_event():
     """Initialize MongoDB connection on startup."""
     logger.info("=== APPLICATION STARTUP ===")
     
-    # Validate environment variables first
-    #validate_environment()
-    
     success = await connect_mongodb()
     if not success:
         logger.error("Failed to connect to MongoDB during startup")
         logger.warning("Application will continue running but MongoDB features will be disabled")
-        logger.info("To fix this issue:")
-        logger.info("1. Ensure MongoDB is installed and running")
-        logger.info("2. Set MONGODB_CONNECTION_STRING environment variable")
-        logger.info("3. For local development: mongodb://localhost:27017")
     else:
         logger.info("MongoDB connection established successfully")
     logger.info("=== STARTUP COMPLETE ===")
@@ -123,7 +88,6 @@ async def connect_mongodb():
         
         if not MONGODB_CONNECTION_STRING or MONGODB_CONNECTION_STRING == "MONGODB_DATABASE_NAME":
             logger.error("Invalid MongoDB connection string. Please set MONGODB_CONNECTION_STRING environment variable.")
-            logger.error("Example: mongodb://localhost:27017 or mongodb://username:password@host:port/database")
             return False
             
         mongodb_client = AsyncIOMotorClient(MONGODB_CONNECTION_STRING)
@@ -135,8 +99,6 @@ async def connect_mongodb():
         return True
     except Exception as e:
         logger.error(f"Failed to connect to MongoDB: {e}")
-        logger.error("Please ensure MongoDB is running and the connection string is correct.")
-        logger.error("For local development, try: mongodb://localhost:27017")
         return False
 
 async def close_mongodb():
@@ -147,16 +109,7 @@ async def close_mongodb():
         logger.info("MongoDB connection closed")
 
 def filter_metadata_for_millis(document: dict) -> dict:
-    """
-    Filter MongoDB document to exclude unwanted fields before sending to Millis AI.
-    
-    Args:
-        document: The MongoDB document containing call data
-        
-    Returns:
-        Filtered dictionary with only the fields that should be sent to Millis AI
-    """
-    # Fields to exclude from metadata sent to Millis AI
+    """Filter MongoDB document to exclude unwanted fields before sending to Millis AI."""
     excluded_fields = {
         "public_key", "call_id", "platform_agent_id", "stored_at", "agent_id"
     }
@@ -165,38 +118,25 @@ def filter_metadata_for_millis(document: dict) -> dict:
     
     for key, value in document.items():
         if key not in excluded_fields:
-            # Convert value to string if it's not None
             if value is not None:
                 filtered_metadata[key] = str(value)
             else:
                 filtered_metadata[key] = ""
     
-    logger.info(f"Filtered metadata - Original fields: {list(document.keys())}")
-    logger.info(f"Filtered metadata - Excluded fields: {list(excluded_fields)}")
     logger.info(f"Filtered metadata - Final fields: {list(filtered_metadata.keys())}")
-    
     return filtered_metadata
 
 async def fetch_call_credentials(call_id: str) -> Dict[str, Any]:
-    """
-    Fetch call credentials and metadata from MongoDB.
-    
-    Args:
-        call_id: The call ID to look up
-        
-    Returns:
-        Dictionary containing platform_agent_id, public_key, metadata, and agent_id
-    """
+    """Fetch call credentials and metadata from MongoDB."""
     if mongodb_db is None:
         logger.error("MongoDB not connected")
         return {}
     
     try:
         logger.info(f"Fetching credentials for call_id: {call_id}")
-        logger.info(f"MongoDB collection: {MONGODB_COLLECTION}")
         collection = mongodb_db[MONGODB_COLLECTION]
         
-        # Query for the call metadata - try multiple field names
+        # Query for the call metadata
         possible_queries = [
             {"call_id": call_id},
             {"metadata.call_id": call_id},
@@ -205,53 +145,25 @@ async def fetch_call_credentials(call_id: str) -> Dict[str, Any]:
         ]
         
         document = None
-        for i, query in enumerate(possible_queries):
-            logger.info(f"MongoDB query attempt {i+1}: {query}")
+        for query in possible_queries:
             document = await collection.find_one(query)
             if document:
-                logger.info(f"Found document with query: {query}")
                 break
         
-        if not document:
-            logger.info(f"No document found for call_id: {call_id}")
-            logger.info("Available documents in collection:")
-            all_docs = await collection.find({}).to_list(length=10)
-            for doc in all_docs:
-                if "call_id" in doc:
-                    logger.info(f"  - call_id: {doc['call_id']}")
-                if "metadata" in doc and "call_id" in doc["metadata"]:
-                    logger.info(f"  - metadata.call_id: {doc['metadata']['call_id']}")
-            
-            # Use the first available document with metadata as fallback
-            for doc in all_docs:
-                if "metadata" in doc and doc["metadata"]:
-                    logger.info(f"Using fallback document with call_id: {doc.get('call_id', 'unknown')}")
-                    document = doc
-                    break
-        
         if not document:
             logger.warning(f"No credentials found for call_id: {call_id}")
-            logger.info("MongoDB query returned: None")
             return {}
         
-        # Log document with masked sensitive data
-        safe_document = document.copy()
-        if "public_key" in safe_document:
-            safe_document["public_key"] = safe_document["public_key"][:10] + "..." if safe_document["public_key"] else "None"
-        logger.info(f"MongoDB document found: {safe_document}")
-        
         # Filter the document to exclude unwanted fields for Millis AI
         filtered_metadata = filter_metadata_for_millis(document)
         
-        # Extract the required fields
         credentials = {
             "platform_agent_id": document.get("platform_agent_id"),
             "public_key": document.get("public_key"),
             "metadata": filtered_metadata
         }
         
-        logger.info(f"Retrieved credentials for call_id {call_id}: {credentials}")
-        logger.info(f"Metadata keys: {list(credentials.get('metadata', {}).keys())}")
+        logger.info(f"Retrieved credentials for call_id {call_id}")
         return credentials
         
     except Exception as e:
@@ -259,17 +171,7 @@ async def fetch_call_credentials(call_id: str) -> Dict[str, Any]:
         return {}
 
 async def fetch_call_credentials_with_fallback(call_id: str, fallback_msg: dict) -> Dict[str, Any]:
-    """
-    Fetch call credentials from MongoDB with fallback to message data.
-    
-    Args:
-        call_id: The call ID to look up
-        fallback_msg: The original message to extract fallback credentials from
-        
-    Returns:
-        Dictionary containing platform_agent_id, public_key, and agent_id
-    """
-    # Try MongoDB first
+    """Fetch call credentials from MongoDB with fallback to message data."""
     logger.info(f"=== MONGODB FALLBACK FOR CALL {call_id} ===")
     credentials = await fetch_call_credentials(call_id)
     
@@ -277,18 +179,16 @@ async def fetch_call_credentials_with_fallback(call_id: str, fallback_msg: dict)
         logger.info(f"Using MongoDB credentials for call_id: {call_id}")
         return credentials
     
-    # Fallback to message data (safety net)
+    # Fallback to message data
     logger.warning(f"MongoDB credentials incomplete for call_id {call_id}, using message fallback")
     
     extra_params = fallback_msg.get("extraParams", {})
     custom_field = fallback_msg.get("custom_field", {})
     
-    # Extract agent_id from fallback sources
     agent_id = (extra_params.get("platform_agent_id") or 
                 custom_field.get("agentId") or 
                 fallback_msg.get("agentId"))
     
-    # Extract public_key from fallback sources
     public_key = (extra_params.get("publicKey") or extra_params.get("public_key") or 
                   custom_field.get("publicKey") or custom_field.get("public_key") or 
                   fallback_msg.get("publicKey") or fallback_msg.get("public_key"))
@@ -299,24 +199,17 @@ async def fetch_call_credentials_with_fallback(call_id: str, fallback_msg: dict)
         "metadata": {}
     }
     
-    logger.warning(f"Using fallback credentials for call_id {call_id}: {fallback_credentials}")
+    logger.warning(f"Using fallback credentials for call_id {call_id}")
     return fallback_credentials
 
 # ---------------------------------------------------------------------------#
-#                          REAL-TIME AUDIO PROCESSOR                         #
+#                          SIMPLIFIED AUDIO PROCESSOR                        #
 # ---------------------------------------------------------------------------#
 class RealTimeAudioProcessor:
     """
-    Manages a single live call, bridging audio between the SAN system and Millis AI.
-    Optimized for faster interruption detection.
+    Simplified audio processor that directly forwards 16kHz PCM audio
+    between SAN system and Millis AI without buffering or conversion.
     """
-    PHONE_RATE = 8000
-    MILLIS_RATE = 16000
-    CHUNK_MS = 40  # Reduced from 20ms to 10ms for faster processing
-    BYTES_PER_SAMPLE = 2
-
-    MILLIS_CHUNK_SIZE = int(MILLIS_RATE * CHUNK_MS / 1000 * BYTES_PER_SAMPLE)
-    PHONE_CHUNK_SIZE = int(PHONE_RATE * CHUNK_MS / 1000 * BYTES_PER_SAMPLE)
 
     def __init__(self, agent_id: str, public_key: str, metadata: dict = None):
         self.agent_id = agent_id
@@ -324,53 +217,13 @@ class RealTimeAudioProcessor:
         self.metadata = metadata or {}
         self.ws: Optional[websockets.WebSocketClientProtocol] = None
         self.connected = False
-
-        self.inbound = bytearray()
-        self.outbound = bytearray()
-        self.in_lock = asyncio.Lock()
-        self.out_lock = asyncio.Lock()
-
         self.is_paused = False
         self.stream_id: Optional[str] = None
         self.call_id: Optional[str] = None
-        self.media_format: dict = {
-            "encoding": "PCM", "sampleRate": self.PHONE_RATE, "channels": 1
-        }
         self._packet_counter = 0
-        
-        # Add interruption detection state
-        self.last_interruption_time = 0
-        self.interruption_threshold = 0.1  # 100ms threshold for interruption detection
-        
-        # Audio level detection configuration
-        self.speech_threshold = 500  # RMS threshold for detecting speech
-        self.silence_threshold = 200  # RMS threshold for detecting silence
-        self.silence_duration = 1.0   # Seconds of silence before unpausing
-        
-        # Debug logging for audio levels (set to True to tune thresholds)
-        self.debug_audio_levels = False
-
-    def enable_audio_debug(self, enabled: bool = True):
-        """Enable or disable debug logging for audio levels to help tune thresholds."""
-        self.debug_audio_levels = enabled
-        logger.info(f"Audio level debug logging {'enabled' if enabled else 'disabled'}")
-
-    def set_audio_thresholds(self, speech_threshold: int = None, silence_threshold: int = None, silence_duration: float = None):
-        """Set audio detection thresholds for interruption detection."""
-        if speech_threshold is not None:
-            self.speech_threshold = speech_threshold
-        if silence_threshold is not None:
-            self.silence_threshold = silence_threshold
-        if silence_duration is not None:
-            self.silence_duration = silence_duration
-        
-        logger.info(f"Audio thresholds updated - Speech: {self.speech_threshold}, Silence: {self.silence_threshold}, Duration: {self.silence_duration}s")
 
     async def connect(self) -> bool:
         logger.info(f"Connecting to Millis AI for call {self.call_id}...")
-        logger.info(f"Agent ID: {self.agent_id}")
-        logger.info(f"Public Key: {self.public_key[:10]}...")
-        logger.info(f"Millis URI: {MILLIS_WS_URI}")
         
         try:
             logger.info("Establishing WebSocket connection...")
@@ -388,27 +241,16 @@ class RealTimeAudioProcessor:
                     "include_metadata_in_prompt": True
                 }
             }
-            logger.info(f"Millis connection metadata: {self.metadata}")
-            logger.info(f"Metadata keys being sent: {list(self.metadata.keys())}")
-            logger.info(f"Sample metadata values:")
-            for key, value in list(self.metadata.items())[:5]:  # Show first 5 items
-                logger.info(f"  {key}: {value}")
-            logger.info(f"Total metadata fields: {len(self.metadata)}")
-            logger.info(f"include_metadata_in_prompt: {initiate_payload['data']['include_metadata_in_prompt']}")
-            logger.info(f"Sending initiate payload: {json.dumps(initiate_payload, indent=2)}")
             
+            logger.info(f"Sending initiate payload to Millis AI")
             await self.ws.send(json.dumps(initiate_payload))
-            logger.info("Initiate payload sent, waiting for response...")
             
             logger.info("Waiting for Millis AI response...")
             msg = await asyncio.wait_for(self.ws.recv(), timeout=10)
-            logger.info(f"Received response: {msg}")
             
             try:
                 parsed_msg = json.loads(msg)
-                logger.info(f"Parsed response: {json.dumps(parsed_msg, indent=2)}")
                 method = parsed_msg.get("method")
-                logger.info(f"Response method: {method}")
                 
                 if method != "onready":
                     logger.error(f"Expected 'onready' method, got '{method}'")
@@ -420,27 +262,8 @@ class RealTimeAudioProcessor:
                 
             except json.JSONDecodeError as e:
                 logger.error(f"Failed to parse response as JSON: {e}")
-                logger.error(f"Raw response: {msg}")
                 raise
                 
-        except asyncio.TimeoutError:
-            logger.error("Connection timeout - Millis AI did not respond within 10 seconds")
-            logger.error(f"Agent ID: {self.agent_id}")
-            logger.error(f"Public Key: {self.public_key[:10]}...")
-            logger.error("Please check:")
-            logger.error("1. Network connectivity to Millis AI")
-            logger.error("2. Agent ID and Public Key are valid")
-            logger.error("3. Millis AI service is running")
-            self.connected = False
-            return False
-        except websockets.exceptions.InvalidURI:
-            logger.error(f"Invalid WebSocket URI: {MILLIS_WS_URI}")
-            self.connected = False
-            return False
-        except websockets.exceptions.ConnectionClosed:
-            logger.error("WebSocket connection was closed unexpectedly")
-            self.connected = False
-            return False
         except Exception as e:
             logger.error(f"Millis AI connection failed: {type(e).__name__}: {e}")
             self.connected = False
@@ -453,143 +276,36 @@ class RealTimeAudioProcessor:
         self.ws = None
         logger.info("Disconnected from Millis AI.")
 
-    @staticmethod
-    def _resample(data: bytes, from_rate: int, to_rate: int) -> bytes:
-        if not data: return b""
-        arr = np.frombuffer(data, dtype=np.int16)
-        if arr.size == 0: return b""
-        new_len = int(arr.size * to_rate / from_rate)
-        resampled = scipy_signal.resample(arr, new_len).astype(np.int16)
-        return resampled.tobytes()
-
-    async def _clear_buffers_immediately(self):
-        """Immediately clear audio buffers for faster interruption response."""
-        async with self.in_lock:
-            self.inbound.clear()
-        async with self.out_lock:
-            self.outbound.clear()
-        logger.info("Audio buffers cleared immediately")
-
-    async def _detect_interruption(self, audio_data: bytes) -> bool:
-        """
-        Detect if user is speaking (interruption) by analyzing audio levels.
-        Returns True if interruption is detected.
-        """
-        if not audio_data:
-            return False
+    async def _forward_to_millis(self, audio_data: bytes):
+        """Forward audio data directly to Millis AI."""
+        if not self.connected or not self.ws:
+            return
             
         try:
-            # Convert bytes to numpy array
-            audio_array = np.frombuffer(audio_data, dtype=np.int16)
-            if len(audio_array) == 0:
-                return False
-                
-            # Calculate RMS (Root Mean Square) to detect audio level
-            rms = np.sqrt(np.mean(audio_array.astype(np.float32) ** 2))
-            
-            # Debug logging for threshold tuning
-            if self.debug_audio_levels and rms > 100:  # Only log when there's significant audio
-                logger.debug(f"Audio RMS: {rms:.2f}, Speech threshold: {self.speech_threshold}")
-            
-            if rms > self.speech_threshold:
-                current_time = asyncio.get_event_loop().time()
-                if current_time - self.last_interruption_time > self.interruption_threshold:
-                    self.last_interruption_time = current_time
-                    logger.info(f"Interruption detected! RMS: {rms:.2f}, Threshold: {self.speech_threshold}")
-                    return True
-                    
-        except Exception as e:
-            logger.warning(f"Error in interruption detection: {e}")
-            
-        return False
-
-    async def _detect_silence(self, audio_data: bytes) -> bool:
-        """
-        Detect if user has stopped speaking (silence) by analyzing audio levels.
-        Returns True if silence is detected.
-        """
-        if not audio_data:
-            return False
+            await self.ws.send(audio_data)
+            self._packet_counter += 1
             
-        try:
-            # Convert bytes to numpy array
-            audio_array = np.frombuffer(audio_data, dtype=np.int16)
-            if len(audio_array) == 0:
-                return False
+            if self._packet_counter >= 1000:
+                logger.info("Sending ping to Millis AI")
+                await self.ws.send(json.dumps({"method": "ping"}))
+                self._packet_counter = 0
                 
-            # Calculate RMS (Root Mean Square) to detect audio level
-            rms = np.sqrt(np.mean(audio_array.astype(np.float32) ** 2))
-            
-            # Debug logging for threshold tuning
-            if self.debug_audio_levels and rms < 300:  # Only log when audio is low
-                logger.debug(f"Silence check - RMS: {rms:.2f}, Silence threshold: {self.silence_threshold}")
-            
-            if rms < self.silence_threshold:
-                current_time = asyncio.get_event_loop().time()
-                # Only unpause after configured silence duration to avoid rapid toggling
-                if current_time - self.last_interruption_time > self.silence_duration:
-                    logger.info(f"Silence detected! RMS: {rms:.2f}, Threshold: {self.silence_threshold}")
-                    return True
-                    
         except Exception as e:
-            logger.warning(f"Error in silence detection: {e}")
-            
-        return False
-
-    async def _pump_inbound_to_millis(self):
-        logger.info(f"Starting inbound audio pump for call {self.call_id}")
-        processed_chunks = 0
-        while self.connected:
-            chunk8 = None
-            async with self.in_lock:
-                # Process smaller chunks for faster response
-                min_chunk_size = max(self.PHONE_CHUNK_SIZE // 2, 160)  # At least 10ms of audio
-                if len(self.inbound) >= min_chunk_size:
-                    chunk8 = self.inbound[:min_chunk_size]
-                    del self.inbound[:min_chunk_size]
-            if not chunk8:
-                await asyncio.sleep(0.002)  # Reduced sleep time
-                continue
-            try:
-                # Check for interruption before sending to Millis
-                if await self._detect_interruption(chunk8):
-                    # Immediately pause AI and clear buffers
-                    self.is_paused = True
-                    await self._clear_buffers_immediately()
-                    logger.info("AI paused due to user interruption")
-                elif self.is_paused and await self._detect_silence(chunk8):
-                    # Unpause AI when user stops speaking
-                    self.is_paused = False
-                    logger.info("AI unpaused due to user silence")
-                    
-                chunk16 = self._resample(chunk8, self.PHONE_RATE, self.MILLIS_RATE)
-                processed_chunks += 1
-                # Only log every 200 chunks to reduce noise
-                if processed_chunks % 200 == 0:
-                    logger.info(f"Processed {processed_chunks} audio chunks to Millis AI")
-                await self.ws.send(chunk16)
-                self._packet_counter += 1
-                if self._packet_counter >= 1_000:
-                    logger.info("Sending ping to Millis AI")
-                    await self.ws.send(json.dumps({"method": "ping"}))
-                    self._packet_counter = 0
-            except Exception as e:
-                logger.error(f"Error in _pump_inbound_to_millis: {e}")
-                self.connected = False
+            logger.error(f"Error forwarding audio to Millis: {e}")
+            self.connected = False
 
-    async def _pump_millis_to_outbound(self):
-        logger.info(f"Starting outbound audio pump for call {self.call_id}")
-        received_chunks = 0
+    async def _handle_millis_messages(self, client_ws: WebSocket):
+        """Handle messages from Millis AI and forward audio to SAN."""
+        logger.info(f"Starting Millis message handler for call {self.call_id}")
+        
         while self.connected and self.ws and self.ws.state == WsState.OPEN:
             try:
                 msg = await self.ws.recv()
+                
                 if isinstance(msg, bytes):
-                    received_chunks += 1
-                    # Only log every 200 chunks to reduce noise
-                    if received_chunks % 200 == 0:
-                        logger.info(f"Received {received_chunks} audio chunks from Millis AI")
-                    async with self.out_lock: 
-                        self.outbound.extend(msg)
+                    # Direct audio data - forward to SAN if not paused
+                    if not self.is_paused:
+                        await self._forward_to_san(client_ws, msg)
                     continue
                 
                 # Handle JSON messages
@@ -597,85 +313,58 @@ class RealTimeAudioProcessor:
                     evt = json.loads(msg)
                 except json.JSONDecodeError as e:
                     logger.warning(f"Failed to parse JSON message from Millis AI: {e}")
-                    logger.warning(f"Raw message: {msg}")
                     continue
                 
                 method = evt.get("method")
-                # Only log important JSON events, not every audio chunk
                 if method not in ["ping", "pong"]:
-                    logger.info(f"Millis event: {method} - {evt.get('data', '')}")
+                    logger.info(f"Millis event: {method}")
+                    
                 if method == "pause": 
                     self.is_paused = True
                     logger.info("Audio paused")
-                    # Immediately clear buffers when paused
-                    await self._clear_buffers_immediately()
                 elif method == "unpause": 
                     self.is_paused = False
                     logger.info("Audio unpaused")
                 elif method in ("clear", "start_answering"):
-                    await self._clear_buffers_immediately()
                     self.is_paused = False
-                    logger.info("Audio buffer cleared")
+                    logger.info("Audio cleared/started")
+                    
             except websockets.exceptions.ConnectionClosed:
                 logger.warning("Millis AI closed the connection.")
                 self.connected = False
             except Exception as e:
-                logger.warning(f"Error reading from Millis AI: {type(e).__name__}: {e}")
-                logger.warning(f"Error details: {str(e)}")
+                logger.error(f"Error in Millis message handler: {e}")
                 self.connected = False
 
-    async def _pump_outbound_to_carrier(self, client_ws: WebSocket):
-        logger.info(f"Starting carrier outbound pump for call {self.call_id}")
-        sent_packets = 0
-        while self.connected:
-            if self.is_paused:
-                await asyncio.sleep(0.005)  # Reduced sleep time
-                continue
-            chunk16 = None
-            async with self.out_lock:
-                # Process smaller chunks for faster response
-                min_chunk_size = max(self.MILLIS_CHUNK_SIZE // 2, 320)  # At least 10ms of audio
-                if len(self.outbound) >= min_chunk_size:
-                    chunk16 = self.outbound[:min_chunk_size]
-                    del self.outbound[:min_chunk_size]
-            if not chunk16:
-                await asyncio.sleep(0.002)  # Reduced sleep time
-                continue
-            try:
-                target_rate = self.media_format.get("sampleRate", self.PHONE_RATE)
-                chunk_resampled = self._resample(chunk16, self.MILLIS_RATE, target_rate)
-                payload = base64.b64encode(chunk_resampled).decode()
-                sent_packets += 1
-                # Only log every 400 packets to reduce noise
-                if sent_packets % 400 == 0:
-                    logger.info(f"Sent {sent_packets} audio packets to SAN")
-
-                ### --- FIX: Modified the JSON payload to match the provider's simple format --- ###
-                reverse_media_payload = {
-                    "event": "reverse-media",
-                    "callid": self.call_id,  # Changed from "callId" to "callid"
-                    "payload": payload,
-                    # Removed "streamId" and "mediaFormat" fields
-                }
-                await client_ws.send_json(reverse_media_payload)
-                ### --- END FIX --- ###
-
-            except Exception as e:
-                logger.error(f"Error in _pump_outbound_to_carrier: {e}")
-                break
+    async def _forward_to_san(self, client_ws: WebSocket, audio_data: bytes):
+        """Forward audio data to SAN system."""
+        try:
+            payload = base64.b64encode(audio_data).decode()
+            
+            reverse_media_payload = {
+                "event": "reverse-media",
+                "callid": self.call_id,
+                "payload": payload,
+            }
+            
+            await client_ws.send_json(reverse_media_payload)
+            
+        except Exception as e:
+            logger.error(f"Error forwarding audio to SAN: {e}")
 
     async def start(self, client_ws: WebSocket) -> list[asyncio.Task]:
+        """Start the audio processor."""
         logger.info(f"Starting RealTimeAudioProcessor for call {self.call_id}")
+        
         if not await self.connect(): 
             logger.error("Failed to connect to Millis AI")
             return []
         
-        logger.info("Creating audio processing tasks")
+        logger.info("Creating audio processing task")
         tasks = [
-            asyncio.create_task(self._pump_millis_to_outbound()),
-            asyncio.create_task(self._pump_inbound_to_millis()),
-            asyncio.create_task(self._pump_outbound_to_carrier(client_ws)),
+            asyncio.create_task(self._handle_millis_messages(client_ws)),
         ]
+        
         logger.info(f"Created {len(tasks)} tasks")
         return tasks
 
@@ -701,160 +390,72 @@ async def media_socket(ws: WebSocket):
     try:
         while True:
             raw = await ws.receive_text()
-            
             msg = json.loads(raw)
             event = msg.get("event")
             
-            # Enhanced logging for start events only
             if event == "start":
                 logger.info("=== START EVENT ===")
-                logger.info(f"Call ID: {msg.get('callId')}, Stream ID: {msg.get('streamId')}")
-                logger.info(f"Agent ID: {msg.get('extraParams', {}).get('platform_agent_id')}")
-                logger.info("=== END START EVENT ===")
-            elif event == "media":
-                payload_b64 = msg.get("payload")
-                if payload_b64:
-                    pcm = base64.b64decode(payload_b64)
-                    # Reduce logging frequency for media chunks
-                    logger.debug(f"Received media chunk: {len(pcm)} bytes for call")
-                else:
-                    logger.warning("Media event received but no payload found")
-            else:
-                logger.info(f"Received event: {event}")
-
-            if event == "start":
-                logger.info("=== START EVENT PROCESSING ===")
                 new_call_id = msg.get("callId")
                 stream_id = msg.get("streamId")
-                logger.info(f"Start event details - callId: {new_call_id}, streamId: {stream_id}")
+                logger.info(f"Call ID: {new_call_id}, Stream ID: {stream_id}")
 
                 if processor and new_call_id != active_call_id:
-                    logger.info(f"New call detected ({active_call_id} -> {new_call_id}). Stopping old processor.")
+                    logger.info(f"New call detected. Stopping old processor.")
                     await stop_processor(processor, tasks)
                     processor, tasks = None, []
 
                 if processor is None:
                     logger.info(f"Starting processor for call: {new_call_id}")
                     
-                    # Extract credentials directly from message first
-                    logger.info(f"=== EXTRACTING CREDENTIALS FOR CALL {new_call_id} ===")
+                    # Extract credentials
                     extra_params = msg.get("extraParams", {})
-                    
-                    # Get credentials from message
                     agent_id = extra_params.get("platform_agent_id")
-                    
-                    # Always use environment variable for public_key (primary source)
                     public_key = PUBLIC_KEY
-                    logger.info(f"Using environment PUBLIC_KEY: {public_key[:10] + '...' if public_key else 'None'}")
                     
-                    # Use environment variables as fallback for missing agent_id
-                    if not agent_id:
-                        agent_id = AGENT_ID
-                        logger.info(f"Using environment AGENT_ID: {agent_id}")
-                    
-                    # If agent_id still missing, try MongoDB as final fallback
-                    if not agent_id:
-                        logger.warning("Agent ID missing, trying MongoDB fallback")
+                    if not agent_id or not public_key:
                         credentials = await fetch_call_credentials_with_fallback(new_call_id, msg)
-                        agent_id = credentials.get("platform_agent_id") or agent_id
-                        # Keep using environment public_key, don't override
-                    
-                    logger.info(f"Using credentials - agent_id: {agent_id}, public_key: {public_key[:10] + '...' if public_key else 'None'}")
+                        agent_id = agent_id or credentials.get("platform_agent_id")
+                        public_key = public_key or credentials.get("public_key")
                     
                     if not agent_id:
                         logger.error(f"No agent_id found for call_id: {new_call_id}")
-                        logger.error("Closing WebSocket due to missing agent_id")
                         await ws.close(code=1008, reason="Missing agent_id")
                         return
                     
                     if not public_key:
-                        logger.error("No public_key found in environment variables")
-                        logger.error("Closing WebSocket due to missing public_key")
+                        logger.error("No public_key found")
                         await ws.close(code=1008, reason="Missing public_key")
                         return
                     
-                    # Try to get metadata from MongoDB using call_id from extraParams
-                    logger.info("=== FETCHING METADATA ===")
-                    extra_params = msg.get("extraParams", {})
+                    # Get metadata from MongoDB
                     mongodb_call_id = extra_params.get("call_id")
-                    
                     if mongodb_call_id:
-                        logger.info(f"Found call_id in extraParams: {mongodb_call_id}")
                         mongodb_credentials = await fetch_call_credentials(mongodb_call_id)
                         metadata = mongodb_credentials.get("metadata", {})
-                        
-                        if metadata:
-                            logger.info(f"Using MongoDB metadata with {len(metadata)} fields")
-                            logger.info(f"MongoDB metadata keys: {list(metadata.keys())}")
-                            # Merge with extraParams to include platform_agent_id
-                            metadata.update(extra_params)
-                            logger.info(f"Metadata after merging with extraParams: {dict(list(metadata.items())[:3])}")
-                        else:
-                            logger.info("No MongoDB metadata found, using extraParams")
-                            metadata = extra_params
+                        metadata.update(extra_params)
                     else:
-                        logger.info("No call_id in extraParams, trying SAN callId")
-                        mongodb_credentials = await fetch_call_credentials(new_call_id)
-                        metadata = mongodb_credentials.get("metadata", {})
-                        
-                        if metadata:
-                            logger.info(f"Using MongoDB metadata with {len(metadata)} fields")
-                            logger.info(f"MongoDB metadata keys: {list(metadata.keys())}")
-                            metadata.update(extra_params)
-                        else:
-                            logger.info("No MongoDB metadata found, using extraParams")
-                            metadata = extra_params
-                    
-                    logger.info("=== METADATA RESOLVED ===")
+                        metadata = extra_params
                     
-                    logger.info("Agent configuration resolved:")
-                    logger.info(f"  - agent_id: {agent_id}")
-                    logger.info(f"  - public_key: {public_key[:10]}...")
-                    logger.info(f"  - metadata: {metadata}")
-                    
-                    logger.info(f"=== CREATING PROCESSOR FOR CALL {new_call_id} ===")
+                    logger.info(f"Creating processor with agent_id: {agent_id}")
                     processor = RealTimeAudioProcessor(agent_id, public_key, metadata)
                     processor.stream_id = stream_id
                     processor.call_id = new_call_id
-                    
-                    # Set audio thresholds from environment or use defaults
-                    speech_threshold = int(os.getenv("SPEECH_THRESHOLD", "500"))
-                    silence_threshold = int(os.getenv("SILENCE_THRESHOLD", "200"))
-                    silence_duration = float(os.getenv("SILENCE_DURATION", "1.0"))
-                    enable_debug = os.getenv("AUDIO_DEBUG", "false").lower() == "true"
-                    
-                    processor.set_audio_thresholds(speech_threshold, silence_threshold, silence_duration)
-                    if enable_debug:
-                        processor.enable_audio_debug(True)
-                    
-                    logger.info(f"Audio processor configured - Speech: {speech_threshold}, Silence: {silence_threshold}, Duration: {silence_duration}, Debug: {enable_debug}")
-                    
-                    if "mediaFormat" in msg:
-                        processor.media_format = msg["mediaFormat"]
-                        logger.info(f"Captured media format from SAN: {processor.media_format}")
-                    else:
-                        logger.warning("No mediaFormat in 'start' event. Using default.")
 
-                    logger.info(f"=== STARTING PROCESSOR ===")
                     tasks = await processor.start(ws)
                     if not tasks:
                         logger.error("Failed to start processor")
                         await ws.close(code=1011, reason="Could not connect to AI backend.")
                         return
                     active_call_id = new_call_id
-                    logger.info(f"=== START EVENT COMPLETED - {len(tasks)} tasks started ===")
+                    logger.info("=== START EVENT COMPLETED ===")
                 continue
 
             elif event == "media" and processor:
                 payload_b64 = msg.get("payload")
                 if payload_b64:
-                    pcm = base64.b64decode(payload_b64)
-                    async with processor.in_lock: 
-                        processor.inbound.extend(pcm)
-                    # Reduce logging frequency for media processing
-                    logger.debug(f"Added {len(pcm)} bytes to inbound buffer for call {active_call_id}")
-                else:
-                    logger.warning("Media event received but no payload found")
+                    # Decode and directly forward audio data to Millis AI
+                    pcm_data = base64.b64decode(payload_b64)
+                    await processor._forward_to_millis(pcm_data)
                 continue
 
             elif event in ("hangup", "stop", "disconnect"):
@@ -872,9 +473,7 @@ async def media_socket(ws: WebSocket):
 
     except WebSocketDisconnect:
         logger.info("=== WEBSOCKET DISCONNECT ===")
-        logger.info("SAN system disconnected the WebSocket.")
     except Exception as e:
-        logger.error("=== UNHANDLED ERROR ===")
         logger.error(f"Unhandled error in media_socket: {e}", exc_info=True)
     finally:
         logger.info("=== FINAL CLEANUP ===")
@@ -890,7 +489,6 @@ async def media_socket(ws: WebSocket):
 async def health():
     logger.info("Health check endpoint called")
     
-    # Check MongoDB connection status
     mongodb_status = "connected" if mongodb_client is not None and mongodb_db is not None else "disconnected"
     
     return {
@@ -902,7 +500,6 @@ async def health():
             "collection": MONGODB_COLLECTION or "not_set"
         },
         "config": {
-            "agent_id": "not_set",
             "public_key": PUBLIC_KEY[:10] + "..." if PUBLIC_KEY else "not_set",
             "millis_ws_uri": MILLIS_WS_URI
         }
@@ -915,7 +512,7 @@ async def test():
 
 @app.get("/test-mongodb")
 async def test_mongodb():
-    """Test MongoDB connectivity and add sample data."""
+    """Test MongoDB connectivity."""
     logger.info("MongoDB test endpoint called")
     
     if mongodb_db is None:
@@ -924,7 +521,6 @@ async def test_mongodb():
     try:
         collection = mongodb_db[MONGODB_COLLECTION]
         
-        # Test insert
         test_doc = {
             "call_id": "test-call-123",
             "platform_agent_id": "test-agent-456",
@@ -933,9 +529,6 @@ async def test_mongodb():
         }
         
         result = await collection.insert_one(test_doc)
-        logger.info(f"Test document inserted with ID: {result.inserted_id}")
-        
-        # Test query
         retrieved = await collection.find_one({"call_id": "test-call-123"})
         
         return {
@@ -954,56 +547,10 @@ async def test_mongodb():
             "mongodb_connected": mongodb_db is not None
         }
 
-@app.get("/audio-config")
-async def get_audio_config():
-    """Get current audio detection configuration."""
-    return {
-        "chunk_ms": RealTimeAudioProcessor.CHUNK_MS,
-        "phone_chunk_size": RealTimeAudioProcessor.PHONE_CHUNK_SIZE,
-        "millis_chunk_size": RealTimeAudioProcessor.MILLIS_CHUNK_SIZE,
-        "interruption_threshold": 0.1,
-        "speech_threshold": 500,
-        "silence_threshold": 200,
-        "silence_duration": 1.0,
-        "description": "Audio detection configuration for interruption handling"
-    }
-
-@app.post("/audio-config")
-async def update_audio_config(
-    speech_threshold: int = None,
-    silence_threshold: int = None,
-    silence_duration: float = None,
-    enable_debug: bool = False
-):
-    """Update audio detection configuration."""
-    # Update global defaults for new processors
-    if speech_threshold is not None:
-        RealTimeAudioProcessor.speech_threshold = speech_threshold
-    if silence_threshold is not None:
-        RealTimeAudioProcessor.silence_threshold = silence_threshold
-    if silence_duration is not None:
-        RealTimeAudioProcessor.silence_duration = silence_duration
-    
-    logger.info(f"Audio config updated - Speech: {speech_threshold}, Silence: {silence_threshold}, Duration: {silence_duration}, Debug: {enable_debug}")
-    
-    return {
-        "status": "updated",
-        "speech_threshold": speech_threshold,
-        "silence_threshold": silence_threshold,
-        "silence_duration": silence_duration,
-        "enable_debug": enable_debug,
-        "message": "Audio configuration updated. Changes will apply to new connections."
-    }
-
-
 if __name__ == "__main__":
-    print("Starting SAN to Millis AI Integration Server (v7 - MongoDB Integration)...")
+    print("Starting SAN to Millis AI Integration Server (v6 - Simplified)...")
     logger.info("=== SERVER STARTING ===")
     logger.info(f"Public Key: {PUBLIC_KEY[:10] + '...' if PUBLIC_KEY else 'NOT_SET'}")
     logger.info(f"Millis URI: {MILLIS_WS_URI}")
-    logger.info(f"MongoDB URI: {MONGODB_CONNECTION_STRING or 'NOT_SET'}")
-    logger.info(f"MongoDB Database: {MONGODB_DATABASE_NAME or 'NOT_SET'}")
-    logger.info(f"MongoDB Collection: {MONGODB_COLLECTION or 'NOT_SET'}")
     logger.info("=== SERVER READY ===")
     uvicorn.run(app, host="0.0.0.0", port=8000)
-