fix(backend): conversation statistics are currently not being persisted to the database (V1). (#11837)

openhands/app_server/app_conversation/app_conversation_info_service.py

@@ -9,6 +9,7 @@ from openhands.app_server.app_conversation.app_conversation_models import (
     AppConversationSortOrder,
 )
 from openhands.app_server.services.injector import Injector
+from openhands.sdk.event import ConversationStateUpdateEvent
 from openhands.sdk.utils.models import DiscriminatedUnionMixin
 
 
@@ -92,6 +93,19 @@ class AppConversationInfoService(ABC):
         Return the stored info
         """
 
+    @abstractmethod
+    async def process_stats_event(
+        self,
+        event: ConversationStateUpdateEvent,
+        conversation_id: UUID,
+    ) -> None:
+        """Process a stats event and update conversation statistics.
+
+        Args:
+            event: The ConversationStateUpdateEvent with key='stats'
+            conversation_id: The ID of the conversation to update
+        """
+
 
 class AppConversationInfoServiceInjector(
     DiscriminatedUnionMixin, Injector[AppConversationInfoService], ABC

openhands/app_server/app_conversation/sql_app_conversation_info_service.py

@@ -45,6 +45,8 @@ from openhands.app_server.utils.sql_utils import (
     create_json_type_decorator,
 )
 from openhands.integrations.provider import ProviderType
+from openhands.sdk.conversation.conversation_stats import ConversationStats
+from openhands.sdk.event import ConversationStateUpdateEvent
 from openhands.sdk.llm import MetricsSnapshot
 from openhands.sdk.llm.utils.metrics import TokenUsage
 from openhands.storage.data_models.conversation_metadata import ConversationTrigger
@@ -354,6 +356,130 @@ class SQLAppConversationInfoService(AppConversationInfoService):
         await self.db_session.commit()
         return info
 
+    async def update_conversation_statistics(
+        self, conversation_id: UUID, stats: ConversationStats
+    ) -> None:
+        """Update conversation statistics from stats event data.
+
+        Args:
+            conversation_id: The ID of the conversation to update
+            stats: ConversationStats object containing usage_to_metrics data from stats event
+        """
+        # Extract agent metrics from usage_to_metrics
+        usage_to_metrics = stats.usage_to_metrics
+        agent_metrics = usage_to_metrics.get('agent')
+
+        if not agent_metrics:
+            logger.debug(
+                'No agent metrics found in stats for conversation %s', conversation_id
+            )
+            return
+
+        # Query existing record using secure select (filters for V1 and user if available)
+        query = await self._secure_select()
+        query = query.where(
+            StoredConversationMetadata.conversation_id == str(conversation_id)
+        )
+        result = await self.db_session.execute(query)
+        stored = result.scalar_one_or_none()
+
+        if not stored:
+            logger.debug(
+                'Conversation %s not found or not accessible, skipping statistics update',
+                conversation_id,
+            )
+            return
+
+        # Extract accumulated_cost and max_budget_per_task from Metrics object
+        accumulated_cost = agent_metrics.accumulated_cost
+        max_budget_per_task = agent_metrics.max_budget_per_task
+
+        # Extract accumulated_token_usage from Metrics object
+        accumulated_token_usage = agent_metrics.accumulated_token_usage
+        if accumulated_token_usage:
+            prompt_tokens = accumulated_token_usage.prompt_tokens
+            completion_tokens = accumulated_token_usage.completion_tokens
+            cache_read_tokens = accumulated_token_usage.cache_read_tokens
+            cache_write_tokens = accumulated_token_usage.cache_write_tokens
+            reasoning_tokens = accumulated_token_usage.reasoning_tokens
+            context_window = accumulated_token_usage.context_window
+            per_turn_token = accumulated_token_usage.per_turn_token
+        else:
+            prompt_tokens = None
+            completion_tokens = None
+            cache_read_tokens = None
+            cache_write_tokens = None
+            reasoning_tokens = None
+            context_window = None
+            per_turn_token = None
+
+        # Update fields only if values are provided (not None)
+        if accumulated_cost is not None:
+            stored.accumulated_cost = accumulated_cost
+        if max_budget_per_task is not None:
+            stored.max_budget_per_task = max_budget_per_task
+        if prompt_tokens is not None:
+            stored.prompt_tokens = prompt_tokens
+        if completion_tokens is not None:
+            stored.completion_tokens = completion_tokens
+        if cache_read_tokens is not None:
+            stored.cache_read_tokens = cache_read_tokens
+        if cache_write_tokens is not None:
+            stored.cache_write_tokens = cache_write_tokens
+        if reasoning_tokens is not None:
+            stored.reasoning_tokens = reasoning_tokens
+        if context_window is not None:
+            stored.context_window = context_window
+        if per_turn_token is not None:
+            stored.per_turn_token = per_turn_token
+
+        # Update last_updated_at timestamp
+        stored.last_updated_at = utc_now()
+
+        await self.db_session.commit()
+
+    async def process_stats_event(
+        self,
+        event: ConversationStateUpdateEvent,
+        conversation_id: UUID,
+    ) -> None:
+        """Process a stats event and update conversation statistics.
+
+        Args:
+            event: The ConversationStateUpdateEvent with key='stats'
+            conversation_id: The ID of the conversation to update
+        """
+        try:
+            # Parse event value into ConversationStats model for type safety
+            # event.value can be a dict (from JSON deserialization) or a ConversationStats object
+            event_value = event.value
+            conversation_stats: ConversationStats | None = None
+
+            if isinstance(event_value, ConversationStats):
+                # Already a ConversationStats object
+                conversation_stats = event_value
+            elif isinstance(event_value, dict):
+                # Parse dict into ConversationStats model
+                # This validates the structure and ensures type safety
+                conversation_stats = ConversationStats.model_validate(event_value)
+            elif hasattr(event_value, 'usage_to_metrics'):
+                # Handle objects with usage_to_metrics attribute (e.g., from tests)
+                # Convert to dict first, then validate
+                stats_dict = {'usage_to_metrics': event_value.usage_to_metrics}
+                conversation_stats = ConversationStats.model_validate(stats_dict)
+
+            if conversation_stats and conversation_stats.usage_to_metrics:
+                # Pass ConversationStats object directly for type safety
+                await self.update_conversation_statistics(
+                    conversation_id, conversation_stats
+                )
+        except Exception:
+            logger.exception(
+                'Error updating conversation statistics for conversation %s',
+                conversation_id,
+                stack_info=True,
+            )
+
     async def _secure_select(self):
         query = select(StoredConversationMetadata).where(
             StoredConversationMetadata.conversation_version == 'V1'

openhands/app_server/event_callback/webhook_router.py

@@ -43,6 +43,7 @@ from openhands.app_server.user.specifiy_user_context import (
 from openhands.app_server.user.user_context import UserContext
 from openhands.integrations.provider import ProviderType
 from openhands.sdk import Event
+from openhands.sdk.event import ConversationStateUpdateEvent
 from openhands.server.user_auth.default_user_auth import DefaultUserAuth
 from openhands.server.user_auth.user_auth import (
     get_for_user as get_user_auth_for_user,
@@ -144,6 +145,13 @@ async def on_event(
             *[event_service.save_event(conversation_id, event) for event in events]
         )
 
+        # Process stats events for V1 conversations
+        for event in events:
+            if isinstance(event, ConversationStateUpdateEvent) and event.key == 'stats':
+                await app_conversation_info_service.process_stats_event(
+                    event, conversation_id
+                )
+
         asyncio.create_task(
             _run_callbacks_in_bg_and_close(
                 conversation_id, app_conversation_info.created_by_user_id, events