Refactor audio processing and conversation management for improved deduplication and tracking

Author: AnkushMalaker

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

@@ -85,14 +85,14 @@ async def upload_and_process_audio_files(
                 content = await file.read()
 
 
-                # Generate audio UUID and timestamp
+                # Track external source for deduplication (Google Drive, etc.)
+                external_source_id = None
+                external_source_type = None
                 if source == "gdrive":
-                    audio_uuid = getattr(file, "audio_uuid", None)
-                    if not audio_uuid:
-                        audio_logger.error(f"Missing audio_uuid for gdrive file: {file.filename}")
-                        audio_uuid = str(uuid.uuid4())
-                else:
-                    audio_uuid = str(uuid.uuid4())
+                    external_source_id = getattr(file, "file_id", None) or getattr(file, "audio_uuid", None)
+                    external_source_type = "gdrive"
+                    if not external_source_id:
+                        audio_logger.warning(f"Missing file_id for gdrive file: {file.filename}")
                 timestamp = int(time.time() * 1000)
 
                 # Validate and prepare audio (read format from WAV file)
@@ -121,11 +121,12 @@ async def upload_and_process_audio_files(
                 title = file.filename.rsplit('.', 1)[0][:50] if file.filename else "Uploaded Audio"
 
                 conversation = create_conversation(
-                    audio_uuid=audio_uuid,
                     user_id=user.user_id,
                     client_id=client_id,
                     title=title,
-                    summary="Processing uploaded audio file..."
+                    summary="Processing uploaded audio file...",
+                    external_source_id=external_source_id,
+                    external_source_type=external_source_type,
                 )
                 await conversation.insert()
                 conversation_id = conversation.conversation_id  # Get the auto-generated ID
@@ -184,22 +185,20 @@ async def upload_and_process_audio_files(
                 transcription_job = transcription_queue.enqueue(
                     transcribe_full_audio_job,
                     conversation_id,
-                    audio_uuid,
                     version_id,
                     "batch",  # trigger
                     job_timeout=1800,  # 30 minutes
                     result_ttl=JOB_RESULT_TTL,
                     job_id=transcribe_job_id,
                     description=f"Transcribe uploaded file {conversation_id[:8]}",
-                    meta={'audio_uuid': audio_uuid, 'conversation_id': conversation_id, 'client_id': client_id}
+                    meta={'conversation_id': conversation_id, 'client_id': client_id}
                 )
 
                 audio_logger.info(f"📥 Enqueued transcription job {transcription_job.id} for uploaded file")
 
                 # Enqueue post-conversation processing job chain (depends on transcription)
                 job_ids = start_post_conversation_jobs(
                     conversation_id=conversation_id,
-                    audio_uuid=audio_uuid,
                     user_id=user.user_id,
                     transcript_version_id=version_id,  # Pass the version_id from transcription job
                     depends_on_job=transcription_job,  # Wait for transcription to complete
@@ -209,7 +208,6 @@ async def upload_and_process_audio_files(
                 processed_files.append({
                     "filename": file.filename,
                     "status": "processing",
-                    "audio_uuid": audio_uuid,
                     "conversation_id": conversation_id,
                     "transcript_job_id": transcription_job.id,
                     "speaker_job_id": job_ids['speaker_recognition'],

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

@@ -109,10 +109,8 @@ async def get_conversation(conversation_id: str, user: User):
         # Build response with explicit curated fields
         response = {
             "conversation_id": conversation.conversation_id,
-            "audio_uuid": conversation.audio_uuid,
             "user_id": conversation.user_id,
             "client_id": conversation.client_id,
-            "audio_path": conversation.audio_path,
             "audio_chunks_count": conversation.audio_chunks_count,
             "audio_total_duration": conversation.audio_total_duration,
             "audio_compression_ratio": conversation.audio_compression_ratio,
@@ -175,10 +173,8 @@ async def get_conversations(user: User, include_deleted: bool = False):
         for conv in user_conversations:
             conversations.append({
                 "conversation_id": conv.conversation_id,
-                "audio_uuid": conv.audio_uuid,
                 "user_id": conv.user_id,
                 "client_id": conv.client_id,
-                "audio_path": conv.audio_path,
                 "audio_chunks_count": conv.audio_chunks_count,
                 "audio_total_duration": conv.audio_total_duration,
                 "audio_compression_ratio": conv.audio_compression_ratio,
@@ -248,7 +244,6 @@ async def _hard_delete_conversation(conversation: Conversation) -> JSONResponse:
     """Permanently delete conversation and chunks (admin only)."""
     conversation_id = conversation.conversation_id
     client_id = conversation.client_id
-    audio_uuid = conversation.audio_uuid
 
     # Delete conversation document
     await conversation.delete()
@@ -268,8 +263,7 @@ async def _hard_delete_conversation(conversation: Conversation) -> JSONResponse:
             "message": f"Successfully permanently deleted conversation '{conversation_id}'",
             "deleted_chunks": deleted_chunks,
             "conversation_id": conversation_id,
-            "client_id": client_id,
-            "audio_uuid": audio_uuid
+            "client_id": client_id
         }
     )
 
@@ -411,8 +405,6 @@ async def reprocess_transcript(conversation_id: str, user: User):
             return JSONResponse(status_code=403, content={"error": "Access forbidden. You can only reprocess your own conversations."})
 
         # Get audio_uuid from conversation
-        audio_uuid = conversation_model.audio_uuid
-
         # Validate audio chunks exist in MongoDB
         chunks = await AudioChunkDocument.find(
             AudioChunkDocument.conversation_id == conversation_id
@@ -439,14 +431,13 @@ async def reprocess_transcript(conversation_id: str, user: User):
         transcript_job = transcription_queue.enqueue(
             transcribe_full_audio_job,
             conversation_id,
-            audio_uuid,
             version_id,
             "reprocess",
             job_timeout=600,
             result_ttl=JOB_RESULT_TTL,
             job_id=f"reprocess_{conversation_id[:8]}",
             description=f"Transcribe audio for {conversation_id[:8]}",
-            meta={'audio_uuid': audio_uuid, 'conversation_id': conversation_id}
+            meta={'conversation_id': conversation_id}
         )
         logger.info(f"📥 RQ: Enqueued transcription job {transcript_job.id}")
 
@@ -468,7 +459,7 @@ async def reprocess_transcript(conversation_id: str, user: User):
                 result_ttl=JOB_RESULT_TTL,
                 job_id=f"speaker_{conversation_id[:8]}",
                 description=f"Recognize speakers for {conversation_id[:8]}",
-                meta={'audio_uuid': audio_uuid, 'conversation_id': conversation_id}
+                meta={'conversation_id': conversation_id}
             )
             speaker_dependency = speaker_job  # Chain for next job
             logger.info(f"📥 RQ: Enqueued speaker recognition job {speaker_job.id} (depends on {transcript_job.id})")
@@ -486,7 +477,7 @@ async def reprocess_transcript(conversation_id: str, user: User):
             result_ttl=JOB_RESULT_TTL,
             job_id=f"memory_{conversation_id[:8]}",
             description=f"Extract memories for {conversation_id[:8]}",
-            meta={'audio_uuid': audio_uuid, 'conversation_id': conversation_id}
+            meta={'conversation_id': conversation_id}
         )
         if speaker_job:
             logger.info(f"📥 RQ: Enqueued memory job {memory_job.id} (depends on speaker job {speaker_job.id})")

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

@@ -211,15 +211,15 @@ def get_jobs(
     }
 
 
-def all_jobs_complete_for_session(session_id: str) -> bool:
+def all_jobs_complete_for_client(client_id: str) -> bool:
     """
-    Check if all jobs associated with a session are in terminal states.
+    Check if all jobs associated with a client are in terminal states.
 
-    Only checks jobs with audio_uuid in job.meta (no backward compatibility).
+    Checks jobs with client_id in job.meta.
     Traverses dependency chains to include dependent jobs.
 
     Args:
-        session_id: The audio_uuid (session ID) to check jobs for
+        client_id: The client device identifier to check jobs for
 
     Returns:
         True if all jobs are complete (or no jobs found), False if any job is still processing
@@ -248,7 +248,7 @@ def is_job_complete(job):
 
         return True
 
-    # Find all jobs for this session
+    # Find all jobs for this client
     all_queues = [transcription_queue, memory_queue, audio_queue, default_queue]
     for queue in all_queues:
         registries = [
@@ -266,8 +266,8 @@ def is_job_complete(job):
                 try:
                     job = Job.fetch(job_id, connection=redis_conn)
 
-                    # Only check jobs with audio_uuid in meta
-                    if job.meta and job.meta.get('audio_uuid') == session_id:
+                    # Only check jobs with client_id in meta
+                    if job.meta and job.meta.get('client_id') == client_id:
                         if not is_job_complete(job):
                             return False
                 except Exception as e:
@@ -289,7 +289,7 @@ def start_streaming_jobs(
     2. Audio persistence job - writes audio chunks to WAV file (file rotation per conversation)
 
     Args:
-        session_id: Stream session ID (audio_uuid)
+        session_id: Stream session ID (equals client_id for streaming)
         user_id: User identifier
         client_id: Client identifier
 
@@ -313,7 +313,7 @@ def start_streaming_jobs(
         failure_ttl=86400,  # Cleanup failed jobs after 24h
         job_id=f"speech-detect_{session_id[:12]}",
         description=f"Listening for speech...",
-        meta={'audio_uuid': session_id, 'client_id': client_id, 'session_level': True}
+        meta={'client_id': client_id, 'session_level': True}
     )
     # Log job enqueue with TTL information for debugging
     actual_ttl = redis_conn.ttl(f"rq:job:{speech_job.id}")
@@ -346,7 +346,7 @@ def start_streaming_jobs(
         failure_ttl=86400,  # Cleanup failed jobs after 24h
         job_id=f"audio-persist_{session_id[:12]}",
         description=f"Audio persistence for session {session_id[:12]}",
-        meta={'audio_uuid': session_id, 'session_level': True}  # Mark as session-level job
+        meta={'client_id': client_id, 'session_level': True}  # Mark as session-level job
     )
     # Log job enqueue with TTL information for debugging
     actual_ttl = redis_conn.ttl(f"rq:job:{audio_job.id}")
@@ -366,7 +366,6 @@ def start_streaming_jobs(
 
 def start_post_conversation_jobs(
     conversation_id: str,
-    audio_uuid: str,
     user_id: str,
     transcript_version_id: Optional[str] = None,
     depends_on_job = None,
@@ -386,7 +385,6 @@ def start_post_conversation_jobs(
 
     Args:
         conversation_id: Conversation identifier
-        audio_uuid: Audio UUID for job tracking
         user_id: User identifier
         transcript_version_id: Transcript version ID (auto-generated if None)
         depends_on_job: Optional job dependency for first job (e.g., transcription for file uploads)
@@ -402,7 +400,7 @@ def start_post_conversation_jobs(
     version_id = transcript_version_id or str(uuid.uuid4())
 
     # Build job metadata (include client_id if provided for UI tracking)
-    job_meta = {'audio_uuid': audio_uuid, 'conversation_id': conversation_id}
+    job_meta = {'conversation_id': conversation_id}
     if client_id:
         job_meta['client_id'] = client_id
 
@@ -416,7 +414,7 @@ def start_post_conversation_jobs(
 
     if speaker_enabled:
         speaker_job_id = f"speaker_{conversation_id[:12]}"
-        logger.info(f"🔍 DEBUG: Creating speaker job with job_id={speaker_job_id}, conversation_id={conversation_id[:12]}, audio_uuid={audio_uuid[:12]}")
+        logger.info(f"🔍 DEBUG: Creating speaker job with job_id={speaker_job_id}, conversation_id={conversation_id[:12]}")
 
         speaker_job = transcription_queue.enqueue(
             recognise_speakers_job,
@@ -440,7 +438,7 @@ def start_post_conversation_jobs(
     # Step 2: Memory extraction job
     # Depends on speaker job if it was created, otherwise depends on upstream (transcription or nothing)
     memory_job_id = f"memory_{conversation_id[:12]}"
-    logger.info(f"🔍 DEBUG: Creating memory job with job_id={memory_job_id}, conversation_id={conversation_id[:12]}, audio_uuid={audio_uuid[:12]}")
+    logger.info(f"🔍 DEBUG: Creating memory job with job_id={memory_job_id}, conversation_id={conversation_id[:12]}")
 
     memory_job = memory_queue.enqueue(
         process_memory_job,
@@ -462,7 +460,7 @@ def start_post_conversation_jobs(
     # Step 3: Title/summary generation job
     # Depends on speaker job if enabled, otherwise on upstream dependency
     title_job_id = f"title_summary_{conversation_id[:12]}"
-    logger.info(f"🔍 DEBUG: Creating title/summary job with job_id={title_job_id}, conversation_id={conversation_id[:12]}, audio_uuid={audio_uuid[:12]}")
+    logger.info(f"🔍 DEBUG: Creating title/summary job with job_id={title_job_id}, conversation_id={conversation_id[:12]}")
 
     title_summary_job = default_queue.enqueue(
         generate_title_summary_job,
@@ -484,14 +482,13 @@ def start_post_conversation_jobs(
     # Step 5: Dispatch conversation.complete event (runs after both memory and title/summary complete)
     # This ensures plugins receive the event after all processing is done
     event_job_id = f"event_complete_{conversation_id[:12]}"
-    logger.info(f"🔍 DEBUG: Creating conversation complete event job with job_id={event_job_id}, conversation_id={conversation_id[:12]}, audio_uuid={audio_uuid[:12]}")
+    logger.info(f"🔍 DEBUG: Creating conversation complete event job with job_id={event_job_id}, conversation_id={conversation_id[:12]}")
 
     # Event job depends on both memory and title/summary jobs completing
     # Use RQ's depends_on list to wait for both
     event_dispatch_job = default_queue.enqueue(
         dispatch_conversation_complete_event_job,
         conversation_id,
-        audio_uuid,
         client_id or "",
         user_id,
         job_timeout=120,  # 2 minutes

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

@@ -838,8 +838,7 @@ async def _process_batch_audio_complete(
             f"📦 Batch mode: Combined {len(client_state.batch_audio_chunks)} chunks into {len(complete_audio)} bytes"
         )
 
-        # Generate audio UUID and timestamp
-        audio_uuid = str(uuid.uuid4())
+        # Timestamp for logging
         timestamp = int(time.time() * 1000)
 
         # Get audio format from batch metadata (set during audio-start)
@@ -859,7 +858,6 @@ async def _process_batch_audio_complete(
         version_id = str(uuid.uuid4())
 
         conversation = create_conversation(
-            audio_uuid=audio_uuid,
             user_id=user_id,
             client_id=client_id,
             title="Batch Recording",
@@ -904,22 +902,20 @@ async def _process_batch_audio_complete(
         transcription_job = transcription_queue.enqueue(
             transcribe_full_audio_job,
             conversation_id,
-            audio_uuid,
             version_id,
             "batch",  # trigger
             job_timeout=1800,  # 30 minutes
             result_ttl=JOB_RESULT_TTL,
             job_id=transcribe_job_id,
             description=f"Transcribe batch audio {conversation_id[:8]}",
-            meta={'audio_uuid': audio_uuid, 'conversation_id': conversation_id, 'client_id': client_id}
+            meta={'conversation_id': conversation_id, 'client_id': client_id}
         )
 
         application_logger.info(f"📥 Batch mode: Enqueued transcription job {transcription_job.id}")
 
         # Enqueue post-conversation processing job chain (depends on transcription)
         job_ids = start_post_conversation_jobs(
             conversation_id=conversation_id,
-            audio_uuid=audio_uuid,
             user_id=None,  # Will be read from conversation in DB by jobs
             depends_on_job=transcription_job,  # Wait for transcription to complete
             client_id=client_id  # Pass client_id for UI tracking