Implement miscellaneous settings management and enhance audio processing

- Introduced functions to retrieve and save miscellaneous settings, including `always_persist_enabled` and `use_provider_segments`, using OmegaConf.
- Updated the system controller and routes to handle new endpoints for managing miscellaneous settings, ensuring admin access control.
- Refactored audio processing jobs to read the `always_persist_enabled` setting from global configuration, improving audio persistence behavior.
- Enhanced the web UI to allow administrators to view and modify miscellaneous settings, providing better control over audio processing features.
- Added integration tests to verify the functionality of the new settings management, ensuring robust handling of audio persistence scenarios.

Author: AnkushMalaker

backends/advanced/src/advanced_omi_backend/config.py

@@ -176,4 +176,56 @@ def get_audio_storage_settings() -> dict:
         Dict with audio_base_path, audio_chunks_path
     """
     cfg = get_backend_config('audio_storage')
-    return OmegaConf.to_container(cfg, resolve=True)
+    return OmegaConf.to_container(cfg, resolve=True)
+
+
+# ============================================================================
+# Miscellaneous Settings (OmegaConf-based)
+# ============================================================================
+
+def get_misc_settings() -> dict:
+    """
+    Get miscellaneous configuration settings using OmegaConf.
+
+    Returns:
+        Dict with always_persist_enabled and use_provider_segments
+    """
+    # Get audio settings for always_persist_enabled
+    audio_cfg = get_backend_config('audio')
+    audio_settings = OmegaConf.to_container(audio_cfg, resolve=True) if audio_cfg else {}
+
+    # Get transcription settings for use_provider_segments
+    transcription_cfg = get_backend_config('transcription')
+    transcription_settings = OmegaConf.to_container(transcription_cfg, resolve=True) if transcription_cfg else {}
+
+    return {
+        'always_persist_enabled': audio_settings.get('always_persist_enabled', False),
+        'use_provider_segments': transcription_settings.get('use_provider_segments', False)
+    }
+
+
+def save_misc_settings(settings: dict) -> bool:
+    """
+    Save miscellaneous settings to config.yml using OmegaConf.
+
+    Args:
+        settings: Dict with always_persist_enabled and/or use_provider_segments
+
+    Returns:
+        True if saved successfully, False otherwise
+    """
+    success = True
+
+    # Save audio settings if always_persist_enabled is provided
+    if 'always_persist_enabled' in settings:
+        audio_settings = {'always_persist_enabled': settings['always_persist_enabled']}
+        if not save_config_section('backend.audio', audio_settings):
+            success = False
+
+    # Save transcription settings if use_provider_segments is provided
+    if 'use_provider_segments' in settings:
+        transcription_settings = {'use_provider_segments': settings['use_provider_segments']}
+        if not save_config_section('backend.transcription', transcription_settings):
+            success = False
+
+    return success

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

@@ -325,8 +325,7 @@ def is_job_complete(job):
 def start_streaming_jobs(
     session_id: str,
     user_id: str,
-    client_id: str,
-    always_persist: bool = False
+    client_id: str
 ) -> Dict[str, str]:
     """
     Enqueue jobs for streaming audio session (initial session setup).
@@ -339,12 +338,13 @@ def start_streaming_jobs(
         session_id: Stream session ID (equals client_id for streaming)
         user_id: User identifier
         client_id: Client identifier
-        always_persist: Whether to create placeholder conversation immediately (default: False)
 
     Returns:
         Dict with job IDs: {'speech_detection': job_id, 'audio_persistence': job_id}
 
-    Note: user_email is fetched from the database when needed.
+    Note:
+        - user_email is fetched from the database when needed.
+        - always_persist setting is read from global config by the audio persistence job.
     """
     from advanced_omi_backend.workers.transcription_jobs import stream_speech_detection_job
     from advanced_omi_backend.workers.audio_jobs import audio_streaming_persistence_job
@@ -383,12 +383,12 @@ def start_streaming_jobs(
     # Enqueue audio persistence job on dedicated audio queue
     # NOTE: This job handles file rotation for multiple conversations automatically
     # Runs for entire session, not tied to individual conversations
+    # The job reads always_persist_enabled from global config internally
     audio_job = audio_queue.enqueue(
         audio_streaming_persistence_job,
         session_id,
         user_id,
         client_id,
-        always_persist,
         job_timeout=86400,  # 24 hours for all-day sessions
         ttl=None,  # No pre-run expiry (job can wait indefinitely in queue)
         result_ttl=JOB_RESULT_TTL,  # Cleanup AFTER completion

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

@@ -15,6 +15,8 @@
 
 from advanced_omi_backend.config import (
     get_diarization_settings as load_diarization_settings,
+    get_misc_settings as load_misc_settings,
+    save_misc_settings,
 )
 from advanced_omi_backend.config import (
     save_diarization_settings,
@@ -333,6 +335,68 @@ async def save_diarization_settings_controller(settings: dict):
         raise e
 
 
+async def get_misc_settings():
+    """Get current miscellaneous settings."""
+    try:
+        # Get settings using OmegaConf
+        settings = load_misc_settings()
+        return {
+            "settings": settings,
+            "status": "success"
+        }
+    except Exception as e:
+        logger.exception("Error getting misc settings")
+        raise e
+
+
+async def save_misc_settings_controller(settings: dict):
+    """Save miscellaneous settings."""
+    try:
+        # Validate settings
+        valid_keys = {"always_persist_enabled", "use_provider_segments"}
+
+        # Filter to only valid keys
+        filtered_settings = {}
+        for key, value in settings.items():
+            if key not in valid_keys:
+                continue  # Skip unknown keys
+
+            # Type validation
+            if not isinstance(value, bool):
+                raise HTTPException(status_code=400, detail=f"Invalid value for {key}: must be boolean")
+
+            filtered_settings[key] = value
+
+        # Reject if NO valid keys provided
+        if not filtered_settings:
+            raise HTTPException(status_code=400, detail="No valid misc settings provided")
+
+        # Save using OmegaConf
+        if save_misc_settings(filtered_settings):
+            # Get updated settings
+            updated_settings = load_misc_settings()
+            logger.info(f"Updated and saved misc settings: {filtered_settings}")
+
+            return {
+                "message": "Miscellaneous settings saved successfully",
+                "settings": updated_settings,
+                "status": "success"
+            }
+        else:
+            logger.warning("Settings save failed")
+            return {
+                "message": "Settings save failed",
+                "settings": load_misc_settings(),
+                "status": "error"
+            }
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.exception("Error saving misc settings")
+        raise e
+
+
 async def get_cleanup_settings_controller(user: User) -> dict:
     """
     Get current cleanup settings (admin only).

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

@@ -451,14 +451,10 @@ async def _initialize_streaming_session(
     # Enqueue streaming jobs (speech detection + audio persistence)
     from advanced_omi_backend.controllers.queue_controller import start_streaming_jobs
 
-    # Get always_persist flag from client state
-    always_persist_flag = getattr(client_state, 'always_persist', False)
-
     job_ids = start_streaming_jobs(
         session_id=client_state.stream_session_id,
         user_id=user_id,
-        client_id=client_id,
-        always_persist=always_persist_flag
+        client_id=client_id
     )
 
     # Store job IDs in Redis session (not in ClientState)
@@ -468,8 +464,8 @@ async def _initialize_streaming_session(
         audio_persistence_job_id=job_ids['audio_persistence']
     )
 
-    # Note: Placeholder conversation creation (if always_persist=True) is now handled
-    # by the audio persistence job itself, making it self-sufficient.
+    # Note: Placeholder conversation creation is handled by the audio persistence job,
+    # which reads the always_persist_enabled setting from global config.
 
     # Launch interim results subscriber if WebSocket provided
     subscriber_task = None
@@ -822,11 +818,11 @@ async def _handle_audio_session_start(
     websocket: Optional[WebSocket] = None
 ) -> tuple[bool, str]:
     """
-    Handle audio-start event - validate mode, set recording mode, and extract always_persist flag.
+    Handle audio-start event - validate mode and set recording mode.
 
     Args:
         client_state: Client state object
-        audio_format: Audio format dict with mode and always_persist
+        audio_format: Audio format dict with mode
         client_id: Client ID
         websocket: Optional WebSocket connection (for WebUI error messages)
 
@@ -836,16 +832,14 @@ async def _handle_audio_session_start(
     from advanced_omi_backend.services.transcription import is_transcription_available
 
     recording_mode = audio_format.get("mode", "batch")
-    always_persist = audio_format.get("always_persist", False)
 
     application_logger.info(
         f"🔴 BACKEND: Received audio-start for {client_id} - "
-        f"mode={recording_mode}, always_persist={always_persist}, full format={audio_format}"
+        f"mode={recording_mode}, full format={audio_format}"
     )
 
     # Store on client state for later use
     client_state.recording_mode = recording_mode
-    client_state.always_persist = always_persist
 
     # VALIDATION: Check if streaming mode is available
     if recording_mode == "streaming":
@@ -898,8 +892,7 @@ async def _handle_audio_session_start(
         f"Format: {audio_format.get('rate')}Hz, "
         f"{audio_format.get('width')}bytes, "
         f"{audio_format.get('channels')}ch, "
-        f"Mode: {recording_mode}, "
-        f"Always Persist: {always_persist}"
+        f"Mode: {recording_mode}"
     )
 
     return True, recording_mode  # Switch to audio streaming mode
@@ -1358,7 +1351,7 @@ async def handle_pcm_websocket(
                             websocket=ws  # Pass websocket for WebUI error display
                         )
 
-                        # Initialize streaming session (for always_persist and job setup)
+                        # Initialize streaming session
                         if recording_mode == "streaming":
                             application_logger.info(f"🔴 BACKEND: Initializing streaming session for {client_id}")
                             interim_subscriber_task = await _initialize_streaming_session(

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

@@ -72,11 +72,18 @@ async def get_job_status(
             logger.error(f"Failed to determine status for job {job_id}: {e}")
             raise HTTPException(status_code=500, detail=str(e))
 
-        return {
+        response = {
             "job_id": job.id,
             "status": status
         }
 
+        # Include error information for failed jobs
+        if status == "failed" and job.exc_info:
+            response["error_message"] = str(job.exc_info)
+            response["exc_info"] = str(job.exc_info)
+
+        return response
+
     except HTTPException:
         # Re-raise HTTPException unchanged (e.g., 403 Forbidden)
         raise

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

@@ -63,6 +63,21 @@ async def save_diarization_settings(
     return await system_controller.save_diarization_settings_controller(settings)
 
 
+@router.get("/misc-settings")
+async def get_misc_settings(current_user: User = Depends(current_superuser)):
+    """Get miscellaneous configuration settings. Admin only."""
+    return await system_controller.get_misc_settings()
+
+
+@router.post("/misc-settings")
+async def save_misc_settings(
+    settings: dict,
+    current_user: User = Depends(current_superuser)
+):
+    """Save miscellaneous configuration settings. Admin only."""
+    return await system_controller.save_misc_settings_controller(settings)
+
+
 @router.get("/cleanup-settings")
 async def get_cleanup_settings(
     current_user: User = Depends(current_superuser)

backends/advanced/src/advanced_omi_backend/workers/audio_jobs.py

@@ -26,7 +26,6 @@ async def audio_streaming_persistence_job(
     session_id: str,
     user_id: str,
     client_id: str,
-    always_persist: bool = False,
     *,
     redis_client=None
 ) -> Dict[str, Any]:
@@ -42,15 +41,22 @@ async def audio_streaming_persistence_job(
         session_id: Stream session ID
         user_id: User ID
         client_id: Client ID
-        always_persist: Whether to create placeholder conversation immediately (default: False)
         redis_client: Redis client (injected by decorator)
 
     Returns:
         Dict with chunk_count, total_bytes, compressed_bytes, duration_seconds
 
-    Note: Replaces disk-based WAV file storage with MongoDB chunk storage.
+    Note:
+        - Replaces disk-based WAV file storage with MongoDB chunk storage.
+        - Reads always_persist_enabled from global config to determine whether to
+          create placeholder conversations immediately.
     """
-    logger.info(f"🎵 Starting MongoDB audio persistence for session {session_id}")
+    # Read always_persist setting from global config
+    from advanced_omi_backend.config import get_misc_settings
+    misc_settings = get_misc_settings()
+    always_persist = misc_settings.get('always_persist_enabled', False)
+
+    logger.info(f"🎵 Starting MongoDB audio persistence for session {session_id} (always_persist={always_persist})")
 
     # Setup audio persistence consumer group (separate from transcription consumer)
     audio_stream_name = f"audio:stream:{client_id}"

backends/advanced/src/advanced_omi_backend/workers/conversation_jobs.py

@@ -691,10 +691,12 @@ async def open_conversation_job(
     # Enqueue post-conversation processing pipeline (no batch transcription needed - using streaming transcript)
     client_id = conversation.client_id if conversation else None
 
+    # Enqueue post-conversation jobs directly (no fallback dependency in success case)
     job_ids = start_post_conversation_jobs(
         conversation_id=conversation_id,
         user_id=user_id,
         transcript_version_id=version_id,  # Pass the streaming transcript version ID
+        depends_on_job=None,  # No dependency - streaming already succeeded
         client_id=client_id,  # Pass client_id for UI tracking
         end_reason=end_reason  # Pass the determined end_reason (websocket_disconnect, inactivity_timeout, etc.)
     )