Implement Redis integration for client-user mapping and enhance wake word processing

- Added asynchronous Redis support in ClientManager for tracking client-user relationships.
- Introduced `initialize_redis_for_client_manager` to set up Redis for cross-container mapping.
- Updated `create_client_state` to use asynchronous tracking for client-user relationships.
- Enhanced wake word processing in PluginRouter with normalization and command extraction.
- Refactored DeepgramStreamingConsumer to utilize async Redis lookups for user ID retrieval.
- Set TTL on Redis streams during client state cleanup for better resource management.

Author: AnkushMalaker

backends/advanced/src/advanced_omi_backend/app_factory.py

@@ -111,6 +111,11 @@ async def lifespan(app: FastAPI):
         from advanced_omi_backend.services.audio_stream import AudioStreamProducer
         app.state.audio_stream_producer = AudioStreamProducer(app.state.redis_audio_stream)
         application_logger.info("✅ Redis client for audio streaming producer initialized")
+
+        # Initialize ClientManager Redis for cross-container client→user mapping
+        from advanced_omi_backend.client_manager import initialize_redis_for_client_manager
+        initialize_redis_for_client_manager(config.redis_url)
+
     except Exception as e:
         application_logger.error(f"Failed to initialize Redis client for audio streaming: {e}", exc_info=True)
         application_logger.warning("Audio streaming producer will not be available")

backends/advanced/src/advanced_omi_backend/client_manager.py

@@ -9,6 +9,7 @@
 import logging
 import uuid
 from typing import TYPE_CHECKING, Dict, Optional
+import redis.asyncio as redis
 
 if TYPE_CHECKING:
     from advanced_omi_backend.client import ClientState
@@ -21,6 +22,9 @@
 _client_to_user_mapping: Dict[str, str] = {}  # Active clients only
 _all_client_user_mappings: Dict[str, str] = {}  # All clients including disconnected
 
+# Redis client for cross-container client→user mapping
+_redis_client: Optional[redis.Redis] = None
+
 
 class ClientManager:
     """
@@ -372,9 +376,33 @@ def unregister_client_user_mapping(client_id: str):
         logger.warning(f"⚠️ Attempted to unregister non-existent client {client_id}")
 
 
+async def track_client_user_relationship_async(client_id: str, user_id: str, ttl: int = 86400):
+    """
+    Track that a client belongs to a user (async, writes to Redis for cross-container support).
+
+    Args:
+        client_id: The client ID
+        user_id: The user ID that owns this client
+        ttl: Time-to-live in seconds (default 24 hours)
+    """
+    _all_client_user_mappings[client_id] = user_id  # In-memory fallback
+
+    if _redis_client:
+        try:
+            await _redis_client.setex(f"client:owner:{client_id}", ttl, user_id)
+            logger.debug(f"✅ Tracked client {client_id} → user {user_id} in Redis (TTL: {ttl}s)")
+        except Exception as e:
+            logger.warning(f"Failed to track client in Redis: {e}")
+    else:
+        logger.debug(f"Tracked client {client_id} relationship to user {user_id} (in-memory only)")
+
+
 def track_client_user_relationship(client_id: str, user_id: str):
     """
-    Track that a client belongs to a user (persists after disconnection for database queries).
+    Track that a client belongs to a user (sync version for backward compatibility).
+
+    WARNING: This is synchronous and cannot use Redis. Use track_client_user_relationship_async()
+    instead in async contexts for cross-container support.
 
     Args:
         client_id: The client ID
@@ -444,9 +472,45 @@ def get_user_clients_active(user_id: str) -> list[str]:
     return user_clients
 
 
+def initialize_redis_for_client_manager(redis_url: str):
+    """
+    Initialize Redis client for cross-container client→user mapping.
+
+    Args:
+        redis_url: Redis connection URL
+    """
+    global _redis_client
+    _redis_client = redis.from_url(redis_url, decode_responses=True)
+    logger.info(f"✅ ClientManager Redis initialized: {redis_url}")
+
+
+async def get_client_owner_async(client_id: str) -> Optional[str]:
+    """
+    Get the user ID that owns a specific client (async Redis lookup).
+
+    Args:
+        client_id: The client ID to look up
+
+    Returns:
+        User ID if found, None otherwise
+    """
+    if _redis_client:
+        try:
+            user_id = await _redis_client.get(f"client:owner:{client_id}")
+            return user_id
+        except Exception as e:
+            logger.warning(f"Redis lookup failed for client {client_id}: {e}")
+
+    # Fallback to in-memory mapping
+    return _all_client_user_mappings.get(client_id)
+
+
 def get_client_owner(client_id: str) -> Optional[str]:
     """
-    Get the user ID that owns a specific client.
+    Get the user ID that owns a specific client (sync version for backward compatibility).
+
+    WARNING: This is synchronous and cannot use Redis. Use get_client_owner_async() instead
+    in async contexts for cross-container support.
 
     Args:
         client_id: The client ID to look up

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

@@ -189,9 +189,9 @@ async def create_client_state(client_id: str, user, device_name: Optional[str] =
         client_id, CHUNK_DIR, user.user_id, user.email
     )
 
-    # Also track in persistent mapping (for database queries)
-    from advanced_omi_backend.client_manager import track_client_user_relationship
-    track_client_user_relationship(client_id, user.user_id)
+    # Also track in persistent mapping (for database queries + cross-container Redis)
+    from advanced_omi_backend.client_manager import track_client_user_relationship_async
+    await track_client_user_relationship_async(client_id, user.user_id)
 
     # Register client in user model (persistent)
     from advanced_omi_backend.users import register_client_to_user
@@ -265,12 +265,12 @@ async def cleanup_client_state(client_id: str):
         if sessions_closed > 0:
             logger.info(f"✅ Closed {sessions_closed} active session(s) for client {client_id}")
 
-        # Delete Redis Streams for this client
+        # Set TTL on Redis Streams for this client (allows consumer groups to finish processing)
         stream_pattern = f"audio:stream:{client_id}"
         stream_key = await async_redis.exists(stream_pattern)
         if stream_key:
-            await async_redis.delete(stream_pattern)
-            logger.info(f"🧹 Deleted Redis stream: {stream_pattern}")
+            await async_redis.expire(stream_pattern, 60)  # 60 second TTL for consumer group fan-out
+            logger.info(f"⏰ Set 60s TTL on Redis stream: {stream_pattern}")
         else:
             logger.debug(f"No Redis stream found for client {client_id}")
 

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

@@ -5,13 +5,80 @@
 """
 
 import logging
+import re
+import string
 from typing import Dict, List, Optional
 
 from .base import BasePlugin, PluginContext, PluginResult
 
 logger = logging.getLogger(__name__)
 
 
+def normalize_text_for_wake_word(text: str) -> str:
+    """
+    Normalize text for wake word matching.
+    - Lowercase
+    - Remove punctuation
+    - Collapse multiple spaces to single space
+    - Strip leading/trailing whitespace
+
+    Example:
+        "Hey, Vivi!" -> "hey vivi"
+        "HEY  VIVI" -> "hey vivi"
+    """
+    # Lowercase
+    text = text.lower()
+    # Remove punctuation
+    text = text.translate(str.maketrans('', '', string.punctuation))
+    # Normalize whitespace (collapse multiple spaces to single space)
+    text = re.sub(r'\s+', ' ', text)
+    # Strip leading/trailing whitespace
+    return text.strip()
+
+
+def extract_command_after_wake_word(transcript: str, wake_word: str) -> str:
+    """
+    Intelligently extract command after wake word in original transcript.
+
+    Handles punctuation and spacing variations by creating a flexible regex pattern.
+
+    Example:
+        transcript: "Hey, Vivi, turn off lights"
+        wake_word: "hey vivi"
+        -> extracts: "turn off lights"
+
+    Args:
+        transcript: Original transcript text with punctuation
+        wake_word: Configured wake word (will be normalized)
+
+    Returns:
+        Command text after wake word, or full transcript if wake word boundary not found
+    """
+    # Split wake word into parts (normalized)
+    wake_word_parts = normalize_text_for_wake_word(wake_word).split()
+
+    if not wake_word_parts:
+        return transcript.strip()
+
+    # Create regex pattern that allows punctuation/whitespace between parts
+    # Example: "hey" + "vivi" -> r"hey[\s,.\-!?]*vivi"
+    pattern_parts = [re.escape(part) for part in wake_word_parts]
+    pattern = r'\s*[\W_]*\s*'.join(pattern_parts)
+    pattern = '^' + pattern  # Must be at start of transcript
+
+    # Try to match wake word at start of transcript (case-insensitive)
+    match = re.match(pattern, transcript, re.IGNORECASE)
+
+    if match:
+        # Extract everything after the matched wake word
+        command = transcript[match.end():].strip()
+        return command
+    else:
+        # Fallback: couldn't find wake word boundary, return full transcript
+        logger.warning(f"Could not find wake word boundary for '{wake_word}' in '{transcript}', using full transcript")
+        return transcript.strip()
+
+
 class PluginRouter:
     """Routes pipeline events to appropriate plugins based on access level and triggers"""
 
@@ -113,9 +180,9 @@ async def _should_trigger(self, plugin: BasePlugin, data: Dict) -> bool:
             return True
 
         elif trigger_type == 'wake_word':
-            # Check if transcript starts with wake word(s)
+            # Normalize transcript for matching (handles punctuation and spacing)
             transcript = data.get('transcript', '')
-            transcript_lower = transcript.lower().strip()
+            normalized_transcript = normalize_text_for_wake_word(transcript)
 
             # Support both singular 'wake_word' and plural 'wake_words' (list)
             wake_words = plugin.trigger.get('wake_words', [])
@@ -125,14 +192,15 @@ async def _should_trigger(self, plugin: BasePlugin, data: Dict) -> bool:
                 if wake_word:
                     wake_words = [wake_word]
 
-            # Check if transcript starts with any wake word
+            # Check if transcript starts with any wake word (after normalization)
             for wake_word in wake_words:
-                wake_word_lower = wake_word.lower()
-                if wake_word_lower and transcript_lower.startswith(wake_word_lower):
-                    # Extract command (remove wake word)
-                    command = transcript[len(wake_word):].strip()
+                normalized_wake_word = normalize_text_for_wake_word(wake_word)
+                if normalized_wake_word and normalized_transcript.startswith(normalized_wake_word):
+                    # Smart extraction: find where wake word actually ends in original text
+                    command = extract_command_after_wake_word(transcript, wake_word)
                     data['command'] = command
                     data['original_transcript'] = transcript
+                    logger.debug(f"Wake word '{wake_word}' detected. Original: '{transcript}', Command: '{command}'")
                     return True
 
             return False

backends/advanced/src/advanced_omi_backend/services/transcription/deepgram_stream_consumer.py

@@ -19,7 +19,7 @@
 
 from advanced_omi_backend.plugins.router import PluginRouter
 from advanced_omi_backend.services.transcription import get_transcription_provider
-from advanced_omi_backend.client_manager import get_client_owner
+from advanced_omi_backend.client_manager import get_client_owner_async
 
 logger = logging.getLogger(__name__)
 
@@ -257,20 +257,20 @@ async def store_final_result(self, session_id: str, result: Dict, chunk_id: str
 
     async def _get_user_id_from_client_id(self, client_id: str) -> Optional[str]:
         """
-        Look up user_id from client_id using ClientManager.
+        Look up user_id from client_id using ClientManager (async Redis lookup).
 
         Args:
             client_id: Client ID to search for
 
         Returns:
             user_id if found, None otherwise
         """
-        user_id = get_client_owner(client_id)
+        user_id = await get_client_owner_async(client_id)
 
         if user_id:
-            logger.debug(f"Found user_id {user_id} for client_id {client_id}")
+            logger.debug(f"Found user_id {user_id} for client_id {client_id} via Redis")
         else:
-            logger.warning(f"No user_id found for client_id {client_id}")
+            logger.warning(f"No user_id found for client_id {client_id} in Redis")
 
         return user_id
 

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

@@ -18,6 +18,7 @@
 
 from advanced_omi_backend.services.plugin_service import init_plugin_router
 from advanced_omi_backend.services.transcription.deepgram_stream_consumer import DeepgramStreamingConsumer
+from advanced_omi_backend.client_manager import initialize_redis_for_client_manager
 
 logging.basicConfig(
     level=logging.INFO,
@@ -48,6 +49,10 @@ async def main():
             decode_responses=False
         )
         logger.info(f"✅ Connected to Redis: {redis_url}")
+
+        # Initialize ClientManager Redis for cross-container client→user mapping
+        initialize_redis_for_client_manager(redis_url)
+
     except Exception as e:
         logger.error(f"Failed to connect to Redis: {e}", exc_info=True)
         sys.exit(1)