make resume sync

Author: icecrasher321

apps/sim/app/api/resume/[workflowId]/[executionId]/[contextId]/route.ts

@@ -132,14 +132,36 @@ export async function POST(
       workflowId,
     })
 
-    PauseResumeManager.startResumeExecution({
+    const resumeArgs = {
       resumeEntryId: enqueueResult.resumeEntryId,
       resumeExecutionId: enqueueResult.resumeExecutionId,
       pausedExecution: enqueueResult.pausedExecution,
       contextId: enqueueResult.contextId,
       resumeInput: enqueueResult.resumeInput,
       userId: enqueueResult.userId,
-    }).catch((error) => {
+    }
+
+    const isApiCaller = access.auth?.authType === AuthType.API_KEY
+
+    if (isApiCaller) {
+      const result = await PauseResumeManager.startResumeExecution(resumeArgs)
+
+      return NextResponse.json({
+        success: result.success,
+        executionId: enqueueResult.resumeExecutionId,
+        output: result.output,
+        error: result.error,
+        metadata: result.metadata
+          ? {
+              duration: result.metadata.duration,
+              startTime: result.metadata.startTime,
+              endTime: result.metadata.endTime,
+            }
+          : undefined,
+      })
+    }
+
+    PauseResumeManager.startResumeExecution(resumeArgs).catch((error) => {
       logger.error('Failed to start resume execution', {
         workflowId,
         parentExecutionId: executionId,

apps/sim/app/api/workflows/[id]/execute/route.ts

@@ -1400,25 +1400,42 @@ async function handleExecutePost(
             return
           }
 
-          sendEvent({
-            type: 'execution:completed',
-            timestamp: new Date().toISOString(),
-            executionId,
-            workflowId,
-            data: {
-              success: result.success,
-              output: includeFileBase64
-                ? await hydrateUserFilesWithBase64(result.output, {
-                    requestId,
-                    executionId,
-                    maxBytes: base64MaxBytes,
-                  })
-                : result.output,
-              duration: result.metadata?.duration || 0,
-              startTime: result.metadata?.startTime || startTime.toISOString(),
-              endTime: result.metadata?.endTime || new Date().toISOString(),
-            },
-          })
+          const sseOutput = includeFileBase64
+            ? await hydrateUserFilesWithBase64(result.output, {
+                requestId,
+                executionId,
+                maxBytes: base64MaxBytes,
+              })
+            : result.output
+
+          if (result.status === 'paused') {
+            sendEvent({
+              type: 'execution:paused',
+              timestamp: new Date().toISOString(),
+              executionId,
+              workflowId,
+              data: {
+                output: sseOutput,
+                duration: result.metadata?.duration || 0,
+                startTime: result.metadata?.startTime || startTime.toISOString(),
+                endTime: result.metadata?.endTime || new Date().toISOString(),
+              },
+            })
+          } else {
+            sendEvent({
+              type: 'execution:completed',
+              timestamp: new Date().toISOString(),
+              executionId,
+              workflowId,
+              data: {
+                success: result.success,
+                output: sseOutput,
+                duration: result.metadata?.duration || 0,
+                startTime: result.metadata?.startTime || startTime.toISOString(),
+                endTime: result.metadata?.endTime || new Date().toISOString(),
+              },
+            })
+          }
           finalMetaStatus = 'complete'
         } catch (error: unknown) {
           const isTimeout = isTimeoutError(error) || timeoutController.isTimedOut()

apps/sim/hooks/use-execution-stream.ts

@@ -9,6 +9,7 @@ import type {
   ExecutionCompletedData,
   ExecutionErrorData,
   ExecutionEvent,
+  ExecutionPausedData,
   ExecutionStartedData,
   StreamChunkData,
   StreamDoneData,
@@ -74,6 +75,9 @@ export async function processSSEStream(
             case 'execution:completed':
               callbacks.onExecutionCompleted?.(event.data)
               break
+            case 'execution:paused':
+              callbacks.onExecutionPaused?.(event.data)
+              break
             case 'execution:error':
               callbacks.onExecutionError?.(event.data)
               break
@@ -114,6 +118,7 @@ export async function processSSEStream(
 export interface ExecutionStreamCallbacks {
   onExecutionStarted?: (data: ExecutionStartedData) => void
   onExecutionCompleted?: (data: ExecutionCompletedData) => void
+  onExecutionPaused?: (data: ExecutionPausedData) => void
   onExecutionError?: (data: ExecutionErrorData) => void
   onExecutionCancelled?: (data: ExecutionCancelledData) => void
   onBlockStarted?: (data: BlockStartedData) => void

apps/sim/lib/a2a/utils.ts

@@ -264,7 +264,7 @@ export interface ParsedSSEChunk {
   /** Final success flag if this chunk contains the final event */
   finalSuccess?: boolean
   /** Terminal task state if known */
-  terminalState?: 'completed' | 'failed' | 'canceled'
+  terminalState?: 'completed' | 'failed' | 'canceled' | 'input-required'
   /** Final artifacts if present on terminal event */
   finalArtifacts?: Artifact[]
   /** Whether this chunk indicates the stream is done */
@@ -326,6 +326,15 @@ export function parseWorkflowSSEChunk(chunk: string): ParsedSSEChunk {
         result.finalSuccess = parsed.data?.success !== false
         result.terminalState = result.finalSuccess ? 'completed' : 'failed'
         result.isDone = true
+      } else if (parsed.type === 'execution:paused') {
+        if (parsed.data?.output?.content) {
+          result.finalContent = parsed.data.output.content
+        } else if (parsed.data?.output) {
+          result.finalContent = JSON.stringify(parsed.data.output)
+        }
+        result.finalSuccess = true
+        result.terminalState = 'input-required'
+        result.isDone = true
       } else if (parsed.type === 'execution:cancelled') {
         result.finalSuccess = false
         result.terminalState = 'canceled'

apps/sim/lib/workflows/executor/execution-events.ts

@@ -8,6 +8,7 @@ import type { SubflowType } from '@/stores/workflows/workflow/types'
 export type ExecutionEventType =
   | 'execution:started'
   | 'execution:completed'
+  | 'execution:paused'
   | 'execution:error'
   | 'execution:cancelled'
   | 'block:started'
@@ -53,6 +54,20 @@ export interface ExecutionCompletedEvent extends BaseExecutionEvent {
   }
 }
 
+/**
+ * Execution paused event (HITL block waiting for human input)
+ */
+export interface ExecutionPausedEvent extends BaseExecutionEvent {
+  type: 'execution:paused'
+  workflowId: string
+  data: {
+    output: any
+    duration: number
+    startTime: string
+    endTime: string
+  }
+}
+
 /**
  * Execution error event
  */
@@ -196,6 +211,7 @@ export interface StreamDoneEvent extends BaseExecutionEvent {
 export type ExecutionEvent =
   | ExecutionStartedEvent
   | ExecutionCompletedEvent
+  | ExecutionPausedEvent
   | ExecutionErrorEvent
   | ExecutionCancelledEvent
   | BlockStartedEvent
@@ -207,6 +223,7 @@ export type ExecutionEvent =
 
 export type ExecutionStartedData = ExecutionStartedEvent['data']
 export type ExecutionCompletedData = ExecutionCompletedEvent['data']
+export type ExecutionPausedData = ExecutionPausedEvent['data']
 export type ExecutionErrorData = ExecutionErrorEvent['data']
 export type ExecutionCancelledData = ExecutionCancelledEvent['data']
 export type BlockStartedData = BlockStartedEvent['data']

apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts

@@ -292,7 +292,7 @@ export class PauseResumeManager {
     })
   }
 
-  static async startResumeExecution(args: StartResumeExecutionArgs): Promise<void> {
+  static async startResumeExecution(args: StartResumeExecutionArgs): Promise<ExecutionResult> {
     const { resumeEntryId, resumeExecutionId, pausedExecution, contextId, resumeInput, userId } =
       args
 
@@ -357,6 +357,8 @@ export class PauseResumeManager {
       })
 
       await PauseResumeManager.processQueuedResumes(pausedExecution.executionId)
+
+      return result
     } catch (error) {
       await PauseResumeManager.markResumeFailed({
         resumeEntryId,
@@ -995,6 +997,20 @@ export class PauseResumeManager {
           data: { duration: result.metadata?.duration || 0 },
         } as ExecutionEvent)
         finalMetaStatus = 'cancelled'
+      } else if (result.status === 'paused') {
+        writeBufferedEvent({
+          type: 'execution:paused',
+          timestamp: new Date().toISOString(),
+          executionId: resumeExecutionId,
+          workflowId,
+          data: {
+            output: result.output,
+            duration: result.metadata?.duration || 0,
+            startTime: result.metadata?.startTime || new Date().toISOString(),
+            endTime: result.metadata?.endTime || new Date().toISOString(),
+          },
+        } as ExecutionEvent)
+        finalMetaStatus = 'complete'
       } else {
         writeBufferedEvent({
           type: 'execution:completed',

apps/sim/lib/workflows/executor/queued-workflow-execution.ts

@@ -216,6 +216,20 @@ export async function executeQueuedWorkflowJob(
           },
         })
         await setExecutionMeta(executionId, { status: 'cancelled' })
+      } else if (result.status === 'paused') {
+        await eventWriter.write({
+          type: 'execution:paused',
+          timestamp: new Date().toISOString(),
+          executionId,
+          workflowId,
+          data: {
+            output: outputWithBase64,
+            duration: result.metadata?.duration || 0,
+            startTime: result.metadata?.startTime || metadata.startTime,
+            endTime: result.metadata?.endTime || new Date().toISOString(),
+          },
+        })
+        await setExecutionMeta(executionId, { status: 'complete' })
       } else {
         await eventWriter.write({
           type: 'execution:completed',