fix: persist workflow execution timeline events to prevent data loss after completion - VoltAgent#647 (VoltAgent#738)

Author: omeraplak

.changeset/thirty-geckos-design.md

@@ -0,0 +1,116 @@
+---
+"@voltagent/postgres": patch
+"@voltagent/supabase": patch
+"@voltagent/libsql": patch
+"@voltagent/core": patch
+---
+
+fix: persist workflow execution timeline events to prevent data loss after completion - #647
+
+## The Problem
+
+When workflows executed, their timeline events (step-start, step-complete, workflow-complete, etc.) were only visible during streaming. Once the workflow completed, the WebSocket state update would replace the execution object without the events field, causing the timeline UI to reset and lose all execution history. Users couldn't see what happened in completed or suspended workflows.
+
+**Symptoms:**
+
+- Timeline showed events during execution
+- Timeline cleared/reset when workflow completed
+- No execution history for completed workflows
+- Events were lost after browser refresh
+
+## The Solution
+
+**Backend (Framework)**:
+
+- Added `events`, `output`, and `cancellation` fields to `WorkflowStateEntry` interface
+- Modified workflow execution to collect all stream events in memory during execution
+- Persist collected events to workflow state when workflow completes, suspends, fails, or is cancelled
+- Updated all storage adapters to support the new fields:
+  - **LibSQL**: Added schema columns + automatic migration method (`addWorkflowStateColumns`)
+  - **Supabase**: Added schema columns + migration detection + ALTER TABLE migration SQL
+  - **Postgres**: Added schema columns + INSERT/UPDATE queries
+  - **In-Memory**: Automatically supported via TypeScript interface
+
+**Frontend (Console)**:
+
+- Updated `WorkflowPlaygroundProvider` to include events when converting `WorkflowStateEntry` → `WorkflowHistoryEntry`
+- Implemented smart merge strategy for WebSocket updates: Use backend persisted events when workflow finishes, keep streaming events during execution
+- Events are now preserved across page refreshes and always visible in timeline UI
+
+## What Gets Persisted
+
+```typescript
+// In WorkflowStateEntry (stored in Memory V2):
+{
+  "events": [
+    {
+      "id": "evt_123",
+      "type": "workflow-start",
+      "name": "Workflow Started",
+      "startTime": "2025-01-24T10:00:00Z",
+      "status": "running",
+      "input": { "userId": "123" }
+    },
+    {
+      "id": "evt_124",
+      "type": "step-complete",
+      "name": "Step: fetch-user",
+      "startTime": "2025-01-24T10:00:01Z",
+      "endTime": "2025-01-24T10:00:02Z",
+      "status": "success",
+      "output": { "user": { "name": "John" } }
+    }
+  ],
+  "output": { "result": "success" },
+  "cancellation": {
+    "cancelledAt": "2025-01-24T10:00:05Z",
+    "reason": "User requested cancellation"
+  }
+}
+```
+
+## Migration Guide
+
+### LibSQL Users
+
+No action required - migrations run automatically on next initialization.
+
+### Supabase Users
+
+When you upgrade and initialize the adapter, you'll see migration SQL in the console. Run it in your Supabase SQL Editor:
+
+```sql
+-- Add workflow event persistence columns
+ALTER TABLE voltagent_workflow_states
+ADD COLUMN IF NOT EXISTS events JSONB;
+
+ALTER TABLE voltagent_workflow_states
+ADD COLUMN IF NOT EXISTS output JSONB;
+
+ALTER TABLE voltagent_workflow_states
+ADD COLUMN IF NOT EXISTS cancellation JSONB;
+```
+
+### Postgres Users
+
+New deployments get the columns automatically. For existing tables, run:
+
+```sql
+ALTER TABLE voltagent_workflow_states
+ADD COLUMN IF NOT EXISTS events JSONB,
+ADD COLUMN IF NOT EXISTS output JSONB,
+ADD COLUMN IF NOT EXISTS cancellation JSONB;
+```
+
+### In-Memory Users
+
+No action required - automatically supported.
+
+## Impact
+
+- ✅ Workflow execution timeline is now persistent and survives completion
+- ✅ Full execution history visible for completed, suspended, and failed workflows
+- ✅ Events, output, and cancellation metadata preserved in database
+- ✅ Console UI timeline works consistently across all workflow states
+- ✅ All storage backends (LibSQL, Supabase, Postgres, In-Memory) behave consistently
+- ✅ No data loss on workflow completion or page refresh

packages/core/src/memory/types.ts

@@ -105,6 +105,30 @@ export interface WorkflowStateEntry {
     };
     suspendData?: any;
   };
+  /**
+   * Stream events collected during execution
+   * Used for timeline visualization in UI
+   */
+  events?: Array<{
+    id: string;
+    type: string;
+    name?: string;
+    from?: string;
+    startTime: string;
+    endTime?: string;
+    status?: string;
+    input?: any;
+    output?: any;
+    metadata?: Record<string, unknown>;
+    context?: Record<string, unknown>;
+  }>;
+  /** Final output of the workflow execution */
+  output?: unknown;
+  /** Cancellation metadata */
+  cancellation?: {
+    cancelledAt: Date;
+    reason?: string;
+  };
   /** User ID if applicable */
   userId?: string;
   /** Conversation ID if applicable */

packages/core/src/workflow/core.ts

@@ -640,6 +640,19 @@ export function createWorkflow<
     executionId: string,
     memory: typeof effectiveMemory,
     logger: Logger,
+    events: Array<{
+      id: string;
+      type: string;
+      name?: string;
+      from?: string;
+      startTime: string;
+      endTime?: string;
+      status?: string;
+      input?: any;
+      output?: any;
+      metadata?: Record<string, unknown>;
+      context?: Record<string, unknown>;
+    }>,
   ): Promise<void> => {
     try {
       logger.trace(`Storing suspension checkpoint for execution ${executionId}`);
@@ -655,6 +668,7 @@ export function createWorkflow<
               suspendData: suspensionData.suspendData,
             }
           : undefined,
+        events,
         updatedAt: new Date(),
       });
       logger.trace(`Successfully stored suspension checkpoint for execution ${executionId}`);
@@ -715,6 +729,58 @@ export function createWorkflow<
     // For normal run, we don't need a stream controller
     const streamController = externalStreamController || null;
 
+    // Collect events during execution for persistence
+    const collectedEvents: Array<{
+      id: string;
+      type: string;
+      name?: string;
+      from?: string;
+      startTime: string;
+      endTime?: string;
+      status?: string;
+      input?: any;
+      output?: any;
+      metadata?: Record<string, unknown>;
+      context?: Record<string, unknown>;
+    }> = [];
+
+    // Helper to emit event and collect for persistence
+    const emitAndCollectEvent = (event: {
+      type: string;
+      executionId: string;
+      from: string;
+      input?: any;
+      output?: any;
+      status: string;
+      context?: any;
+      timestamp: string;
+      stepIndex?: number;
+      stepType?: string;
+      metadata?: Record<string, any>;
+      error?: any;
+    }) => {
+      // Emit to stream if available
+      if (streamController) {
+        streamController.emit(event as any);
+      }
+
+      // Collect for persistence (convert to storage format)
+      const collectedEvent = {
+        id: randomUUID(),
+        type: event.type,
+        name: event.from,
+        from: event.from,
+        startTime: event.timestamp,
+        endTime: event.timestamp, // Will be updated on complete events
+        status: event.status,
+        input: event.input,
+        output: event.output,
+        metadata: event.metadata,
+        context: event.context as Record<string, unknown> | undefined,
+      };
+      collectedEvents.push(collectedEvent);
+    };
+
     // Get observability instance
     const observability = getObservability();
 
@@ -882,7 +948,7 @@ export function createWorkflow<
       };
 
       // Emit workflow start event
-      streamController?.emit({
+      emitAndCollectEvent({
         type: "workflow-start",
         executionId,
         from: name,
@@ -967,7 +1033,7 @@ export function createWorkflow<
               stepData.output = stateManager.state.data;
             }
 
-            streamController?.emit({
+            emitAndCollectEvent({
               type: "step-complete",
               executionId,
               from: stepName,
@@ -991,6 +1057,11 @@ export function createWorkflow<
             try {
               await effectiveMemory.updateWorkflowState(executionId, {
                 status: "cancelled",
+                events: collectedEvents,
+                cancellation: {
+                  cancelledAt: new Date(),
+                  reason,
+                },
                 metadata: {
                   ...(stateManager.state?.usage ? { usage: stateManager.state.usage } : {}),
                   cancellationReason: reason,
@@ -1003,7 +1074,7 @@ export function createWorkflow<
               });
             }
 
-            streamController?.emit({
+            emitAndCollectEvent({
               type: "workflow-cancelled",
               executionId,
               from: name,
@@ -1124,7 +1195,13 @@ export function createWorkflow<
             // Save suspension state to memory
             const suspensionData = stateManager.state.suspension;
             try {
-              await saveSuspensionState(suspensionData, executionId, effectiveMemory, runLogger);
+              await saveSuspensionState(
+                suspensionData,
+                executionId,
+                effectiveMemory,
+                runLogger,
+                collectedEvents,
+              );
             } catch (_) {
               // Error already logged in saveSuspensionState, don't throw
             }
@@ -1191,7 +1268,7 @@ export function createWorkflow<
           executionContext.streamWriter = stepWriter;
 
           // Emit step start event
-          streamController?.emit({
+          emitAndCollectEvent({
             type: "step-start",
             executionId,
             from: step.name || step.id,
@@ -1341,7 +1418,7 @@ export function createWorkflow<
             );
 
             // Emit step complete event
-            streamController?.emit({
+            emitAndCollectEvent({
               type: "step-complete",
               executionId,
               from: stepName,
@@ -1398,7 +1475,7 @@ export function createWorkflow<
               runLogger.debug(`Workflow suspended at step ${index}`, suspensionMetadata);
 
               // Emit suspension event to stream
-              streamController?.emit({
+              emitAndCollectEvent({
                 type: "workflow-suspended",
                 executionId,
                 from: step.name || step.id,
@@ -1437,6 +1514,7 @@ export function createWorkflow<
                   executionId,
                   effectiveMemory,
                   runLogger,
+                  collectedEvents,
                 );
               } catch (_) {
                 // Error already logged in saveSuspensionState, don't throw
@@ -1477,10 +1555,12 @@ export function createWorkflow<
         traceContext.setUsage(stateManager.state.usage);
         traceContext.end("completed");
 
-        // Update Memory V2 state to completed
+        // Update Memory V2 state to completed with events and output
         try {
           await effectiveMemory.updateWorkflowState(executionContext.executionId, {
             status: "completed",
+            events: collectedEvents,
+            output: finalState.result,
             updatedAt: new Date(),
           });
         } catch (memoryError) {
@@ -1502,7 +1582,7 @@ export function createWorkflow<
         );
 
         // Emit workflow complete event
-        streamController?.emit({
+        emitAndCollectEvent({
           type: "workflow-complete",
           executionId,
           from: name,
@@ -1543,7 +1623,7 @@ export function createWorkflow<
 
           workflowRegistry.activeExecutions.delete(executionId);
 
-          streamController?.emit({
+          emitAndCollectEvent({
             type: "workflow-cancelled",
             executionId,
             from: name,
@@ -1626,7 +1706,7 @@ export function createWorkflow<
         );
 
         // Emit workflow error event
-        streamController?.emit({
+        emitAndCollectEvent({
           type: "workflow-error",
           executionId,
           from: name,
@@ -1644,6 +1724,7 @@ export function createWorkflow<
         try {
           await effectiveMemory.updateWorkflowState(executionId, {
             status: "error",
+            events: collectedEvents,
             // Store a lightweight error summary in metadata for debugging
             metadata: {
               ...(stateManager.state?.usage ? { usage: stateManager.state.usage } : {}),