Implement OpenMemory user registration and enhance MCP client functionality

- Added an asynchronous function to initialize and register an OpenMemory user if the OpenMemory MCP provider is configured, improving user management.
- Enhanced the MCPClient to accept custom metadata when adding memories, allowing for better tracking and filtering of memories by user.
- Updated the OpenMemoryMCPService to utilize the configured OpenMemory user for memory operations, ensuring accurate user context in memory processing.
- Modified integration tests to use shorter device names for consistency and to avoid truncation issues, improving test reliability.

Author: AnkushMalaker

backends/advanced/src/advanced_omi_backend/app_factory.py

@@ -42,6 +42,52 @@
 application_logger = logging.getLogger("audio_processing")
 
 
+async def initialize_openmemory_user() -> None:
+    """Initialize and register OpenMemory user if using OpenMemory MCP provider.
+
+    This function:
+    - Checks if OpenMemory MCP is configured as the memory provider
+    - Registers the configured user with OpenMemory server
+    - Creates a test memory and deletes it to trigger user creation
+    - Logs success or warning if OpenMemory is not reachable
+    """
+    from advanced_omi_backend.services.memory.config import build_memory_config_from_env, MemoryProvider
+
+    memory_provider_config = build_memory_config_from_env()
+
+    if memory_provider_config.memory_provider != MemoryProvider.OPENMEMORY_MCP:
+        return
+
+    try:
+        from advanced_omi_backend.services.memory.providers.mcp_client import MCPClient
+
+        # Get configured user_id and server_url
+        openmemory_config = memory_provider_config.openmemory_config
+        user_id = openmemory_config.get("user_id", "openmemory") if openmemory_config else "openmemory"
+        server_url = openmemory_config.get("server_url", "http://host.docker.internal:8765") if openmemory_config else "http://host.docker.internal:8765"
+        client_name = openmemory_config.get("client_name", "chronicle") if openmemory_config else "chronicle"
+
+        application_logger.info(f"Registering OpenMemory user: {user_id} at {server_url}")
+
+        # Make a lightweight registration call (create and delete dummy memory)
+        async with MCPClient(server_url=server_url, client_name=client_name, user_id=user_id) as client:
+            # Test connection first
+            is_connected = await client.test_connection()
+            if is_connected:
+                # Create and immediately delete a dummy memory to trigger user creation
+                memory_ids = await client.add_memories("Chronicle initialization - user registration test")
+                if memory_ids:
+                    # Delete the test memory
+                    await client.delete_memory(memory_ids[0])
+                application_logger.info(f"✅ Registered OpenMemory user: {user_id}")
+            else:
+                application_logger.warning(f"⚠️  OpenMemory MCP not reachable at {server_url}")
+                application_logger.info("User will be auto-created on first memory operation")
+    except Exception as e:
+        application_logger.warning(f"⚠️  Could not register OpenMemory user: {e}")
+        application_logger.info("User will be auto-created on first memory operation")
+
+
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     """Manage application lifespan events."""
@@ -126,6 +172,9 @@ async def lifespan(app: FastAPI):
     # Memory service will be lazily initialized when first used
     application_logger.info("Memory service will be initialized on first use (lazy loading)")
 
+    # Register OpenMemory user if using openmemory_mcp provider
+    await initialize_openmemory_user()
+
     # SystemTracker is used for monitoring and debugging
     application_logger.info("Using SystemTracker for monitoring and debugging")
 

backends/advanced/src/advanced_omi_backend/services/memory/providers/mcp_client.py

@@ -61,7 +61,7 @@ async def __aenter__(self):
     async def __aexit__(self, exc_type, exc_val, exc_tb):
         await self.close()
 
-    async def add_memories(self, text: str) -> List[str]:
+    async def add_memories(self, text: str, metadata: Dict[str, Any] = None) -> List[str]:
         """Add memories to the OpenMemory server.
         
         Uses the REST API to create memories. OpenMemory will handle:
@@ -109,35 +109,30 @@ async def add_memories(self, text: str) -> List[str]:
                 memory_logger.error("No apps found in OpenMemory - cannot create memory")
                 raise MCPError("No apps found in OpenMemory")
 
+            # Merge custom metadata with default metadata
+            default_metadata = {
+                "source": "chronicle",
+                "client": self.client_name,
+                "user_email": self.user_email
+            }
+            if metadata:
+                default_metadata.update(metadata)
+
             # Use REST API endpoint for creating memories
             # The 'app' field can be either app name (string) or app UUID
             payload = {
                 "user_id": self.user_id,
                 "text": text,
-                "app": self.client_name,  # Use app name (OpenMemory accepts name or UUID)
-                "metadata": {
-                    "source": "friend_lite",
-                    "client": self.client_name,
-                    "user_email": self.user_email
-                },
+                "app": self.client_name,
+                "metadata": default_metadata,
                 "infer": True
             }
 
             memory_logger.info(f"POSTing memory to {self.server_url}/api/v1/memories/ with payload={payload}")
 
             response = await self.client.post(
                 f"{self.server_url}/api/v1/memories/",
-                json={
-                    "user_id": self.user_id,
-                    "text": text,
-                    "app": self.client_name,  # Use app name (OpenMemory accepts name or UUID)
-                    "metadata": {
-                        "source": "chronicle",
-                        "client": self.client_name,
-                        "user_email": self.user_email
-                    },
-                    "infer": True
-                }
+                json=payload
             )
 
             response_body = response.text[:500] if response.status_code != 200 else "..."

backends/advanced/src/advanced_omi_backend/services/memory/providers/openmemory_mcp.py

@@ -142,25 +142,21 @@ async def add_memory(
                 memory_logger.info(f"Skipping empty transcript for {source_id}")
                 return True, []
 
-            # Pass Friend-Lite user details to OpenMemory for proper user tracking
-            # OpenMemory will auto-create users if they don't exist
-            original_user_id = self.mcp_client.user_id
-            original_user_email = self.mcp_client.user_email
-            self.mcp_client.user_id = user_id  # Use the actual Chronicle user's ID
-            self.mcp_client.user_email = user_email  # Use the actual user's email
+            # Use configured OpenMemory user (from config) for all Chronicle users
+            # Chronicle user_id and email are stored in metadata for filtering
+            enriched_transcript = f"[Source: {source_id}, Client: {client_id}] {transcript}"
 
-            try:
-                # Thin client approach: Send raw transcript to OpenMemory MCP server
-                # OpenMemory handles: extraction, deduplication, vector storage, ACL
-                enriched_transcript = f"[Source: {source_id}, Client: {client_id}] {transcript}"
+            memory_logger.info(f"Delegating memory processing to OpenMemory for user {user_id} (email: {user_email}), source {source_id}")
 
-                memory_logger.info(f"Delegating memory processing to OpenMemory for user {user_id} (email: {user_email}), source {source_id}")
-                memory_ids = await self.mcp_client.add_memories(text=enriched_transcript)
+            # Pass Chronicle user details in metadata for filtering/search
+            metadata = {
+                "chronicle_user_id": user_id,
+                "chronicle_user_email": user_email,
+                "source_id": source_id,
+                "client_id": client_id
+            }
 
-            finally:
-                # Restore original user context
-                self.mcp_client.user_id = original_user_id
-                self.mcp_client.user_email = original_user_email
+            memory_ids = await self.mcp_client.add_memories(text=enriched_transcript, metadata=metadata)
 
             # Update database relationships if helper provided
             if memory_ids and db_helper:
@@ -204,23 +200,27 @@ async def search_memories(
         """
         if not self._initialized:
             await self.initialize()
-        
-        # Update MCP client user context for this search operation
-        original_user_id = self.mcp_client.user_id
-        self.mcp_client.user_id = user_id  # Use the actual Chronicle user's ID
 
+        # Use configured OpenMemory user (not Chronicle user_id)
+        # Search all memories, then filter by chronicle_user_id in metadata
         try:
+            # Get more results since we'll filter by user
             results = await self.mcp_client.search_memory(
                 query=query,
-                limit=limit
+                limit=limit * 3  # Get extra to account for filtering
             )
 
-            # Convert MCP results to MemoryEntry objects
+            # Convert MCP results to MemoryEntry objects and filter by user
             memory_entries = []
             for result in results:
-                memory_entry = self._mcp_result_to_memory_entry(result, user_id)
-                if memory_entry:
-                    memory_entries.append(memory_entry)
+                # Check if memory belongs to this Chronicle user via metadata
+                metadata = result.get("metadata", {})
+                if metadata.get("chronicle_user_id") == user_id:
+                    memory_entry = self._mcp_result_to_memory_entry(result, user_id)
+                    if memory_entry:
+                        memory_entries.append(memory_entry)
+                        if len(memory_entries) >= limit:
+                            break  # Got enough results
 
             memory_logger.info(f"🔍 Found {len(memory_entries)} memories for query '{query}' (user: {user_id})")
             return memory_entries
@@ -231,9 +231,6 @@ async def search_memories(
         except Exception as e:
             memory_logger.error(f"Search memories failed: {e}")
             return []
-        finally:
-            # Restore original user context
-            self.mcp_client.user_id = original_user_id
 
     async def get_all_memories(
         self, 
@@ -254,20 +251,24 @@ async def get_all_memories(
         """
         if not self._initialized:
             await self.initialize()
-        
-        # Update MCP client user context for this operation
-        original_user_id = self.mcp_client.user_id
-        self.mcp_client.user_id = user_id  # Use the actual Chronicle user's ID
 
+        # Use configured OpenMemory user (not Chronicle user_id)
+        # List all memories, then filter by chronicle_user_id in metadata
         try:
-            results = await self.mcp_client.list_memories(limit=limit)
+            # Get more results since we'll filter by user
+            results = await self.mcp_client.list_memories(limit=limit * 3)
 
-            # Convert MCP results to MemoryEntry objects
+            # Convert MCP results to MemoryEntry objects and filter by user
             memory_entries = []
             for result in results:
-                memory_entry = self._mcp_result_to_memory_entry(result, user_id)
-                if memory_entry:
-                    memory_entries.append(memory_entry)
+                # Check if memory belongs to this Chronicle user via metadata
+                metadata = result.get("metadata", {})
+                if metadata.get("chronicle_user_id") == user_id:
+                    memory_entry = self._mcp_result_to_memory_entry(result, user_id)
+                    if memory_entry:
+                        memory_entries.append(memory_entry)
+                        if len(memory_entries) >= limit:
+                            break  # Got enough results
 
             memory_logger.info(f"📚 Retrieved {len(memory_entries)} memories for user {user_id}")
             return memory_entries
@@ -278,9 +279,6 @@ async def get_all_memories(
         except Exception as e:
             memory_logger.error(f"Get all memories failed: {e}")
             return []
-        finally:
-            # Restore original user_id
-            self.mcp_client.user_id = original_user_id
 
     async def get_memory(self, memory_id: str, user_id: Optional[str] = None) -> Optional[MemoryEntry]:
         """Get a specific memory by ID.

tests/infrastructure/infra_tests.robot

@@ -159,7 +159,7 @@ Worker Registration Loss Detection Test
     ...                - Health endpoint reports 0 workers when registration is lost
     ...                - Self-healing mechanism detects the issue
     ...                - Workers automatically re-register within monitoring interval
-    [Tags]    infra	queue
+    [Tags]    infra	queue	slow
 
     # Step 1: Verify workers are initially registered
     Log To Console    \n📊 Step 1: Check initial worker registration

tests/integration/always_persist_audio_tests.robot

@@ -180,12 +180,13 @@ Multiple Sessions Create Separate Conversations
     ...                creates separate placeholder conversations for each session.
     [Tags]    conversation	audio-streaming
 
-    ${device_name}=    Set Variable    test-multi
+    # NOTE: Device names must be <=10 chars to be unique (backend truncates to 10 chars)
+    # Using short names: multi-1, multi-2, multi-3 (7 chars each)
 
     # Get client IDs for each device
-    ${client_id_1}=    Get Client ID From Device Name    ${device_name}-1
-    ${client_id_2}=    Get Client ID From Device Name    ${device_name}-2
-    ${client_id_3}=    Get Client ID From Device Name    ${device_name}-3
+    ${client_id_1}=    Get Client ID From Device Name    multi-1
+    ${client_id_2}=    Get Client ID From Device Name    multi-2
+    ${client_id_3}=    Get Client ID From Device Name    multi-3
 
     # Get baseline conversation counts for each client
     ${convs_before_1}=    Get Conversations By Client ID    ${client_id_1}
@@ -199,11 +200,11 @@ Multiple Sessions Create Separate Conversations
     ${expected_count_3}=    Evaluate    ${count_before_3} + 1
 
     # Start 3 separate sessions
-    ${stream_1}=    Open Audio Stream With Always Persist    device_name=${device_name}-1
+    ${stream_1}=    Open Audio Stream With Always Persist    device_name=multi-1
     Sleep    1s
-    ${stream_2}=    Open Audio Stream With Always Persist    device_name=${device_name}-2
+    ${stream_2}=    Open Audio Stream With Always Persist    device_name=multi-2
     Sleep    1s
-    ${stream_3}=    Open Audio Stream With Always Persist    device_name=${device_name}-3
+    ${stream_3}=    Open Audio Stream With Always Persist    device_name=multi-3
 
     # Poll for each conversation to be created (audio persistence jobs may take 10-15s)
     ${convs_after_1}=    Wait Until Keyword Succeeds    30s    2s