Refactor audio processing and enhance error handling

- Updated `worker_orchestrator.py` to use `logger.exception` for improved error logging.
- Changed default MongoDB database name from "friend-lite" to "chronicle" in multiple files for consistency.
- Added a new method `close_stream_without_stop` in `audio_stream_client.py` to handle abrupt WebSocket disconnections.
- Enhanced audio validation in `audio_utils.py` to support automatic resampling of audio data if sample rates do not match.
- Improved logging in various modules to provide clearer insights during audio processing and event dispatching.

Author: AnkushMalaker

backends/advanced/src/advanced_omi_backend/app_config.py

@@ -29,8 +29,7 @@ class AppConfig:
     def __init__(self):
         # MongoDB Configuration
         self.mongodb_uri = os.getenv("MONGODB_URI", "mongodb://mongo:27017")
-        # default to legacy value to avoid breaking peoples .env
-        self.mongodb_database = os.getenv("MONGODB_DATABASE", "friend-lite")
+        self.mongodb_database = os.getenv("MONGODB_DATABASE", "chronicle")
         self.mongo_client = AsyncIOMotorClient(self.mongodb_uri)
         self.db = self.mongo_client.get_default_database(self.mongodb_database)
         self.users_col = self.db["users"]

backends/advanced/src/advanced_omi_backend/clients/audio_stream_client.py

@@ -555,6 +555,39 @@ async def _stop():
         logger.info(f"Stream {stream_id} stopped, sent {total_chunks} chunks")
         return total_chunks
 
+    def close_stream_without_stop(self, stream_id: str) -> int:
+        """Close WebSocket connection without sending audio-stop event.
+
+        This simulates abrupt disconnection (network failure, client crash)
+        and should trigger websocket_disconnect end_reason.
+
+        Args:
+            stream_id: Stream session ID
+
+        Returns:
+            Total chunks sent during this session
+        """
+        session = self._sessions.get(stream_id)
+        if not session:
+            raise ValueError(f"Unknown stream_id: {stream_id}")
+
+        async def _close_abruptly():
+            # Just close the connection without audio-stop
+            await session.client.close()
+
+        future = asyncio.run_coroutine_threadsafe(_close_abruptly(), session.loop)
+        future.result(timeout=10)
+
+        # Stop the event loop
+        session.loop.call_soon_threadsafe(session.loop.stop)
+        session.thread.join(timeout=5)
+
+        total_chunks = session.chunk_count
+        del self._sessions[stream_id]
+
+        logger.info(f"Stream {stream_id} closed abruptly (no audio-stop), sent {total_chunks} chunks")
+        return total_chunks
+
     def get_session(self, stream_id: str) -> Optional[StreamSession]:
         """Get session info for a stream."""
         return self._sessions.get(stream_id)

backends/advanced/src/advanced_omi_backend/controllers/audio_controller.py

@@ -94,7 +94,8 @@ async def upload_and_process_audio_files(
                     audio_data, sample_rate, sample_width, channels, duration = await validate_and_prepare_audio(
                         audio_data=content,
                         expected_sample_rate=16000,  # Expecting 16kHz
-                        convert_to_mono=True  # Convert stereo to mono
+                        convert_to_mono=True,  # Convert stereo to mono
+                        auto_resample=True  # Auto-resample if sample rate doesn't match
                     )
                 except AudioValidationError as e:
                     processed_files.append({

backends/advanced/src/advanced_omi_backend/controllers/queue_controller.py

@@ -537,6 +537,7 @@ def start_post_conversation_jobs(
         conversation_id,
         client_id or "",
         user_id,
+        "file_upload",  # Explicit end_reason for file upload processing
         job_timeout=120,  # 2 minutes
         result_ttl=JOB_RESULT_TTL,
         depends_on=[memory_job, title_summary_job],  # Wait for both parallel jobs

backends/advanced/src/advanced_omi_backend/controllers/session_controller.py

@@ -56,12 +56,13 @@ async def mark_session_complete(
         await mark_session_complete(redis, session_id, "all_jobs_complete")
     """
     session_key = f"audio:session:{session_id}"
+    mark_time = time.time()
     await redis_client.hset(session_key, mapping={
         "status": "finished",
-        "completed_at": str(time.time()),
+        "completed_at": str(mark_time),
         "completion_reason": reason
     })
-    logger.info(f"✅ Session {session_id[:12]} marked finished: {reason}")
+    logger.info(f"✅ Session {session_id[:12]} marked finished: {reason} [TIME: {mark_time:.3f}]")
 
 
 async def get_session_info(redis_client, session_id: str) -> Optional[Dict]:

backends/advanced/src/advanced_omi_backend/controllers/websocket_controller.py

@@ -238,6 +238,10 @@ async def cleanup_client_state(client_id: str):
         redis_url = os.getenv("REDIS_URL", "redis://localhost:6379/0")
         async_redis = redis.from_url(redis_url, decode_responses=False)
 
+        # Get audio stream producer for finalization
+        from advanced_omi_backend.services.audio_stream.producer import get_audio_stream_producer
+        audio_stream_producer = get_audio_stream_producer()
+
         # Find all session keys for this client and mark them complete
         pattern = f"audio:session:*"
         cursor = 0
@@ -250,8 +254,18 @@ async def cleanup_client_state(client_id: str):
                 # Check if this session belongs to this client
                 client_id_bytes = await async_redis.hget(key, "client_id")
                 if client_id_bytes and client_id_bytes.decode() == client_id:
-                    # Mark session as complete (WebSocket disconnected)
                     session_id = key.decode().replace("audio:session:", "")
+
+                    # Check session status
+                    status_bytes = await async_redis.hget(key, "status")
+                    status = status_bytes.decode() if status_bytes else None
+
+                    # If session is still active, finalize it first (sets status + completion_reason atomically)
+                    if status in ["active", None]:
+                        logger.info(f"📊 Finalizing active session {session_id[:12]} due to WebSocket disconnect")
+                        await audio_stream_producer.finalize_session(session_id, completion_reason="websocket_disconnect")
+
+                    # Mark session as complete (WebSocket disconnected)
                     await mark_session_complete(async_redis, session_id, "websocket_disconnect")
                     sessions_closed += 1
 
@@ -485,8 +499,8 @@ async def _finalize_streaming_session(
         # Send end-of-session signal to workers
         await audio_stream_producer.send_session_end_signal(session_id)
 
-        # Mark session as finalizing
-        await audio_stream_producer.finalize_session(session_id)
+        # Mark session as finalizing with user_stopped reason (audio-stop event)
+        await audio_stream_producer.finalize_session(session_id, completion_reason="user_stopped")
 
         # NOTE: Finalize job disabled - open_conversation_job now handles everything
         # The open_conversation_job will:

backends/advanced/src/advanced_omi_backend/database.py

@@ -14,7 +14,7 @@
 
 # MongoDB Configuration
 MONGODB_URI = os.getenv("MONGODB_URI", "mongodb://mongo:27017")
-MONGODB_DATABASE = os.getenv("MONGODB_DATABASE", "friend-lite")
+MONGODB_DATABASE = os.getenv("MONGODB_DATABASE", "chronicle")
 
 mongo_client = AsyncIOMotorClient(
     MONGODB_URI,

backends/advanced/src/advanced_omi_backend/middleware/app_middleware.py

@@ -60,6 +60,7 @@ class RequestLoggingMiddleware(BaseHTTPMiddleware):
         "/health",
         "/auth/health",
         "/readiness",
+        "/api/queue/dashboard",  # Auto-refresh endpoint, too noisy
     }
 
     # Binary content types to exclude

backends/advanced/src/advanced_omi_backend/models/job.py

@@ -44,7 +44,7 @@ async def _ensure_beanie_initialized():
             mongodb_uri = os.getenv("MONGODB_URI", "mongodb://localhost:27017")
 
             # Create MongoDB client
-            mongodb_database = os.getenv("MONGODB_DATABASE", "friend-lite")
+            mongodb_database = os.getenv("MONGODB_DATABASE", "chronicle")
             client = AsyncIOMotorClient(mongodb_uri)
             try:
                 database = client.get_default_database(mongodb_database)

backends/advanced/src/advanced_omi_backend/plugins/router.py

@@ -122,23 +122,37 @@ async def dispatch_event(
         Returns:
             List of plugin results
         """
+        # Add at start
+        logger.info(f"🔌 ROUTER: Dispatching '{event}' event (user={user_id})")
+
         results = []
 
         # Get plugins subscribed to this event
         plugin_ids = self._plugins_by_event.get(event, [])
 
+        # Add subscription check
+        if not plugin_ids:
+            logger.warning(f"🔌 ROUTER: No plugins subscribed to event '{event}'")
+            return results
+
+        logger.info(f"🔌 ROUTER: Found {len(plugin_ids)} subscribed plugin(s): {plugin_ids}")
+
         for plugin_id in plugin_ids:
             plugin = self.plugins[plugin_id]
 
             if not plugin.enabled:
+                logger.info(f"   ⊘ Skipping '{plugin_id}': disabled")
                 continue
 
             # Check execution condition (wake_word, etc.)
+            logger.info(f"   → Checking execution condition for '{plugin_id}'")
             if not await self._should_execute(plugin, data):
+                logger.info(f"   ⊘ Skipping '{plugin_id}': condition not met")
                 continue
 
             # Execute plugin
             try:
+                logger.info(f"   ▶ Executing '{plugin_id}' for event '{event}'")
                 context = PluginContext(
                     user_id=user_id,
                     event=event,
@@ -149,15 +163,30 @@ async def dispatch_event(
                 result = await self._execute_plugin(plugin, event, context)
 
                 if result:
+                    status_icon = "✓" if result.success else "✗"
+                    logger.info(
+                        f"   {status_icon} Plugin '{plugin_id}' completed: "
+                        f"success={result.success}, message={result.message}"
+                    )
                     results.append(result)
 
                     # If plugin says stop processing, break
                     if not result.should_continue:
-                        logger.info(f"Plugin '{plugin_id}' stopped further processing")
+                        logger.info(f"   ⊗ Plugin '{plugin_id}' stopped further processing")
                         break
 
             except Exception as e:
-                logger.error(f"Error executing plugin '{plugin_id}': {e}", exc_info=True)
+                # CRITICAL: Log exception details
+                logger.error(
+                    f"   ✗ Plugin '{plugin_id}' FAILED with exception: {e}",
+                    exc_info=True
+                )
+
+        # Add at end
+        logger.info(
+            f"🔌 ROUTER: Dispatch complete for '{event}': "
+            f"{len(results)} plugin(s) executed successfully"
+        )
 
         return results