🐛 Bug fix: northbound api cannot store and get conversation history list

Author: Phinease

backend/services/agent_service.py

@@ -835,6 +835,7 @@ async def run_agent_stream(
     authorization: str,
     user_id: str = None,
     tenant_id: str = None,
+    skip_user_save: bool = False,
 ):
     """
     Start an agent run and stream responses.
@@ -850,10 +851,13 @@ async def run_agent_stream(
     )
 
     # Save user message only if not in debug mode (before streaming starts)
-    if not agent_request.is_debug:
-        save_messages(agent_request, target=MESSAGE_ROLE["USER"],
-                      user_id=resolved_user_id,
-                      tenant_id=resolved_tenant_id)
+    if not agent_request.is_debug and not skip_user_save:
+        save_messages(
+            agent_request,
+            target=MESSAGE_ROLE["USER"],
+            user_id=resolved_user_id,
+            tenant_id=resolved_tenant_id,
+        )
 
     memory_ctx_preview = build_memory_context(
         resolved_user_id, resolved_tenant_id, agent_request.agent_id

backend/services/northbound_service.py

@@ -25,6 +25,7 @@
     get_agent_id_by_name
 )
 from services.conversation_management_service import (
+    save_conversation_user,
     get_conversation_list_service,
     create_new_conversation,
     update_conversation_title as update_conversation_title_service,
@@ -165,7 +166,7 @@ async def start_streaming_chat(
             add_mapping_id(internal_id=internal_conversation_id, external_id=external_conversation_id, tenant_id=ctx.tenant_id, user_id=ctx.user_id)
 
         # Get history according to internal_conversation_id
-        history = await get_conversation_history(ctx, external_conversation_id)
+        history_resp = await get_conversation_history(ctx, external_conversation_id)
         agent_id = await get_agent_id_by_name(agent_name=agent_name, tenant_id=ctx.tenant_id)
         # Idempotency: only prevent concurrent duplicate starts
         composed_key = idempotency_key or _build_idempotency_key(ctx.tenant_id, external_conversation_id, agent_id, query)
@@ -174,11 +175,18 @@ async def start_streaming_chat(
             conversation_id=internal_conversation_id,
             agent_id=agent_id,
             query=query,
-            history=history.get("history"),
+            history=(history_resp.get("data", {})).get("history", []),
             minio_files=None,
             is_debug=False,
         )
 
+        # Synchronously persist the user message before starting the stream to avoid race conditions
+        try:
+            save_conversation_user(
+                agent_request, user_id=ctx.user_id, tenant_id=ctx.tenant_id)
+        except Exception as e:
+            raise Exception(f"Failed to persist user message: {str(e)}")
+
     except LimitExceededError as _:
         raise LimitExceededError("Query rate exceeded limit. Please try again later.")
     except UnauthorizedError as _:
@@ -193,6 +201,7 @@ async def start_streaming_chat(
             authorization=ctx.authorization,
             user_id=ctx.user_id,
             tenant_id=ctx.tenant_id,
+            skip_user_save=True,
         )
     finally:
         if composed_key: