Added queue management page and linked repreocess transcription

backends/advanced/src/advanced_omi_backend/config.py

@@ -37,6 +37,12 @@
     "speech_inactivity_threshold": 60,      # Speech gap threshold for closure (1 minute)
 }
 
+# Default audio storage settings
+DEFAULT_AUDIO_STORAGE_SETTINGS = {
+    "audio_base_path": "/app/data",  # Main audio directory (where volume is mounted)
+    "audio_chunks_path": "/app/data/audio_chunks",  # Full path to audio chunks subfolder
+}
+
 # Global cache for diarization settings
 _diarization_settings = None
 
@@ -140,5 +146,18 @@ def get_conversation_stop_settings():
     }
 
 
+def get_audio_storage_settings():
+    """Get audio storage settings from environment or defaults."""
+
+    # Get base path and derive chunks path
+    audio_base_path = os.getenv("AUDIO_BASE_PATH", DEFAULT_AUDIO_STORAGE_SETTINGS["audio_base_path"])
+    audio_chunks_path = os.getenv("AUDIO_CHUNKS_PATH", f"{audio_base_path}/audio_chunks")
+
+    return {
+        "audio_base_path": audio_base_path,
+        "audio_chunks_path": audio_chunks_path,
+    }
+
+
 # Initialize settings on module load
 _diarization_settings = load_diarization_settings_from_file()

backends/advanced/src/advanced_omi_backend/database.py

@@ -706,6 +706,50 @@ async def activate_transcript_version(self, conversation_id: str, version_id: st
             logger.info(f"Activated transcript version {version_id} for conversation {conversation_id}")
         return result.modified_count > 0
 
+    async def update_transcript_version(
+        self, 
+        conversation_id: str, 
+        version_id: str, 
+        transcript: str = None, 
+        segments: list = None, 
+        processing_time_seconds: float = None,
+        provider: str = None,
+        model: str = None
+    ) -> bool:
+        """Update a specific transcript version with processing results."""
+        update_fields = {}
+
+        if transcript is not None:
+            update_fields["transcript_versions.$.transcript"] = transcript
+        if segments is not None:
+            update_fields["transcript_versions.$.segments"] = segments
+        if processing_time_seconds is not None:
+            update_fields["transcript_versions.$.processing_time_seconds"] = processing_time_seconds
+        if provider is not None:
+            update_fields["transcript_versions.$.provider"] = provider
+        if model is not None:
+            update_fields["transcript_versions.$.model"] = model
+
+        # Always update the completion timestamp
+        update_fields["transcript_versions.$.completed_at"] = datetime.now(UTC).isoformat()
+        update_fields["transcript_versions.$.status"] = "completed"
+
+        if not update_fields:
+            return False
+
+        result = await self.col.update_one(
+            {
+                "conversation_id": conversation_id,
+                "transcript_versions.version_id": version_id
+            },
+            {"$set": update_fields}
+        )
+
+        if result.modified_count > 0:
+            logger.info(f"Updated transcript version {version_id} for conversation {conversation_id}")
+            return True
+        return False
+
     async def activate_memory_version(self, conversation_id: str, version_id: str) -> bool:
         """Activate a specific memory version in conversation."""
         # First verify the version exists

backends/advanced/src/advanced_omi_backend/main.py

@@ -48,6 +48,9 @@
     get_processor_manager,
     init_processor_manager,
 )
+
+from advanced_omi_backend.simple_queue import get_simple_queue
+
 from advanced_omi_backend.audio_utils import process_audio_chunk
 from advanced_omi_backend.task_manager import init_task_manager, get_task_manager
 from advanced_omi_backend.transcript_coordinator import get_transcript_coordinator
@@ -320,6 +323,25 @@ async def lifespan(app: FastAPI):
     processor_manager = init_processor_manager(CHUNK_DIR, ac_repository)
     await processor_manager.start()
 
+    application_logger.info("Application-level processors started")
+
+    # Initialize simple queue system
+    try:
+        queue = await get_simple_queue()
+        application_logger.info("Simple queue system started")
+    except Exception as e:
+        application_logger.error(f"Failed to start simple queue: {e}")
+        # Don't raise as queue system is not critical for basic operation
+
+    # Skip memory service pre-initialization to avoid blocking FastAPI startup
+    # Memory service will be lazily initialized when first used
+    application_logger.info("Memory service will be initialized on first use (lazy loading)")
+
+    # SystemTracker is used for monitoring and debugging
+    application_logger.info("Using SystemTracker for monitoring and debugging")
+
+    application_logger.info("Application ready - using application-level processing architecture.")
+
     logger.info("App ready")
     try:
         yield
@@ -331,6 +353,14 @@ async def lifespan(app: FastAPI):
         for client_id in client_manager.get_all_client_ids():
             await cleanup_client_state(client_id)
 
+        # Shutdown simple queue system
+        try:
+            if queue:
+                await queue.stop_worker()
+                application_logger.info("Simple queue system shut down")
+        except Exception as e:
+            application_logger.error(f"Error shutting down simple queue: {e}")
+
         # Shutdown processor manager
         processor_manager = get_processor_manager()
         await processor_manager.shutdown()

backends/advanced/src/advanced_omi_backend/routers/api_router.py

@@ -14,6 +14,7 @@
     client_router,
     conversation_router,
     memory_router,
+    queue_router,
     system_router,
     user_router,
 )
@@ -31,6 +32,7 @@
 router.include_router(conversation_router)
 router.include_router(memory_router)
 router.include_router(system_router)
+router.include_router(queue_router)
 
 
 logger.info("API router initialized with all sub-modules")

backends/advanced/src/advanced_omi_backend/routers/modules/__init__.py

@@ -8,6 +8,7 @@
 - conversation_routes: Conversation CRUD and audio processing
 - memory_routes: Memory management, search, and debug
 - system_routes: System utilities, metrics, and file processing
+- queue_routes: Job queue management and monitoring
 """
 
 from .chat_routes import router as chat_router
@@ -16,5 +17,6 @@
 from .memory_routes import router as memory_router
 from .system_routes import router as system_router
 from .user_routes import router as user_router
+from .queue_routes import router as queue_router
 
-__all__ = ["user_router", "chat_router", "client_router", "conversation_router", "memory_router", "system_router"]
+__all__ = ["user_router", "chat_router", "client_router", "conversation_router", "memory_router", "system_router", "queue_router"]

backends/advanced/src/advanced_omi_backend/routers/modules/queue_routes.py

@@ -0,0 +1,147 @@
+"""
+Simple queue API routes for job monitoring.
+Provides basic endpoints for viewing job status and statistics.
+"""
+
+import logging
+from fastapi import APIRouter, Depends, Query, HTTPException
+from pydantic import BaseModel
+from typing import List, Optional
+
+from advanced_omi_backend.auth import current_active_user
+from advanced_omi_backend.simple_queue import get_simple_queue
+from advanced_omi_backend.users import User
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/queue", tags=["queue"])
+
+
+@router.get("/jobs")
+async def list_jobs(
+    limit: int = Query(20, ge=1, le=100, description="Number of jobs to return"),
+    offset: int = Query(0, ge=0, description="Number of jobs to skip"),
+    status: str = Query(None, description="Filter by job status"),
+    job_type: str = Query(None, description="Filter by job type"),
+    priority: str = Query(None, description="Filter by job priority"),
+    current_user: User = Depends(current_active_user)
+):
+    """List jobs with pagination and filtering."""
+    try:
+        # Build filters dict
+        filters = {}
+        if status:
+            filters["status"] = status
+        if job_type:
+            filters["job_type"] = job_type
+        if priority:
+            filters["priority"] = priority
+
+        queue = await get_simple_queue()
+        result = await queue.get_jobs(limit=limit, offset=offset, filters=filters)
+
+            # Filter jobs by user if not admin
+        if not current_user.is_superuser:
+            result["jobs"] = [
+                job for job in result["jobs"]
+                if job["user_id"] == str(current_user.user_id)
+            ]
+            result["pagination"]["total"] = len(result["jobs"])
+
+        return result
+
+    except Exception as e:
+        logger.error(f"Failed to list jobs: {e}")
+        return {"error": "Failed to list jobs", "jobs": [], "pagination": {"total": 0, "limit": limit, "offset": offset, "has_more": False}}
+
+
+@router.get("/stats")
+async def get_queue_stats(
+    current_user: User = Depends(current_active_user)
+):
+    """Get queue statistics."""
+    try:
+        queue = await get_simple_queue()
+        stats = await queue.get_job_stats()
+        return stats
+
+    except Exception as e:
+        logger.error(f"Failed to get queue stats: {e}")
+        return {"queued": 0, "processing": 0, "completed": 0, "failed": 0}
+
+
+@router.get("/health")
+async def get_queue_health():
+    """Get queue system health status."""
+    try:
+        queue = await get_simple_queue()
+
+        return {
+            "status": "healthy" if queue.running else "stopped",
+            "worker_running": queue.running,
+            "message": "Simple queue is operational" if queue.running else "Simple queue worker not running"
+        }
+
+    except Exception as e:
+        logger.error(f"Failed to get queue health: {e}")
+        return {
+            "status": "unhealthy",
+            "message": f"Health check failed: {str(e)}"
+        }
+
+
+class FlushJobsRequest(BaseModel):
+    older_than_hours: int = 24
+    statuses: Optional[List[str]] = None
+
+
+class FlushAllJobsRequest(BaseModel):
+    confirm: bool = False
+
+
+@router.post("/flush")
+async def flush_inactive_jobs(
+    request: FlushJobsRequest,
+    current_user: User = Depends(current_active_user)
+):
+    """Flush inactive jobs from the database (admin only)."""
+    if not current_user.is_superuser:
+        raise HTTPException(status_code=403, detail="Admin access required")
+
+    try:
+        queue = await get_simple_queue()
+        result = await queue.flush_inactive_jobs(
+            older_than_hours=request.older_than_hours,
+            statuses=request.statuses
+        )
+        return result
+
+    except Exception as e:
+        logger.error(f"Failed to flush inactive jobs: {e}")
+        raise HTTPException(status_code=500, detail=f"Failed to flush jobs: {str(e)}")
+
+
+@router.post("/flush-all")
+async def flush_all_jobs(
+    request: FlushAllJobsRequest,
+    current_user: User = Depends(current_active_user)
+):
+    """Flush ALL jobs from the database (admin only). USE WITH EXTREME CAUTION!"""
+    if not current_user.is_superuser:
+        raise HTTPException(status_code=403, detail="Admin access required")
+
+    try:
+        if not request.confirm:
+            raise HTTPException(
+                status_code=400,
+                detail="Must set confirm=true to flush all jobs. This is a destructive operation."
+            )
+
+        queue = await get_simple_queue()
+        result = await queue.flush_all_jobs(confirm=request.confirm)
+        return result
+
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    except Exception as e:
+        logger.error(f"Failed to flush all jobs: {e}")
+        raise HTTPException(status_code=500, detail=f"Failed to flush all jobs: {str(e)}")