fix(copilot): fix execute workflow from diff store (#1894)

Sg312 · web-flow · commit c86f2a053703 · 2025-11-11T11:09:17.000-08:00
* Fix run from diff store

* Fix copilot run workflow
diff --git a/apps/sim/app/api/workflows/[id]/execute/route.ts b/apps/sim/app/api/workflows/[id]/execute/route.ts
@@ -30,6 +30,15 @@ const ExecuteWorkflowSchema = z.object({
   useDraftState: z.boolean().optional(),
   input: z.any().optional(),
   startBlockId: z.string().optional(),
+  // Optional workflow state override (for executing diff workflows)
+  workflowStateOverride: z
+    .object({
+      blocks: z.record(z.any()),
+      edges: z.array(z.any()),
+      loops: z.record(z.any()).optional(),
+      parallels: z.record(z.any()).optional(),
+    })
+    .optional(),
 })
 
 export const runtime = 'nodejs'
@@ -310,14 +319,22 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
       stream: streamParam,
       useDraftState,
       input: validatedInput,
+      workflowStateOverride,
     } = validation.data
 
     // For API key auth, the entire body is the input (except for our control fields)
     // For session auth, the input is explicitly provided in the input field
     const input =
       auth.authType === 'api_key'
         ? (() => {
-            const { selectedOutputs, triggerType, stream, useDraftState, ...rest } = body
+            const {
+              selectedOutputs,
+              triggerType,
+              stream,
+              useDraftState,
+              workflowStateOverride,
+              ...rest
+            } = body
             return Object.keys(rest).length > 0 ? rest : validatedInput
           })()
         : validatedInput
@@ -460,6 +477,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
           triggerType,
           useDraftState: shouldUseDraftState,
           startTime: new Date().toISOString(),
+          workflowStateOverride,
         }
 
         const snapshot = new ExecutionSnapshot(
@@ -714,6 +732,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
             triggerType,
             useDraftState: shouldUseDraftState,
             startTime: new Date().toISOString(),
+            workflowStateOverride,
           }
 
           const snapshot = new ExecutionSnapshot(
diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-workflow-execution.ts b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/hooks/use-workflow-execution.ts
@@ -1,5 +1,6 @@
 import { useCallback, useState } from 'react'
 import { v4 as uuidv4 } from 'uuid'
+import { shallow } from 'zustand/shallow'
 import { createLogger } from '@/lib/logs/console/logger'
 import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
 import { processStreamingBlockLogs } from '@/lib/tokenization'
@@ -16,6 +17,7 @@ import { useExecutionStore } from '@/stores/execution/store'
 import { useVariablesStore } from '@/stores/panel/variables/store'
 import { useEnvironmentStore } from '@/stores/settings/environment/store'
 import { useTerminalConsoleStore } from '@/stores/terminal'
+import { useWorkflowDiffStore } from '@/stores/workflow-diff'
 import { useWorkflowRegistry } from '@/stores/workflows/registry/store'
 import { mergeSubblockState } from '@/stores/workflows/utils'
 import { useCurrentWorkflow } from './use-current-workflow'
@@ -99,6 +101,26 @@ export function useWorkflowExecution() {
   } = useExecutionStore()
   const [executionResult, setExecutionResult] = useState<ExecutionResult | null>(null)
   const executionStream = useExecutionStream()
+  const {
+    diffWorkflow: executionDiffWorkflow,
+    isDiffReady: isDiffWorkflowReady,
+    isShowingDiff: isViewingDiff,
+  } = useWorkflowDiffStore(
+    useCallback(
+      (state) => ({
+        diffWorkflow: state.diffWorkflow,
+        isDiffReady: state.isDiffReady,
+        isShowingDiff: state.isShowingDiff,
+      }),
+      []
+    ),
+    shallow
+  )
+  const hasActiveDiffWorkflow =
+    isDiffWorkflowReady &&
+    isViewingDiff &&
+    !!executionDiffWorkflow &&
+    Object.keys(executionDiffWorkflow.blocks || {}).length > 0
 
   /**
    * Validates debug state before performing debug operations
@@ -645,8 +667,14 @@ export function useWorkflowExecution() {
     onBlockComplete?: (blockId: string, output: any) => Promise<void>,
     overrideTriggerType?: 'chat' | 'manual' | 'api'
   ): Promise<ExecutionResult | StreamingExecution> => {
-    // Use currentWorkflow but check if we're in diff mode
-    const { blocks: workflowBlocks, edges: workflowEdges } = currentWorkflow
+    // Use diff workflow for execution when available, regardless of canvas view state
+    const executionWorkflowState =
+      hasActiveDiffWorkflow && executionDiffWorkflow ? executionDiffWorkflow : null
+    const usingDiffForExecution = executionWorkflowState !== null
+    const workflowBlocks = (executionWorkflowState?.blocks ??
+      currentWorkflow.blocks) as typeof currentWorkflow.blocks
+    const workflowEdges = (executionWorkflowState?.edges ??
+      currentWorkflow.edges) as typeof currentWorkflow.edges
 
     // Filter out blocks without type (these are layout-only blocks)
     const validBlocks = Object.entries(workflowBlocks).reduce(
@@ -665,6 +693,9 @@ export function useWorkflowExecution() {
 
     logger.info('Executing workflow', {
       isDiffMode: currentWorkflow.isDiffMode,
+      usingDiffForExecution,
+      isViewingDiff,
+      executingDiffWorkflow: usingDiffForExecution && isViewingDiff,
       isExecutingFromChat,
       totalBlocksCount: Object.keys(workflowBlocks).length,
       validBlocksCount: Object.keys(validBlocks).length,
@@ -838,6 +869,15 @@ export function useWorkflowExecution() {
           selectedOutputs,
           triggerType: overrideTriggerType || 'manual',
           useDraftState: true,
+          // Pass diff workflow state if available for execution
+          workflowStateOverride: executionWorkflowState
+            ? {
+                blocks: executionWorkflowState.blocks,
+                edges: executionWorkflowState.edges,
+                loops: executionWorkflowState.loops,
+                parallels: executionWorkflowState.parallels,
+              }
+            : undefined,
           callbacks: {
             onExecutionStarted: (data) => {
               logger.info('Server execution started:', data)
diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.ts b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.ts
@@ -1,4 +1,7 @@
+import { v4 as uuidv4 } from 'uuid'
 import type { ExecutionResult, StreamingExecution } from '@/executor/types'
+import { useTerminalConsoleStore } from '@/stores/terminal'
+import { useWorkflowDiffStore } from '@/stores/workflow-diff/store'
 import { useWorkflowRegistry } from '@/stores/workflows/registry/store'
 
 export interface WorkflowExecutionOptions {
@@ -11,7 +14,7 @@ export interface WorkflowExecutionOptions {
 
 /**
  * Execute workflow with full logging (used by copilot tools)
- * This now delegates to the server-side executor via API
+ * Handles SSE streaming and populates console logs in real-time
  */
 export async function executeWorkflowWithFullLogging(
   options: WorkflowExecutionOptions = {}
@@ -22,25 +25,148 @@ export async function executeWorkflowWithFullLogging(
     throw new Error('No active workflow')
   }
 
-  // For copilot tool calls, we use non-SSE execution to get a simple result
+  // Check if there's an active diff workflow to execute
+  const { diffWorkflow, isDiffReady, isShowingDiff } = useWorkflowDiffStore.getState()
+  const hasActiveDiffWorkflow =
+    isDiffReady &&
+    isShowingDiff &&
+    !!diffWorkflow &&
+    Object.keys(diffWorkflow.blocks || {}).length > 0
+
+  const executionId = options.executionId || uuidv4()
+  const { addConsole } = useTerminalConsoleStore.getState()
+
+  // Build request payload
+  const payload: any = {
+    input: options.workflowInput,
+    stream: true,
+    triggerType: options.overrideTriggerType || 'manual',
+    useDraftState: true,
+  }
+
+  // Add diff workflow override if active
+  if (hasActiveDiffWorkflow) {
+    payload.workflowStateOverride = {
+      blocks: diffWorkflow.blocks,
+      edges: diffWorkflow.edges,
+      loops: diffWorkflow.loops,
+      parallels: diffWorkflow.parallels,
+    }
+  }
+
   const response = await fetch(`/api/workflows/${activeWorkflowId}/execute`, {
     method: 'POST',
     headers: {
       'Content-Type': 'application/json',
     },
-    body: JSON.stringify({
-      input: options.workflowInput,
-      stream: false, // Copilot doesn't need SSE streaming
-      triggerType: options.overrideTriggerType || 'manual',
-      useDraftState: true,
-    }),
+    body: JSON.stringify(payload),
   })
 
   if (!response.ok) {
     const error = await response.json()
     throw new Error(error.error || 'Workflow execution failed')
   }
 
-  const result = await response.json()
-  return result as ExecutionResult
+  if (!response.body) {
+    throw new Error('No response body')
+  }
+
+  // Parse SSE stream
+  const reader = response.body.getReader()
+  const decoder = new TextDecoder()
+  let buffer = ''
+  let executionResult: ExecutionResult = {
+    success: false,
+    output: {},
+    logs: [],
+  }
+
+  try {
+    while (true) {
+      const { done, value } = await reader.read()
+      if (done) break
+
+      buffer += decoder.decode(value, { stream: true })
+      const lines = buffer.split('\n\n')
+      buffer = lines.pop() || ''
+
+      for (const line of lines) {
+        if (!line.trim() || !line.startsWith('data: ')) continue
+
+        const data = line.substring(6).trim()
+        if (data === '[DONE]') continue
+
+        try {
+          const event = JSON.parse(data)
+
+          switch (event.type) {
+            case 'block:completed':
+              addConsole({
+                input: event.data.input || {},
+                output: event.data.output,
+                success: true,
+                durationMs: event.data.durationMs,
+                startedAt: new Date(Date.now() - event.data.durationMs).toISOString(),
+                endedAt: new Date().toISOString(),
+                workflowId: activeWorkflowId,
+                blockId: event.data.blockId,
+                executionId,
+                blockName: event.data.blockName,
+                blockType: event.data.blockType,
+                iterationCurrent: event.data.iterationCurrent,
+                iterationTotal: event.data.iterationTotal,
+                iterationType: event.data.iterationType,
+              })
+
+              if (options.onBlockComplete) {
+                options.onBlockComplete(event.data.blockId, event.data.output).catch(() => {})
+              }
+              break
+
+            case 'block:error':
+              addConsole({
+                input: event.data.input || {},
+                output: {},
+                success: false,
+                error: event.data.error,
+                durationMs: event.data.durationMs,
+                startedAt: new Date(Date.now() - event.data.durationMs).toISOString(),
+                endedAt: new Date().toISOString(),
+                workflowId: activeWorkflowId,
+                blockId: event.data.blockId,
+                executionId,
+                blockName: event.data.blockName,
+                blockType: event.data.blockType,
+                iterationCurrent: event.data.iterationCurrent,
+                iterationTotal: event.data.iterationTotal,
+                iterationType: event.data.iterationType,
+              })
+              break
+
+            case 'execution:completed':
+              executionResult = {
+                success: event.data.success,
+                output: event.data.output,
+                logs: [],
+                metadata: {
+                  duration: event.data.duration,
+                  startTime: event.data.startTime,
+                  endTime: event.data.endTime,
+                },
+              }
+              break
+
+            case 'execution:error':
+              throw new Error(event.data.error || 'Execution failed')
+          }
+        } catch (parseError) {
+          // Skip malformed SSE events
+        }
+      }
+    }
+  } finally {
+    reader.releaseLock()
+  }
+
+  return executionResult
 }
diff --git a/apps/sim/executor/execution/snapshot.ts b/apps/sim/executor/execution/snapshot.ts
@@ -13,6 +13,12 @@ export interface ExecutionMetadata {
   startTime: string
   pendingBlocks?: string[]
   resumeFromSnapshot?: boolean
+  workflowStateOverride?: {
+    blocks: Record<string, any>
+    edges: Edge[]
+    loops?: Record<string, any>
+    parallels?: Record<string, any>
+  }
 }
 
 export interface ExecutionCallbacks {
diff --git a/apps/sim/hooks/use-execution-stream.ts b/apps/sim/hooks/use-execution-stream.ts
@@ -61,6 +61,12 @@ export interface ExecuteStreamOptions {
   startBlockId?: string
   triggerType?: string
   useDraftState?: boolean
+  workflowStateOverride?: {
+    blocks: Record<string, any>
+    edges: any[]
+    loops?: Record<string, any>
+    parallels?: Record<string, any>
+  }
   callbacks?: ExecutionStreamCallbacks
 }
 
diff --git a/apps/sim/lib/workflows/executor/execution-core.ts b/apps/sim/lib/workflows/executor/execution-core.ts
@@ -114,7 +114,18 @@ export async function executeWorkflowCore(
     let loops
     let parallels
 
-    if (useDraftState) {
+    // Use workflowStateOverride if provided (for diff workflows)
+    if (metadata.workflowStateOverride) {
+      blocks = metadata.workflowStateOverride.blocks
+      edges = metadata.workflowStateOverride.edges
+      loops = metadata.workflowStateOverride.loops || {}
+      parallels = metadata.workflowStateOverride.parallels || {}
+
+      logger.info(`[${requestId}] Using workflow state override (diff workflow execution)`, {
+        blocksCount: Object.keys(blocks).length,
+        edgesCount: edges.length,
+      })
+    } else if (useDraftState) {
       const draftData = await loadWorkflowFromNormalizedTables(workflowId)
 
       if (!draftData) {
