Merge pull request #1129 from asimurka/rebase_conversation_history

tisnik · web-flow · commit 81e303a33392 · 2026-02-10T20:52:09.000+01:00
LCORE-1166: Rebased conversation history changes
diff --git a/src/app/endpoints/query.py b/src/app/endpoints/query.py
@@ -190,13 +190,16 @@ async def query_endpoint_handler(
         quota_limiters=configuration.quota_limiters, user_id=user_id
     )
 
+    completed_at = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
     conversation_id = normalize_conversation_id(responses_params.conversation)
+
     logger.info("Storing query results")
     store_query_results(
         user_id=user_id,
         conversation_id=conversation_id,
-        model_id=responses_params.model,
+        model=responses_params.model,
         started_at=started_at,
+        completed_at=completed_at,
         summary=turn_summary,
         query_request=query_request,
         configuration=configuration,
diff --git a/src/app/endpoints/streaming_query.py b/src/app/endpoints/streaming_query.py
@@ -366,13 +366,15 @@ async def generate_response(
         turn_summary.referenced_documents,
         context.query_request.media_type or MEDIA_TYPE_JSON,
     )
+    completed_at = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
 
     # Store query results (transcript, conversation details, cache)
     logger.info("Storing query results")
     store_query_results(
         user_id=context.user_id,
         conversation_id=context.conversation_id,
-        model_id=responses_params.model,
+        model=responses_params.model,
+        completed_at=completed_at,
         started_at=context.started_at,
         summary=turn_summary,
         query_request=context.query_request,
diff --git a/src/utils/conversations.py b/src/utils/conversations.py
@@ -297,85 +297,128 @@ def _create_turn_from_db_metadata(
     )
 
 
-def build_conversation_turns_from_items(
+def _group_items_into_turns(
     items: list[ItemListResponse],
-    turns_metadata: list[UserTurn],
-    conversation_start_time: datetime,
-) -> list[ConversationTurn]:
-    """Build conversation turns from Conversations API items and turns metadata.
+) -> list[list[ItemListResponse]]:
+    """Group conversation items into turns.
+
+    Each turn starts with a user message. All subsequent messages and tool items
+    belong to that turn until the next user message.
 
     Args:
         items: Conversation items list from Conversations API, oldest first
-        turns_metadata: List of UserTurn database objects ordered by turn_number.
-            Can be empty for legacy conversations without stored metadata.
-        conversation_start_time: Timestamp to use for dummy metadata in legacy conversations.
-            Typically the conversation's created_at timestamp.
 
     Returns:
-        List of ConversationTurn objects, oldest first
+        List of turns, where each turn is a list of items belonging to that turn
     """
-    chat_history: list[ConversationTurn] = []
-    current_messages: list[Message] = []
-    current_tool_calls: list[ToolCallSummary] = []
-    current_tool_results: list[ToolResultSummary] = []
-    current_turn_index = 0
+    turns: list[list[ItemListResponse]] = []
+    current_turn_items: list[ItemListResponse] = []
 
     for item in items:
         item_type = getattr(item, "type", None)
 
-        # Parse message items
+        # User message marks the beginning of a new turn
         if item_type == "message":
             message_item = cast(MessageOutput, item)
-            message = _parse_message_item(message_item)
-
-            # User message marks the beginning of a new turn
-            if message.type == "user":
+            if message_item.role == "user":
                 # If we have accumulated items, finish the previous turn
-                if current_messages or current_tool_calls or current_tool_results:
-                    turn_metadata = (
-                        turns_metadata[current_turn_index]
-                        if current_turn_index < len(turns_metadata)
-                        else _create_dummy_turn_metadata(conversation_start_time)
-                    )
-                    chat_history.append(
-                        _create_turn_from_db_metadata(
-                            turn_metadata,
-                            current_messages,
-                            current_tool_calls,
-                            current_tool_results,
-                        )
-                    )
-                    current_turn_index += 1
+                if current_turn_items:
+                    turns.append(current_turn_items)
+                    current_turn_items = []
 
                 # Start new turn with this user message
-                current_messages = [message]
-                current_tool_calls = []
-                current_tool_results = []
+                current_turn_items = [item]
             else:
                 # Add non-user message to current turn
-                current_messages.append(message)
+                current_turn_items.append(item)
+        else:
+            # Add tool-related items to current turn
+            current_turn_items.append(item)
 
-        # Parse tool-related items
+    # Add final turn if there are items
+    if current_turn_items:
+        turns.append(current_turn_items)
+
+    return turns
+
+
+def _process_turn_items(
+    turn_items: list[ItemListResponse],
+) -> tuple[list[Message], list[ToolCallSummary], list[ToolResultSummary]]:
+    """Process items from a single turn into messages, tool calls, and tool results.
+
+    Args:
+        turn_items: List of items belonging to a single turn
+
+    Returns:
+        Tuple of (messages, tool_calls, tool_results)
+    """
+    messages: list[Message] = []
+    tool_calls: list[ToolCallSummary] = []
+    tool_results: list[ToolResultSummary] = []
+
+    for item in turn_items:
+        item_type = getattr(item, "type", None)
+
+        if item_type == "message":
+            message_item = cast(MessageOutput, item)
+            message = _parse_message_item(message_item)
+            messages.append(message)
         else:
             tool_call, tool_result = _build_tool_call_summary_from_item(item)
             if tool_call is not None:
-                current_tool_calls.append(tool_call)
+                tool_calls.append(tool_call)
             if tool_result is not None:
-                current_tool_results.append(tool_result)
+                tool_results.append(tool_result)
 
-    # Add final turn if there are items
-    if current_messages or current_tool_calls or current_tool_results:
-        turn_metadata = (
-            turns_metadata[current_turn_index]
-            if current_turn_index < len(turns_metadata)
-            else _create_dummy_turn_metadata(conversation_start_time)
-        )
+    return messages, tool_calls, tool_results
+
+
+def build_conversation_turns_from_items(
+    items: list[ItemListResponse],
+    turns_metadata: list[UserTurn],
+    conversation_start_time: datetime,
+) -> list[ConversationTurn]:
+    """Build conversation turns from Conversations API items and turns metadata.
+
+    Args:
+        items: Conversation items list from Conversations API, oldest first
+        turns_metadata: List of UserTurn database objects ordered by turn_number.
+            Can be empty for legacy conversations without stored metadata.
+            For extended legacy conversations, only the newer turns have metadata.
+        conversation_start_time: Timestamp to use for dummy metadata in legacy conversations.
+            Typically the conversation's created_at timestamp.
+
+    Returns:
+        List of ConversationTurn objects, oldest first
+    """
+    # Group items into turns first
+    turn_items_list = _group_items_into_turns(items)
+
+    # Calculate how many legacy turns don't have metadata
+    total_turns = len(turn_items_list)
+    legacy_turns_count = total_turns - len(turns_metadata)
+
+    # Process each turn with its corresponding metadata
+    chat_history: list[ConversationTurn] = []
+    for turn_index, turn_items in enumerate(turn_items_list):
+        # Process items into messages, tool calls, and tool results
+        messages, tool_calls, tool_results = _process_turn_items(turn_items)
+
+        # Select appropriate metadata for this turn
+        if turn_index < legacy_turns_count:
+            turn_metadata = _create_dummy_turn_metadata(conversation_start_time)
+        else:
+            metadata_index = turn_index - legacy_turns_count
+            turn_metadata = turns_metadata[metadata_index]
+
+        # Create ConversationTurn from metadata and processed items
         chat_history.append(
             _create_turn_from_db_metadata(
                 turn_metadata,
-                current_messages,
-                current_tool_calls,
-                current_tool_results,
+                messages,
+                tool_calls,
+                tool_results,
             )
         )
 
diff --git a/src/utils/query.py b/src/utils/query.py
@@ -13,10 +13,11 @@
 from llama_stack_client.types import ModelListResponse, Shield
 
 from fastapi import HTTPException
+from sqlalchemy import func
 from configuration import AppConfig, configuration
 from models.cache_entry import CacheEntry
 from models.config import Action
-from models.database.conversations import UserConversation
+from models.database.conversations import UserConversation, UserTurn
 import constants
 from models.requests import Attachment, QueryRequest
 from models.responses import (
@@ -330,8 +331,9 @@ def prepare_input(query_request: QueryRequest) -> str:
 def store_query_results(  # pylint: disable=too-many-arguments,too-many-locals
     user_id: str,
     conversation_id: str,
-    model_id: str,
+    model: str,
     started_at: str,
+    completed_at: str,
     summary: TurnSummary,
     query_request: QueryRequest,
     configuration: AppConfig,
@@ -349,8 +351,9 @@ def store_query_results(  # pylint: disable=too-many-arguments,too-many-locals
     Args:
         user_id: The authenticated user ID
         conversation_id: The conversation ID
-        model_id: The model identifier
+        model: The model identifier
         started_at: ISO formatted timestamp when the request started
+        completed_at: ISO formatted timestamp when the request completed
         summary: Summary of the turn including LLM response and tool calls
         query_request: The original query request
         configuration: Application configuration
@@ -360,7 +363,7 @@ def store_query_results(  # pylint: disable=too-many-arguments,too-many-locals
     Raises:
         HTTPException: On any database, cache, or IO errors during processing
     """
-    provider, model = extract_provider_and_model_from_model_id(model_id)
+    provider_id, model_id = extract_provider_and_model_from_model_id(model)
     # Store transcript if enabled
     if is_transcripts_enabled():
         try:
@@ -370,8 +373,8 @@ def store_query_results(  # pylint: disable=too-many-arguments,too-many-locals
             store_transcript(
                 user_id=user_id,
                 conversation_id=conversation_id,
-                model_id=model,
-                provider_id=provider,
+                model_id=model_id,
+                provider_id=provider_id,
                 query_is_valid=True,  # TODO(lucasagomes): implement as part of query validation
                 query=query_request.query,
                 query_request=query_request,
@@ -394,8 +397,10 @@ def store_query_results(  # pylint: disable=too-many-arguments,too-many-locals
         persist_user_conversation_details(
             user_id=user_id,
             conversation_id=conversation_id,
-            model=model,
-            provider_id=provider,
+            started_at=started_at,
+            completed_at=completed_at,
+            model_id=model_id,
+            provider_id=provider_id,
             topic_summary=topic_summary,
         )
     except SQLAlchemyError as e:
@@ -405,12 +410,11 @@ def store_query_results(  # pylint: disable=too-many-arguments,too-many-locals
 
     # Store conversation in cache
     try:
-        completed_at = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
         cache_entry = CacheEntry(
             query=query_request.query,
             response=summary.llm_response,
-            provider=provider,
-            model=model,
+            provider=provider_id,
+            model=model_id,
             started_at=started_at,
             completed_at=completed_at,
             referenced_documents=summary.referenced_documents,
@@ -484,7 +488,9 @@ def is_transcripts_enabled() -> bool:
 def persist_user_conversation_details(
     user_id: str,
     conversation_id: str,
-    model: str,
+    started_at: str,
+    completed_at: str,
+    model_id: str,
     provider_id: str,
     topic_summary: Optional[str],
 ) -> None:
@@ -493,7 +499,9 @@ def persist_user_conversation_details(
     Args:
         user_id: The authenticated user ID
         conversation_id: The conversation ID
-        model: The model identifier
+        started_at: The timestamp when the conversation started
+        completed_at: The timestamp when the conversation completed
+        model_id: The model identifier
         provider_id: The provider identifier
         topic_summary: Optional topic summary for the conversation
     """
@@ -515,7 +523,7 @@ def persist_user_conversation_details(
             conversation = UserConversation(
                 id=normalized_id,
                 user_id=user_id,
-                last_used_model=model,
+                last_used_model=model_id,
                 last_used_provider=provider_id,
                 topic_summary=topic_summary or "",
                 message_count=1,
@@ -525,7 +533,7 @@ def persist_user_conversation_details(
                 "Associated conversation %s to user %s", normalized_id, user_id
             )
         else:
-            existing_conversation.last_used_model = model
+            existing_conversation.last_used_model = model_id
             existing_conversation.last_used_provider = provider_id
             existing_conversation.last_message_at = datetime.now(UTC)
             existing_conversation.message_count += 1
@@ -536,6 +544,27 @@ def persist_user_conversation_details(
                 existing_conversation.message_count,
             )
 
+        max_turn_number = (
+            session.query(func.max(UserTurn.turn_number))
+            .filter_by(conversation_id=normalized_id)
+            .scalar()
+        )
+        turn_number = (max_turn_number or 0) + 1
+        turn = UserTurn(
+            conversation_id=normalized_id,
+            turn_number=turn_number,
+            started_at=datetime.fromisoformat(started_at),
+            completed_at=datetime.fromisoformat(completed_at),
+            provider=provider_id,
+            model=model_id,
+        )
+        session.add(turn)
+        logger.debug(
+            "Created conversation turn - Conversation: %s, Turn: %d",
+            normalized_id,
+            turn_number,
+        )
+
         session.commit()
         logger.debug(
             "Successfully committed conversation %s to database", normalized_id
diff --git a/tests/unit/app/endpoints/test_conversations.py b/tests/unit/app/endpoints/test_conversations.py
diff --git a/tests/unit/utils/test_query.py b/tests/unit/utils/test_query.py
