Refactor job status handling to align with RQ standards

- Updated job status checks across various modules to use "started" and "finished" instead of "processing" and "completed" for consistency with RQ's naming conventions.
- Adjusted related logging and response messages to reflect the new status terminology.
- Simplified Docker Compose project name handling in test scripts to avoid conflicts and improve clarity in test environment setup.

Author: AnkushMalaker

backends/advanced/docker-compose-test.yml

@@ -2,6 +2,8 @@
 # Isolated test environment for integration tests
 # Uses different ports to avoid conflicts with development environment
 
+name: backend-test
+
 services:
   chronicle-backend-test:
     build:

backends/advanced/run-test.sh

@@ -219,17 +219,13 @@ if [ -d "./data/test_audio_chunks/" ] || [ -d "./data/test_data/" ] || [ -d "./d
     docker run --rm -v "$(pwd)/data:/data" alpine sh -c 'rm -rf /data/test_*' 2>/dev/null || true
 fi
 
-# Use unique project name to avoid conflicts with development environment
-export COMPOSE_PROJECT_NAME="advanced-backend-test"
+# Note: Project name 'backend-test' is set in docker-compose-test.yml
+# No need to export COMPOSE_PROJECT_NAME - it's handled by the compose file
 
 # Stop any existing test containers
 print_info "Stopping existing test containers..."
-# Try cleanup with current project name
 docker compose -f docker-compose-test.yml down -v || true
 
-# Also try cleanup with default project name (in case containers were started without COMPOSE_PROJECT_NAME)
-COMPOSE_PROJECT_NAME=advanced docker compose -f docker-compose-test.yml down -v 2>/dev/null || true
-
 # Run integration tests
 print_info "Running integration tests..."
 print_info "Using fresh mode (CACHED_MODE=False) for clean testing"
@@ -268,8 +264,6 @@ else
     if [ "${CLEANUP_CONTAINERS:-true}" != "false" ]; then
         print_info "Cleaning up test containers after failure..."
         docker compose -f docker-compose-test.yml down -v || true
-        # Also cleanup with default project name
-        COMPOSE_PROJECT_NAME=advanced docker compose -f docker-compose-test.yml down -v 2>/dev/null || true
         docker system prune -f || true
     else
         print_warning "Skipping cleanup (CLEANUP_CONTAINERS=false) - containers left running for debugging"
@@ -282,8 +276,6 @@ fi
 if [ "${CLEANUP_CONTAINERS:-true}" != "false" ]; then
     print_info "Cleaning up test containers..."
     docker compose -f docker-compose-test.yml down -v || true
-    # Also cleanup with default project name
-    COMPOSE_PROJECT_NAME=advanced docker compose -f docker-compose-test.yml down -v 2>/dev/null || true
     docker system prune -f || true
 else
     print_warning "Skipping cleanup (CLEANUP_CONTAINERS=false) - containers left running"

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

@@ -203,7 +203,7 @@ async def upload_and_process_audio_files(
 
                 processed_files.append({
                     "filename": file.filename,
-                    "status": "processing",
+                    "status": "started",  # RQ standard: job has been enqueued
                     "conversation_id": conversation_id,
                     "transcript_job_id": transcription_job.id,
                     "speaker_job_id": job_ids['speaker_recognition'],
@@ -233,7 +233,7 @@ async def upload_and_process_audio_files(
                     "error": str(e),
                 })
 
-        successful_files = [f for f in processed_files if f.get("status") == "processing"]
+        successful_files = [f for f in processed_files if f.get("status") == "started"]
         failed_files = [f for f in processed_files if f.get("status") == "error"]
 
         return {
@@ -242,7 +242,7 @@ async def upload_and_process_audio_files(
             "files": processed_files,
             "summary": {
                 "total": len(files),
-                "processing": len(successful_files),
+                "started": len(successful_files),  # RQ standard
                 "failed": len(failed_files),
             },
         }

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

@@ -17,7 +17,7 @@
 
 import redis
 from rq import Queue, Worker
-from rq.job import Job
+from rq.job import Job, JobStatus
 from rq.registry import ScheduledJobRegistry, DeferredJobRegistry
 
 from advanced_omi_backend.models.job import JobPriority
@@ -30,6 +30,52 @@
 REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
 redis_conn = redis.from_url(REDIS_URL)
 
+
+def get_job_status_from_rq(job: Job) -> str:
+    """
+    Get job status using RQ's native method.
+
+    Uses job.get_status() which is the Redis Queue standard approach.
+    Returns RQ's standard status names.
+
+    Returns one of: queued, started, finished, failed, deferred, scheduled, canceled, stopped
+
+    Raises:
+        RuntimeError: If job status is unexpected (should never happen with RQ's method)
+    """
+    rq_status = job.get_status()
+
+    # RQ returns status as JobStatus enum or string
+    # Convert to string if it's an enum
+    if isinstance(rq_status, JobStatus):
+        status_str = rq_status.value
+    else:
+        status_str = str(rq_status)
+
+    # Validate it's a known RQ status
+    valid_statuses = {
+        JobStatus.QUEUED.value,
+        JobStatus.STARTED.value,
+        JobStatus.FINISHED.value,
+        JobStatus.FAILED.value,
+        JobStatus.DEFERRED.value,
+        JobStatus.SCHEDULED.value,
+        JobStatus.CANCELED.value,
+        JobStatus.STOPPED.value,
+    }
+
+    if status_str not in valid_statuses:
+        logger.error(
+            f"Job {job.id} has unexpected RQ status: {status_str}. "
+            f"This indicates RQ library added a new status we don't know about."
+        )
+        raise RuntimeError(
+            f"Job {job.id} has unknown RQ status: {status_str}. "
+            f"Please update get_job_status_from_rq() to handle this new status."
+        )
+
+    return status_str
+
 # Queue name constants
 TRANSCRIPTION_QUEUE = "transcription"
 MEMORY_QUEUE = "memory"
@@ -61,34 +107,34 @@ def get_queue(queue_name: str = DEFAULT_QUEUE) -> Queue:
 
 
 def get_job_stats() -> Dict[str, Any]:
-    """Get statistics about jobs in all queues matching frontend expectations."""
+    """Get statistics about jobs in all queues using RQ standard status names."""
     total_jobs = 0
     queued_jobs = 0
-    processing_jobs = 0
-    completed_jobs = 0
+    started_jobs = 0  # RQ standard: "started" not "processing"
+    finished_jobs = 0  # RQ standard: "finished" not "completed"
     failed_jobs = 0
-    cancelled_jobs = 0
+    canceled_jobs = 0  # RQ standard: "canceled" not "cancelled"
     deferred_jobs = 0  # Jobs waiting for dependencies (depends_on)
 
     for queue_name in QUEUE_NAMES:
         queue = get_queue(queue_name)
 
         queued_jobs += len(queue)
-        processing_jobs += len(queue.started_job_registry)
-        completed_jobs += len(queue.finished_job_registry)
+        started_jobs += len(queue.started_job_registry)
+        finished_jobs += len(queue.finished_job_registry)
         failed_jobs += len(queue.failed_job_registry)
-        cancelled_jobs += len(queue.canceled_job_registry)
+        canceled_jobs += len(queue.canceled_job_registry)
         deferred_jobs += len(queue.deferred_job_registry)
 
-    total_jobs = queued_jobs + processing_jobs + completed_jobs + failed_jobs + cancelled_jobs + deferred_jobs
+    total_jobs = queued_jobs + started_jobs + finished_jobs + failed_jobs + canceled_jobs + deferred_jobs
 
     return {
         "total_jobs": total_jobs,
         "queued_jobs": queued_jobs,
-        "processing_jobs": processing_jobs,
-        "completed_jobs": completed_jobs,
+        "started_jobs": started_jobs,
+        "finished_jobs": finished_jobs,
         "failed_jobs": failed_jobs,
-        "cancelled_jobs": cancelled_jobs,
+        "canceled_jobs": canceled_jobs,
         "deferred_jobs": deferred_jobs,
         "timestamp": datetime.utcnow().isoformat()
     }
@@ -124,11 +170,11 @@ def get_jobs(
     for qname in queues_to_check:
         queue = get_queue(qname)
 
-        # Collect jobs from all registries
+        # Collect jobs from all registries (using RQ standard status names)
         registries = [
             (queue.job_ids, "queued"),
-            (queue.started_job_registry.get_job_ids(), "processing"),
-            (queue.finished_job_registry.get_job_ids(), "completed"),
+            (queue.started_job_registry.get_job_ids(), "started"),  # RQ standard, not "processing"
+            (queue.finished_job_registry.get_job_ids(), "finished"),  # RQ standard, not "completed"
             (queue.failed_job_registry.get_job_ids(), "failed"),
             (queue.deferred_job_registry.get_job_ids(), "deferred"),  # Jobs waiting for dependencies
         ]

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

@@ -57,11 +57,11 @@ async def mark_session_complete(
     """
     session_key = f"audio:session:{session_id}"
     await redis_client.hset(session_key, mapping={
-        "status": "complete",
+        "status": "finished",
         "completed_at": str(time.time()),
         "completion_reason": reason
     })
-    logger.info(f"✅ Session {session_id[:12]} marked complete: {reason}")
+    logger.info(f"✅ Session {session_id[:12]} marked finished: {reason}")
 
 
 async def get_session_info(redis_client, session_id: str) -> Optional[Dict]:
@@ -231,15 +231,15 @@ async def get_streaming_status(request):
             # Check if all jobs are complete (including failed jobs)
             all_jobs_done = all_jobs_complete_for_session(session_id)
 
-            # Session is completed if:
-            # 1. Redis status says complete/finalized AND all jobs done, OR
-            # 2. All jobs are done (even if status isn't complete yet)
-            # This ensures sessions with failed jobs move to completed
-            if status in ["complete", "completed", "finalized"] or all_jobs_done:
+            # Session is finished if:
+            # 1. Redis status says finished AND all jobs done, OR
+            # 2. All jobs are done (even if status isn't finished yet)
+            # This ensures sessions with failed jobs move to finished
+            if status == "finished" or all_jobs_done:
                 if all_jobs_done:
-                    # All jobs complete - this is truly a completed session
-                    # Update Redis status if it wasn't already marked complete
-                    if status not in ["complete", "completed", "finalized"]:
+                    # All jobs finished - this is truly a finished session
+                    # Update Redis status if it wasn't already marked finished
+                    if status != "finished":
                         await mark_session_complete(redis_client, session_id, "all_jobs_complete")
 
                     # Get additional session data for completed sessions
@@ -251,7 +251,7 @@ async def get_streaming_status(request):
                         "client_id": session_obj.get("client_id", ""),
                         "conversation_id": session_data.get(b"conversation_id", b"").decode() if session_data and b"conversation_id" in session_data else None,
                         "has_conversation": bool(session_data and session_data.get(b"conversation_id", b"")),
-                        "action": session_data.get(b"action", b"complete").decode() if session_data and b"action" in session_data else "complete",
+                        "action": session_data.get(b"action", b"finished").decode() if session_data and b"action" in session_data else "finished",
                         "reason": session_data.get(b"reason", b"").decode() if session_data and b"reason" in session_data else "",
                         "completed_at": session_obj.get("last_chunk_at", 0),
                         "audio_file": session_data.get(b"audio_file", b"").decode() if session_data and b"audio_file" in session_data else "",
@@ -450,26 +450,26 @@ async def get_streaming_status(request):
         rq_stats = {
             "transcription_queue": {
                 "queued": transcription_queue.count,
-                "processing": len(transcription_queue.started_job_registry),
-                "completed": len(transcription_queue.finished_job_registry),
+                "started": len(transcription_queue.started_job_registry),
+                "finished": len(transcription_queue.finished_job_registry),
                 "failed": len(transcription_queue.failed_job_registry),
-                "cancelled": len(transcription_queue.canceled_job_registry),
+                "canceled": len(transcription_queue.canceled_job_registry),
                 "deferred": len(transcription_queue.deferred_job_registry)
             },
             "memory_queue": {
                 "queued": memory_queue.count,
-                "processing": len(memory_queue.started_job_registry),
-                "completed": len(memory_queue.finished_job_registry),
+                "started": len(memory_queue.started_job_registry),
+                "finished": len(memory_queue.finished_job_registry),
                 "failed": len(memory_queue.failed_job_registry),
-                "cancelled": len(memory_queue.canceled_job_registry),
+                "canceled": len(memory_queue.canceled_job_registry),
                 "deferred": len(memory_queue.deferred_job_registry)
             },
             "default_queue": {
                 "queued": default_queue.count,
-                "processing": len(default_queue.started_job_registry),
-                "completed": len(default_queue.finished_job_registry),
+                "started": len(default_queue.started_job_registry),
+                "finished": len(default_queue.finished_job_registry),
                 "failed": len(default_queue.failed_job_registry),
-                "cancelled": len(default_queue.canceled_job_registry),
+                "canceled": len(default_queue.canceled_job_registry),
                 "deferred": len(default_queue.deferred_job_registry)
             }
         }

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

@@ -176,14 +176,12 @@ async def get_status(job_id: str, current_user: User = Depends(current_active_us
         status = job.get_status()
         if status == "started":
             status = "running"
-        if status == "canceled":
-            status = "cancelled"
-            
+
         # Get metadata
         meta = job.meta or {}
-        
+
         # If meta has status, prefer it (for granular updates)
-        if "status" in meta and meta["status"] in ("running", "completed", "failed", "cancelled"):
+        if "status" in meta and meta["status"] in ("running", "finished", "failed", "canceled"):
              status = meta["status"]
 
         total = meta.get("total_files", 0)