Persist chat history metadata in OpenSearch for multi-instance backends #1073

src/agent.py

@@ -11,9 +11,9 @@
 active_conversations = {}
 
 
-def get_user_conversations(user_id: str):
+async def get_user_conversations(user_id: str):
     """Get conversation metadata for a user from persistent storage"""
-    return conversation_persistence.get_user_conversations(user_id)
+    return await conversation_persistence.get_user_conversations(user_id)
 
 
 def get_conversation_thread(user_id: str, previous_response_id: str = None):
@@ -82,7 +82,7 @@ async def store_conversation_thread(user_id: str, response_id: str, conversation
 
 
 # Legacy function for backward compatibility
-def get_user_conversation(user_id: str):
+async def get_user_conversation(user_id: str):
     """Get the most recent conversation for a user (for backward compatibility)"""
     # Check in-memory conversations first (with function calls)
     if user_id in active_conversations and active_conversations[user_id]:
@@ -93,7 +93,7 @@ def get_user_conversation(user_id: str):
         return active_conversations[user_id][latest_response_id]
 
     # Fallback to metadata-only conversations
-    conversations = get_user_conversations(user_id)
+    conversations = await get_user_conversations(user_id)
     if not conversations:
         return get_conversation_thread(user_id)
 
@@ -461,7 +461,7 @@ async def async_chat(
         )
 
         # Debug: Check what's in user_conversations now
-        conversations = get_user_conversations(user_id)
+        conversations = await get_user_conversations(user_id)
         logger.debug(
             "User conversations updated",
             user_id=user_id,
@@ -668,7 +668,7 @@ async def async_langflow_chat(
         )
 
         # Debug: Check what's in user_conversations now
-        conversations = get_user_conversations(user_id)
+        conversations = await get_user_conversations(user_id)
         logger.debug(
             "User conversations updated",
             user_id=user_id,

src/main.py

@@ -59,6 +59,10 @@
 from services.auth_service import AuthService
 from services.langflow_mcp_service import LangflowMCPService
 from services.chat_service import ChatService
+from services.conversation_persistence_service import (
+    CONVERSATION_METADATA_INDEX_BODY,
+    CONVERSATION_METADATA_INDEX_NAME,
+)
 
 # Services
 from services.document_service import DocumentService
@@ -230,6 +234,22 @@ async def init_index():
                 index_name=API_KEYS_INDEX_NAME,
             )
 
+        # Create chat conversation metadata index for horizontally scaled backends
+        if not await clients.opensearch.indices.exists(index=CONVERSATION_METADATA_INDEX_NAME):
+            await clients.opensearch.indices.create(
+                index=CONVERSATION_METADATA_INDEX_NAME,
+                body=CONVERSATION_METADATA_INDEX_BODY,
+            )
+            logger.info(
+                "Created chat conversation metadata index",
+                index_name=CONVERSATION_METADATA_INDEX_NAME,
+            )
+        else:
+            logger.info(
+                "Chat conversation metadata index already exists, skipping creation",
+                index_name=CONVERSATION_METADATA_INDEX_NAME,
+            )
+
         # Configure alerting plugin security settings
         await configure_alerting_security()
 

src/services/chat_service.py

@@ -368,7 +368,7 @@ async def get_chat_history(self, user_id: str):
             return {"error": "User ID is required", "conversations": []}
 
         # Get metadata from persistent storage
-        conversations_dict = get_user_conversations(user_id)
+        conversations_dict = await get_user_conversations(user_id)
 
         # Get in-memory conversations (with function calls)
         in_memory_conversations = active_conversations.get(user_id, {})
@@ -484,7 +484,7 @@ async def get_langflow_history(self, user_id: str):
 
         try:
             # 1. Get local conversation metadata (no actual messages stored here)
-            conversations_dict = get_user_conversations(user_id)
+            conversations_dict = await get_user_conversations(user_id)
             local_metadata = {}
 
             for response_id, conversation_metadata in conversations_dict.items():