feat(tarko-agent-server, agent-ui): return initialization events from session create (#1662)

Author: ulivz

multimodal/omni-tars/gui-agent/src/GuiAgentPlugin.ts

@@ -168,7 +168,9 @@ export class GuiAgentPlugin extends AgentPlugin {
   private async emitPresetUserQuey(): Promise<void> {
     const eventStream = this.agent.getEventStream();
     const events = eventStream.getEvents();
-    if (events.length === 0) {
+    // Only emit if no user messages exist yet
+    const hasUserMessage = events.some(event => event.type === 'user_message');
+    if (!hasUserMessage) {
       const event = eventStream.createEvent('user_message', {
         content: `Goto: ${this.agentMode!.link}`,
       });
@@ -179,10 +181,13 @@ export class GuiAgentPlugin extends AgentPlugin {
   private async emitPresetAssistantMessage(): Promise<void> {
     const eventStream = this.agent.getEventStream();
     const events = eventStream.getEvents();
-    guiLogger.debug('emitPresetAssistantMessage events length:', events.length);
-    const event = eventStream.createEvent('assistant_message', {
-      content: `Successfully navigated to ${this.agentMode!.link}, page loaded completely`,
-    });
-    eventStream.sendEvent(event);
+    // Only emit if no assistant messages exist yet
+    const hasAssistantMessage = events.some(event => event.type === 'assistant_message');
+    if (!hasAssistantMessage) {
+      const event = eventStream.createEvent('assistant_message', {
+        content: `Successfully navigated to ${this.agentMode!.link}, page loaded successfully`,
+      });
+      eventStream.sendEvent(event);
+    }
   }
 }

multimodal/tarko/agent-server/src/api/controllers/sessions.ts

@@ -61,26 +61,8 @@ export async function createSession(req: Request, res: Response) {
       }
     }
 
-    // Pass custom AGIO provider, session metadata, and agent options if available
-    const session = new AgentSession(
-      server,
-      sessionId,
-      server.getCustomAgioProvider(),
-      sessionInfo || undefined,
-      agentOptions, // Pass agentOptions for one-time Agent initialization
-    );
-
-    server.sessions[sessionId] = session;
-
-    const { storageUnsubscribe } = await session.initialize();
-
-    // Save unsubscribe function for cleanup
-    if (storageUnsubscribe) {
-      server.storageUnsubscribes[sessionId] = storageUnsubscribe;
-    }
-
     let savedSessionInfo: SessionInfo | undefined;
-    // Store session metadata if we have storage
+    // Store session metadata FIRST if we have storage
     if (server.storageProvider) {
       const now = Date.now();
 
@@ -110,20 +92,59 @@ export async function createSession(req: Request, res: Response) {
       };
 
       savedSessionInfo = await server.storageProvider.createSession(sessionInfo);
+    }
 
-      // If runtime settings were provided and session is active, update the agent configuration
-      if (runtimeSettings && savedSessionInfo) {
-        try {
-          await session.updateSessionConfig(savedSessionInfo);
-          console.log('Session created with runtime settings', { sessionId, runtimeSettings });
-        } catch (error) {
-          console.error('Failed to apply runtime settings to new session', { sessionId, error });
-          // Continue execution - the runtime settings are saved, will apply on next session restart
-        }
+    // Pass custom AGIO provider, session metadata, and agent options if available
+    const session = new AgentSession(
+      server,
+      sessionId,
+      server.getCustomAgioProvider(),
+      savedSessionInfo || undefined,
+      agentOptions, // Pass agentOptions for one-time Agent initialization
+    );
+
+    server.sessions[sessionId] = session;
+
+    const { storageUnsubscribe } = await session.initialize();
+
+    // Save unsubscribe function for cleanup
+    if (storageUnsubscribe) {
+      server.storageUnsubscribes[sessionId] = storageUnsubscribe;
+    }
+
+    // If runtime settings were provided and session is active, update the agent configuration
+    if (runtimeSettings && savedSessionInfo) {
+      try {
+        await session.updateSessionConfig(savedSessionInfo);
+        console.log('Session created with runtime settings', { sessionId, runtimeSettings });
+      } catch (error) {
+        console.error('Failed to apply runtime settings to new session', { sessionId, error });
+        // Continue execution - the runtime settings are saved, will apply on next session restart
+      }
+    }
+
+    // Wait a short time to ensure all initialization events are persisted
+    // This handles the async nature of event storage during agent initialization
+    await session.waitForEventSavesToComplete();
+
+    // Get events that were created during agent initialization
+    let initializationEvents: any[] = [];
+    if (server.storageProvider) {
+      try {
+        initializationEvents = await server.storageProvider.getSessionEvents(sessionId);
+      } catch (error) {
+        console.warn('Failed to retrieve initialization events:', error);
+        // Continue without events - not critical for session creation
       }
     }
 
-    res.status(201).json({ sessionId, session: savedSessionInfo });
+    console.log('Return initializationEvents', initializationEvents);
+
+    res.status(201).json({
+      sessionId,
+      session: savedSessionInfo,
+      events: initializationEvents,
+    });
   } catch (error) {
     console.error('Failed to create session:', error);
     res.status(500).json({ error: 'Failed to create session' });

multimodal/tarko/agent-server/src/core/AgentSession.ts

@@ -71,19 +71,26 @@ export class AgentSession {
   private agioProviderConstructor?: AgioProviderConstructor;
   private sessionInfo?: SessionInfo;
   private storageUnsubscribeMap = new WeakMap<IAgent, () => void>();
+  private pendingEventSaves = new Set<Promise<void>>();
 
   /**
    * Create event handler for storage and AGIO processing
    */
   private createEventHandler() {
     return async (event: AgentEventStream.Event) => {
+      console.log('Receive event', event);
+
       // Save to storage if available and event should be stored
       if (this.server.storageProvider && shouldStoreEvent(event)) {
-        try {
-          await this.server.storageProvider.saveEvent(this.id, event);
-        } catch (error) {
-          console.error(`Failed to save event to storage: ${error}`);
-        }
+        const savePromise = this.server.storageProvider.saveEvent(this.id, event)
+          .catch(error => {
+            console.error(`Failed to save event to storage: ${error}`);
+          })
+          .finally(() => {
+            this.pendingEventSaves.delete(savePromise);
+          });
+        
+        this.pendingEventSaves.add(savePromise);
       }
 
       // Process AGIO events if collector is configured
@@ -97,6 +104,16 @@ export class AgentSession {
     };
   }
 
+  /**
+   * Wait for all pending event saves to complete
+   * This ensures that all events emitted during initialization are persisted before querying storage
+   */
+  async waitForEventSavesToComplete(): Promise<void> {
+    if (this.pendingEventSaves.size > 0) {
+      await Promise.all(Array.from(this.pendingEventSaves));
+    }
+  }
+
   /**
    * Create and initialize a complete agent instance with all wrappers and configuration
    */

multimodal/tarko/agent-ui/src/common/services/apiService.ts

@@ -57,7 +57,7 @@ class ApiService {
   async createSession(
     runtimeSettings?: Record<string, any>,
     agentOptions?: Record<string, any>,
-  ): Promise<SessionInfo> {
+  ): Promise<{ session: SessionInfo; events: AgentEventStream.Event[] }> {
     try {
       const response = await fetch(`${API_BASE_URL}${API_ENDPOINTS.CREATE_SESSION}`, {
         method: 'POST',
@@ -69,8 +69,8 @@ class ApiService {
         throw new Error(`Failed to create session: ${response.statusText}`);
       }
 
-      const { sessionId, session } = await response.json();
-      return session as SessionInfo;
+      const { sessionId, session, events } = await response.json();
+      return { session: session as SessionInfo, events: events || [] };
     } catch (error) {
       console.error('Error creating session:', error);
       throw error;
@@ -588,30 +588,6 @@ class ApiService {
       return { success: false };
     }
   }
-
-  /**
-   * Delete a session
-   */
-  async deleteSession(sessionId: string): Promise<{ success: boolean }> {
-    try {
-      const response = await fetch(`${API_BASE_URL}${API_ENDPOINTS.DELETE_SESSION}`, {
-        method: 'POST',
-        headers: {
-          'Content-Type': 'application/json',
-        },
-        body: JSON.stringify({ sessionId }),
-      });
-
-      if (!response.ok) {
-        throw new Error(`Failed to delete session: ${response.statusText}`);
-      }
-
-      return { success: true };
-    } catch (error) {
-      console.error('Error deleting session:', error);
-      return { success: false };
-    }
-  }
 }
 
 // Export singleton instance

multimodal/tarko/agent-ui/src/common/state/actions/eventProcessors/handlers/SystemHandler.ts

@@ -2,7 +2,7 @@ import { v4 as uuidv4 } from 'uuid';
 import { EventHandler, EventHandlerContext } from '../types';
 import { AgentEventStream, Message } from '@/common/types';
 import { messagesAtom } from '@/common/state/atoms/message';
-import { activePanelContentAtom } from '@/common/state/atoms/ui';
+import { sessionPanelContentAtom } from '@/common/state/atoms/ui';
 import { shouldUpdatePanelContent } from '../utils/panelContentUpdater';
 import { ChatCompletionContentPartImage } from '@tarko/agent-interface';
 
@@ -82,18 +82,25 @@ export class EnvironmentInputHandler
       );
 
       if (imageContent && imageContent.image_url) {
-        const currentPanel = get(activePanelContentAtom);
+        const currentPanelContent = get(sessionPanelContentAtom);
+        const sessionMessages = get(messagesAtom)[sessionId] || [];
+        
+        // Check if this is the first environment_input event in the session
+        const isFirstEnvironmentInput = sessionMessages.filter(msg => msg.role === 'environment').length === 0;
+        const currentSessionPanel = currentPanelContent[sessionId];
 
-        // Only update if current panel is browser_vision_control to maintain context
-        if (currentPanel && currentPanel.type === 'browser_vision_control') {
-          set(activePanelContentAtom, {
-            ...currentPanel,
-            type: 'browser_vision_control',
-            title: currentPanel.title,
-            timestamp: event.timestamp,
-            originalContent: event.content,
-            environmentId: event.id,
-          });
+        // Always show first environment_input (initialization screenshot) or update existing browser_vision_control panel
+        if (isFirstEnvironmentInput || (currentSessionPanel && currentSessionPanel.type === 'browser_vision_control')) {
+          set(sessionPanelContentAtom, (prev) => ({
+            ...prev,
+            [sessionId]: {
+              type: 'browser_vision_control',
+              title: event.description || 'Browser Screenshot',
+              timestamp: event.timestamp,
+              originalContent: event.content,
+              environmentId: event.id,
+            },
+          }));
         }
         // Skip update for other panel types to avoid duplicate Browser Screenshot rendering
       }

multimodal/tarko/agent-ui/src/common/state/actions/sessionActions.ts

@@ -83,7 +83,7 @@ export const loadSessionsAction = atom(null, async (get, set) => {
 
 export const createSessionAction = atom(null, async (get, set, runtimeSettings?: Record<string, any>, agentOptions?: Record<string, any>) => {
   try {
-    const newSession = await apiService.createSession(runtimeSettings, agentOptions);
+    const { session: newSession, events: initializationEvents } = await apiService.createSession(runtimeSettings, agentOptions);
 
     set(sessionsAtom, (prev) => [newSession, ...prev]);
 
@@ -107,6 +107,17 @@ export const createSessionAction = atom(null, async (get, set, runtimeSettings?:
     }));
     set(activeSessionIdAtom, newSession.id);
 
+    // Process initialization events if any were returned
+    if (initializationEvents && initializationEvents.length > 0) {
+      console.log(`Processing ${initializationEvents.length} initialization events for session ${newSession.id}`);
+      
+      const processedEvents = preprocessStreamingEvents(initializationEvents);
+      
+      for (const event of processedEvents) {
+        await set(processEventAction, { sessionId: newSession.id, event });
+      }
+    }
+
     return newSession.id;
   } catch (error) {
     console.error('Failed to create session:', error);

multimodal/tarko/agent-ui/src/standalone/home/CreatingPage.tsx

@@ -1,4 +1,4 @@
-import React, { useEffect, useState } from 'react';
+import React, { useEffect, useState, useRef } from 'react';
 import { useLocation, useNavigate, useSearchParams } from 'react-router-dom';
 import { useAtomValue, useSetAtom } from 'jotai';
 import { useSession } from '@/common/hooks/useSession';
@@ -26,8 +26,15 @@ const CreatingPage: React.FC = () => {
   const resetGlobalSettings = useSetAtom(resetGlobalRuntimeSettingsAction);
   const createSession = useSetAtom(createSessionAction);
   const [isCreating, setIsCreating] = useState(true);
+  const hasExecuted = useRef(false);
 
   useEffect(() => {
+    // Prevent double execution
+    if (hasExecuted.current) {
+      return;
+    }
+    hasExecuted.current = true;
+
     const createSessionWithOptions = async () => {
       try {
         // Get parameters from multiple sources (priority order):
@@ -105,7 +112,7 @@ const CreatingPage: React.FC = () => {
     };
 
     createSessionWithOptions();
-  }, [location.state, searchParams, globalSettings, resetGlobalSettings, navigate, sendMessage, createSession]);
+  }, []); // Empty dependency array since we use useRef to prevent double execution
 
   return (
     <div className="h-full flex items-center justify-center">