cleanup streaming server

icecrasher321 · waleedlatif1 · commit 544780c0f693 · 2025-12-22T14:48:23.000-08:00
diff --git a/apps/sim/app/api/chat/[identifier]/route.test.ts b/apps/sim/app/api/chat/[identifier]/route.test.ts
@@ -70,19 +70,6 @@ vi.mock('@/lib/core/utils/request', () => ({
   generateRequestId: vi.fn().mockReturnValue('test-request-id'),
 }))
 
-vi.mock('@/app/api/workflows/[id]/execute/route', () => ({
-  createFilteredResult: vi.fn().mockImplementation((result: any) => ({
-    ...result,
-    logs: undefined,
-    metadata: result.metadata
-      ? {
-          ...result.metadata,
-          workflowConnections: undefined,
-        }
-      : undefined,
-  })),
-}))
-
 describe('Chat Identifier API Route', () => {
   const mockAddCorsHeaders = vi.fn().mockImplementation((response) => response)
   const mockValidateChatAuth = vi.fn().mockResolvedValue({ authorized: true })
diff --git a/apps/sim/app/api/chat/[identifier]/route.ts b/apps/sim/app/api/chat/[identifier]/route.ts
@@ -206,7 +206,6 @@ export async function POST(
 
       const { createStreamingResponse } = await import('@/lib/workflows/streaming/streaming')
       const { SSE_HEADERS } = await import('@/lib/core/utils/sse')
-      const { createFilteredResult } = await import('@/app/api/workflows/[id]/execute/route')
 
       const workflowInput: any = { input, conversationId }
       if (files && Array.isArray(files) && files.length > 0) {
@@ -267,7 +266,6 @@ export async function POST(
           isSecureMode: true,
           workflowTriggerType: 'chat',
         },
-        createFilteredResult,
         executionId,
       })
 
diff --git a/apps/sim/app/api/workflows/[id]/execute/route.ts b/apps/sim/app/api/workflows/[id]/execute/route.ts
@@ -49,126 +49,6 @@ const ExecuteWorkflowSchema = z.object({
 export const runtime = 'nodejs'
 export const dynamic = 'force-dynamic'
 
-/**
- * Execute workflow with streaming support - used by chat and other streaming endpoints
- *
- * This function assumes preprocessing has already been completed.
- * Callers must run preprocessExecution() first to validate workflow, check usage limits,
- * and resolve actor before calling this function.
- *
- * This is a wrapper function that:
- * - Supports streaming callbacks (onStream, onBlockComplete)
- * - Returns ExecutionResult instead of NextResponse
- * - Handles pause/resume logic
- *
- * Used by:
- * - Chat execution (/api/chat/[identifier]/route.ts)
- * - Streaming responses (lib/workflows/streaming.ts)
- */
-export async function executeWorkflow(
-  workflow: any,
-  requestId: string,
-  input: any | undefined,
-  actorUserId: string,
-  streamConfig?: {
-    enabled: boolean
-    selectedOutputs?: string[]
-    isSecureMode?: boolean
-    workflowTriggerType?: 'api' | 'chat'
-    onStream?: (streamingExec: any) => Promise<void>
-    onBlockComplete?: (blockId: string, output: any) => Promise<void>
-    skipLoggingComplete?: boolean
-  },
-  providedExecutionId?: string
-): Promise<any> {
-  const workflowId = workflow.id
-  const executionId = providedExecutionId || uuidv4()
-  const triggerType = streamConfig?.workflowTriggerType || 'api'
-  const loggingSession = new LoggingSession(workflowId, executionId, triggerType, requestId)
-
-  try {
-    const metadata: ExecutionMetadata = {
-      requestId,
-      executionId,
-      workflowId,
-      workspaceId: workflow.workspaceId,
-      userId: actorUserId,
-      workflowUserId: workflow.userId,
-      triggerType,
-      useDraftState: false,
-      startTime: new Date().toISOString(),
-      isClientSession: false,
-    }
-
-    const snapshot = new ExecutionSnapshot(
-      metadata,
-      workflow,
-      input,
-      workflow.variables || {},
-      streamConfig?.selectedOutputs || []
-    )
-
-    const result = await executeWorkflowCore({
-      snapshot,
-      callbacks: {
-        onStream: streamConfig?.onStream,
-        onBlockComplete: streamConfig?.onBlockComplete
-          ? async (blockId: string, _blockName: string, _blockType: string, output: any) => {
-              await streamConfig.onBlockComplete!(blockId, output)
-            }
-          : undefined,
-      },
-      loggingSession,
-    })
-
-    if (result.status === 'paused') {
-      if (!result.snapshotSeed) {
-        logger.error(`[${requestId}] Missing snapshot seed for paused execution`, {
-          executionId,
-        })
-      } else {
-        await PauseResumeManager.persistPauseResult({
-          workflowId,
-          executionId,
-          pausePoints: result.pausePoints || [],
-          snapshotSeed: result.snapshotSeed,
-          executorUserId: result.metadata?.userId,
-        })
-      }
-    } else {
-      await PauseResumeManager.processQueuedResumes(executionId)
-    }
-
-    if (streamConfig?.skipLoggingComplete) {
-      return {
-        ...result,
-        _streamingMetadata: {
-          loggingSession,
-          processedInput: input,
-        },
-      }
-    }
-
-    return result
-  } catch (error: any) {
-    logger.error(`[${requestId}] Workflow execution failed:`, error)
-    throw error
-  }
-}
-
-export function createFilteredResult(result: any) {
-  return {
-    ...result,
-    logs: undefined,
-    metadata: result.metadata
-      ? {
-          ...result.metadata,
-          workflowConnections: undefined,
-        }
-      : undefined,
-  }
-}
-
 function resolveOutputIds(
   selectedOutputs: string[] | undefined,
   blocks: Record<string, any>
@@ -606,7 +486,6 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
           isSecureMode: false,
           workflowTriggerType: triggerType === 'chat' ? 'chat' : 'api',
         },
-        createFilteredResult,
         executionId,
       })
 
diff --git a/apps/sim/lib/workflows/executor/execute-workflow.ts b/apps/sim/lib/workflows/executor/execute-workflow.ts
@@ -0,0 +1,114 @@
+import { v4 as uuidv4 } from 'uuid'
+import { createLogger } from '@/lib/logs/console/logger'
+import { LoggingSession } from '@/lib/logs/execution/logging-session'
+import { executeWorkflowCore } from '@/lib/workflows/executor/execution-core'
+import { PauseResumeManager } from '@/lib/workflows/executor/human-in-the-loop-manager'
+import { type ExecutionMetadata, ExecutionSnapshot } from '@/executor/execution/snapshot'
+
+const logger = createLogger('WorkflowExecution')
+
+export interface ExecuteWorkflowOptions {
+  enabled: boolean
+  selectedOutputs?: string[]
+  isSecureMode?: boolean
+  workflowTriggerType?: 'api' | 'chat'
+  onStream?: (streamingExec: any) => Promise<void>
+  onBlockComplete?: (blockId: string, output: any) => Promise<void>
+  skipLoggingComplete?: boolean
+}
+
+export interface WorkflowInfo {
+  id: string
+  userId: string
+  workspaceId?: string | null
+  isDeployed?: boolean
+  variables?: Record<string, any>
+}
+
+export async function executeWorkflow(
+  workflow: WorkflowInfo,
+  requestId: string,
+  input: any | undefined,
+  actorUserId: string,
+  streamConfig?: ExecuteWorkflowOptions,
+  providedExecutionId?: string
+): Promise<any> {
+  if (!workflow.workspaceId) {
+    throw new Error(`Workflow ${workflow.id} has no workspaceId`)
+  }
+
+  const workflowId = workflow.id
+  const workspaceId = workflow.workspaceId
+  const executionId = providedExecutionId || uuidv4()
+  const triggerType = streamConfig?.workflowTriggerType || 'api'
+  const loggingSession = new LoggingSession(workflowId, executionId, triggerType, requestId)
+
+  try {
+    const metadata: ExecutionMetadata = {
+      requestId,
+      executionId,
+      workflowId,
+      workspaceId,
+      userId: actorUserId,
+      workflowUserId: workflow.userId,
+      triggerType,
+      useDraftState: false,
+      startTime: new Date().toISOString(),
+      isClientSession: false,
+    }
+
+    const snapshot = new ExecutionSnapshot(
+      metadata,
+      workflow,
+      input,
+      workflow.variables || {},
+      streamConfig?.selectedOutputs || []
+    )
+
+    const result = await executeWorkflowCore({
+      snapshot,
+      callbacks: {
+        onStream: streamConfig?.onStream,
+        onBlockComplete: streamConfig?.onBlockComplete
+          ? async (blockId: string, _blockName: string, _blockType: string, output: any) => {
+              await streamConfig.onBlockComplete!(blockId, output)
+            }
+          : undefined,
+      },
+      loggingSession,
+    })
+
+    if (result.status === 'paused') {
+      if (!result.snapshotSeed) {
+        logger.error(`[${requestId}] Missing snapshot seed for paused execution`, {
+          executionId,
+        })
+      } else {
+        await PauseResumeManager.persistPauseResult({
+          workflowId,
+          executionId,
+          pausePoints: result.pausePoints || [],
+          snapshotSeed: result.snapshotSeed,
+          executorUserId: result.metadata?.userId,
+        })
+      }
+    } else {
+      await PauseResumeManager.processQueuedResumes(executionId)
+    }
+
+    if (streamConfig?.skipLoggingComplete) {
+      return {
+        ...result,
+        _streamingMetadata: {
+          loggingSession,
+          processedInput: input,
+        },
+      }
+    }
+
+    return result
+  } catch (error: any) {
+    logger.error(`[${requestId}] Workflow execution failed:`, error)
+    throw error
+  }
+}
diff --git a/apps/sim/lib/workflows/streaming/streaming.ts b/apps/sim/lib/workflows/streaming/streaming.ts
