Merge pull request #32 from agentevals-dev/fix/ordering

Fix conversation element ordering issue with batching

Author: krisztianfekete

src/agentevals/streaming/incremental_processor.py

@@ -174,9 +174,9 @@ def process_log(self, log_event: dict) -> list[dict]:
         event_name = log_event.get("event_name", "")
         body = log_event.get("body", {})
 
-        invocation_id = self.current_invocation_id
+        invocation_id = log_event.get("span_id")
         if not invocation_id:
-            invocation_id = log_event.get("span_id")
+            invocation_id = self.current_invocation_id
         if not invocation_id:
             return updates
 

src/agentevals/streaming/ws_server.py

@@ -261,15 +261,6 @@ async def get_or_create_otlp_session(
         self._active_session_for_name[session_name] = session_id
         self.incremental_extractors[session_id] = IncrementalInvocationExtractor()
 
-        replayed = self._replay_orphan_logs(session)
-        extractor = self.incremental_extractors.get(session_id)
-        if extractor and replayed:
-            for log_event in replayed:
-                updates = extractor.process_log(log_event)
-                for update in updates:
-                    update["sessionId"] = session_id
-                    await self.broadcast_to_ui(update)
-
         await self.broadcast_to_ui(WSSessionStartedEvent(
             session=SessionInfo(
                 session_id=session_id,
@@ -282,6 +273,15 @@ async def get_or_create_otlp_session(
             ),
         ).model_dump(by_alias=True))
 
+        replayed = self._replay_orphan_logs(session)
+        extractor = self.incremental_extractors.get(session_id)
+        if extractor and replayed:
+            for log_event in replayed:
+                updates = extractor.process_log(log_event)
+                for update in updates:
+                    update["sessionId"] = session_id
+                    await self.broadcast_to_ui(update)
+
         logger.info("Auto-created OTLP session: %s (trace: %s)", session_id, trace_id)
         return session
 

ui/src/components/streaming/LiveStreamingView.tsx

@@ -2,7 +2,38 @@ import { useEffect, useRef, useState } from 'react';
 import { useTraceContext } from '../../context/TraceContext';
 import { SessionCard } from './SessionCard';
 import { config } from '../../config';
-import type { LiveSession, StreamingInvocation } from '../../lib/types';
+import type { ConversationElement, LiveSession, StreamingInvocation } from '../../lib/types';
+
+function invocationsToElements(invocations: StreamingInvocation[]): ConversationElement[] {
+  return invocations.flatMap((inv, idx) => {
+    const elements: ConversationElement[] = [];
+    if (inv.userText) {
+      elements.push({
+        type: 'user_input',
+        timestamp: idx * 3,
+        invocationId: inv.invocationId,
+        data: { text: inv.userText },
+      });
+    }
+    for (const tc of inv.toolCalls || []) {
+      elements.push({
+        type: 'tool_call',
+        timestamp: idx * 3 + 1,
+        invocationId: inv.invocationId,
+        data: { toolCall: tc },
+      });
+    }
+    if (inv.agentText) {
+      elements.push({
+        type: 'agent_response',
+        timestamp: idx * 3 + 2,
+        invocationId: inv.invocationId,
+        data: { text: inv.agentText },
+      });
+    }
+    return elements;
+  });
+}
 
 export function LiveStreamingView() {
   const { state, actions } = useTraceContext();
@@ -67,7 +98,9 @@ export function LiveStreamingView() {
               status: s.isComplete ? 'complete' : 'active',
               metadata: (s.metadata ?? {}) as Record<string, string>,
               invocations: s.invocations,
-              liveElements: [],
+              liveElements: s.invocations?.length
+                ? invocationsToElements(s.invocations)
+                : [],
               liveStats: { totalInputTokens: 0, totalOutputTokens: 0 },
               startedAt: s.startedAt,
             });
@@ -272,7 +305,9 @@ export function LiveStreamingView() {
                   status: 'complete',
                   metadata: {},
                   invocations: data.invocations,
-                  liveElements: [],
+                  liveElements: data.invocations?.length
+                    ? invocationsToElements(data.invocations)
+                    : [],
                   liveStats: {
                     totalInputTokens: 0,
                     totalOutputTokens: 0,
@@ -284,6 +319,9 @@ export function LiveStreamingView() {
                   ...session,
                   status: 'complete',
                   invocations: data.invocations,
+                  liveElements: data.invocations?.length
+                    ? invocationsToElements(data.invocations)
+                    : session.liveElements,
                 });
               }