feat(hitl): add human in the loop block (#1832)

* fix(billing): should allow restoring subscription (#1728)

* fix(already-cancelled-sub): UI should allow restoring subscription

* restore functionality fixed

* fix

* Add pause resume block

* Add db schema

* Initial test passes

* Tests pass

* Execution pauses

* Snapshot serializer

* Ui checkpoint

* Works 1

* Pause resume simple v1

* Hitl block works in parallel branches without timing overlap

* Pending status to logs

* Pause resume ui link

* Big context consolidation

* HITL works in loops

* Fix parallels

* Reference blocks properly

* Fix tag dropdown and start block resolution

* Filter console logs for hitl block

* Fix notifs

* Fix logs page

* Fix logs page again

* Fix

* Checkpoint

* Cleanup v1

* Refactor v2

* Refactor v3

* Refactor v4

* Refactor v5

* Resume page

* Fix variables in loops

* Fix var res bugs

* Ui changes

* Approval block

* Hitl works e2e v1

* Fix tets

* Row level lock

---------

Co-authored-by: Waleed <[email protected]>
Co-authored-by: Vikhyath Mondreti <[email protected]>

Author: 3 people

apps/sim/app/api/logs/route.ts

@@ -1,5 +1,5 @@
 import { db } from '@sim/db'
-import { permissions, workflow, workflowExecutionLogs } from '@sim/db/schema'
+import { pausedExecutions, permissions, workflow, workflowExecutionLogs } from '@sim/db/schema'
 import { and, desc, eq, gte, inArray, lte, type SQL, sql } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { z } from 'zod'
@@ -68,6 +68,9 @@ export async function GET(request: NextRequest) {
               workflowWorkspaceId: workflow.workspaceId,
               workflowCreatedAt: workflow.createdAt,
               workflowUpdatedAt: workflow.updatedAt,
+              pausedStatus: pausedExecutions.status,
+              pausedTotalPauseCount: pausedExecutions.totalPauseCount,
+              pausedResumedCount: pausedExecutions.resumedCount,
             }
           : {
               // Basic mode - exclude large fields for better performance
@@ -92,11 +95,18 @@ export async function GET(request: NextRequest) {
               workflowWorkspaceId: workflow.workspaceId,
               workflowCreatedAt: workflow.createdAt,
               workflowUpdatedAt: workflow.updatedAt,
+              pausedStatus: pausedExecutions.status,
+              pausedTotalPauseCount: pausedExecutions.totalPauseCount,
+              pausedResumedCount: pausedExecutions.resumedCount,
             }
 
       const baseQuery = db
         .select(selectColumns)
         .from(workflowExecutionLogs)
+        .leftJoin(
+          pausedExecutions,
+          eq(pausedExecutions.executionId, workflowExecutionLogs.executionId)
+        )
         .innerJoin(
           workflow,
           and(
@@ -186,6 +196,10 @@ export async function GET(request: NextRequest) {
       const countQuery = db
         .select({ count: sql<number>`count(*)` })
         .from(workflowExecutionLogs)
+        .leftJoin(
+          pausedExecutions,
+          eq(pausedExecutions.executionId, workflowExecutionLogs.executionId)
+        )
         .innerJoin(
           workflow,
           and(
@@ -340,13 +354,18 @@ export async function GET(request: NextRequest) {
         return {
           id: log.id,
           workflowId: log.workflowId,
-          executionId: params.details === 'full' ? log.executionId : undefined,
+          executionId: log.executionId,
           level: log.level,
           duration: log.totalDurationMs ? `${log.totalDurationMs}ms` : null,
           trigger: log.trigger,
           createdAt: log.startedAt.toISOString(),
           files: params.details === 'full' ? log.files || undefined : undefined,
           workflow: workflowSummary,
+          pauseSummary: {
+            status: log.pausedStatus ?? null,
+            total: log.pausedTotalPauseCount ?? 0,
+            resumed: log.pausedResumedCount ?? 0,
+          },
           executionData:
             params.details === 'full'
               ? {
@@ -361,6 +380,10 @@ export async function GET(request: NextRequest) {
             params.details === 'full'
               ? (costSummary as any)
               : { total: (costSummary as any)?.total || 0 },
+          hasPendingPause:
+            (Number(log.pausedTotalPauseCount ?? 0) > 0 &&
+              Number(log.pausedResumedCount ?? 0) < Number(log.pausedTotalPauseCount ?? 0)) ||
+            (log.pausedStatus && log.pausedStatus !== 'fully_resumed'),
         }
       })
       return NextResponse.json(

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

@@ -0,0 +1,116 @@
+import { type NextRequest, NextResponse } from 'next/server'
+import { createLogger } from '@/lib/logs/console/logger'
+import { PauseResumeManager } from '@/lib/workflows/executor/pause-resume-manager'
+import { validateWorkflowAccess } from '@/app/api/workflows/middleware'
+
+const logger = createLogger('WorkflowResumeAPI')
+
+export const runtime = 'nodejs'
+export const dynamic = 'force-dynamic'
+
+export async function POST(
+  request: NextRequest,
+  {
+    params,
+  }: {
+    params: Promise<{ workflowId: string; executionId: string; contextId: string }>
+  }
+) {
+  const { workflowId, executionId, contextId } = await params
+
+  const access = await validateWorkflowAccess(request, workflowId, false)
+  if (access.error) {
+    return NextResponse.json({ error: access.error.message }, { status: access.error.status })
+  }
+
+  const workflow = access.workflow!
+
+  let payload: any = {}
+  try {
+    payload = await request.json()
+  } catch {
+    payload = {}
+  }
+
+  const resumeInput = payload?.input ?? payload ?? {}
+  const userId = workflow.userId ?? ''
+
+  try {
+    const enqueueResult = await PauseResumeManager.enqueueOrStartResume({
+      executionId,
+      contextId,
+      resumeInput,
+      userId,
+    })
+
+    if (enqueueResult.status === 'queued') {
+      return NextResponse.json({
+        status: 'queued',
+        executionId: enqueueResult.resumeExecutionId,
+        queuePosition: enqueueResult.queuePosition,
+        message: 'Resume queued. It will run after current resumes finish.',
+      })
+    }
+
+    PauseResumeManager.startResumeExecution({
+      resumeEntryId: enqueueResult.resumeEntryId,
+      resumeExecutionId: enqueueResult.resumeExecutionId,
+      pausedExecution: enqueueResult.pausedExecution,
+      contextId: enqueueResult.contextId,
+      resumeInput: enqueueResult.resumeInput,
+      userId: enqueueResult.userId,
+    }).catch((error) => {
+      logger.error('Failed to start resume execution', {
+        workflowId,
+        parentExecutionId: executionId,
+        resumeExecutionId: enqueueResult.resumeExecutionId,
+        error,
+      })
+    })
+
+    return NextResponse.json({
+      status: 'started',
+      executionId: enqueueResult.resumeExecutionId,
+      message: 'Resume execution started.',
+    })
+  } catch (error: any) {
+    logger.error('Resume request failed', {
+      workflowId,
+      executionId,
+      contextId,
+      error,
+    })
+    return NextResponse.json(
+      { error: error.message || 'Failed to queue resume request' },
+      { status: 400 }
+    )
+  }
+}
+
+export async function GET(
+  request: NextRequest,
+  {
+    params,
+  }: {
+    params: Promise<{ workflowId: string; executionId: string; contextId: string }>
+  }
+) {
+  const { workflowId, executionId, contextId } = await params
+
+  const access = await validateWorkflowAccess(request, workflowId, false)
+  if (access.error) {
+    return NextResponse.json({ error: access.error.message }, { status: access.error.status })
+  }
+
+  const detail = await PauseResumeManager.getPauseContextDetail({
+    workflowId,
+    executionId,
+    contextId,
+  })
+
+  if (!detail) {
+    return NextResponse.json({ error: 'Pause context not found' }, { status: 404 })
+  }
+
+  return NextResponse.json(detail)
+}

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

@@ -0,0 +1,48 @@
+import { type NextRequest, NextResponse } from 'next/server'
+import { createLogger } from '@/lib/logs/console/logger'
+import { PauseResumeManager } from '@/lib/workflows/executor/pause-resume-manager'
+import { validateWorkflowAccess } from '@/app/api/workflows/middleware'
+
+const logger = createLogger('WorkflowResumeExecutionAPI')
+
+export const runtime = 'nodejs'
+export const dynamic = 'force-dynamic'
+
+export async function GET(
+  request: NextRequest,
+  {
+    params,
+  }: {
+    params: Promise<{ workflowId: string; executionId: string }>
+  }
+) {
+  const { workflowId, executionId } = await params
+
+  const access = await validateWorkflowAccess(request, workflowId, false)
+  if (access.error) {
+    return NextResponse.json({ error: access.error.message }, { status: access.error.status })
+  }
+
+  try {
+    const detail = await PauseResumeManager.getPausedExecutionDetail({
+      workflowId,
+      executionId,
+    })
+
+    if (!detail) {
+      return NextResponse.json({ error: 'Paused execution not found' }, { status: 404 })
+    }
+
+    return NextResponse.json(detail)
+  } catch (error: any) {
+    logger.error('Failed to load paused execution detail', {
+      workflowId,
+      executionId,
+      error,
+    })
+    return NextResponse.json(
+      { error: error?.message || 'Failed to load paused execution detail' },
+      { status: 500 }
+    )
+  }
+}

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

@@ -12,6 +12,7 @@ import {
 } from '@/lib/workflows/db-helpers'
 import { executeWorkflowCore } from '@/lib/workflows/executor/execution-core'
 import { type ExecutionEvent, encodeSSEEvent } from '@/lib/workflows/executor/execution-events'
+import { PauseResumeManager } from '@/lib/workflows/executor/pause-resume-manager'
 import { validateWorkflowAccess } from '@/app/api/workflows/middleware'
 import { type ExecutionMetadata, ExecutionSnapshot } from '@/executor/execution/snapshot'
 import type { StreamingExecution } from '@/executor/types'
@@ -135,6 +136,24 @@ export async function executeWorkflow(
       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,
@@ -605,6 +624,24 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
             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 (result.error === 'Workflow execution was cancelled') {
             logger.info(`[${requestId}] Workflow execution was cancelled`)
             sendEvent({

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

@@ -0,0 +1,34 @@
+import { type NextRequest, NextResponse } from 'next/server'
+import { PauseResumeManager } from '@/lib/workflows/executor/pause-resume-manager'
+import { validateWorkflowAccess } from '@/app/api/workflows/middleware'
+
+export const runtime = 'nodejs'
+export const dynamic = 'force-dynamic'
+
+export async function GET(
+  request: NextRequest,
+  {
+    params,
+  }: {
+    params: { id: string; executionId: string }
+  }
+) {
+  const workflowId = params.id
+  const executionId = params.executionId
+
+  const access = await validateWorkflowAccess(request, workflowId, false)
+  if (access.error) {
+    return NextResponse.json({ error: access.error.message }, { status: access.error.status })
+  }
+
+  const detail = await PauseResumeManager.getPausedExecutionDetail({
+    workflowId,
+    executionId,
+  })
+
+  if (!detail) {
+    return NextResponse.json({ error: 'Paused execution not found' }, { status: 404 })
+  }
+
+  return NextResponse.json(detail)
+}

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

@@ -0,0 +1,31 @@
+import { type NextRequest, NextResponse } from 'next/server'
+import { PauseResumeManager } from '@/lib/workflows/executor/pause-resume-manager'
+import { validateWorkflowAccess } from '@/app/api/workflows/middleware'
+
+export const runtime = 'nodejs'
+export const dynamic = 'force-dynamic'
+
+export async function GET(
+  request: NextRequest,
+  {
+    params,
+  }: {
+    params: { id: string }
+  }
+) {
+  const workflowId = params.id
+
+  const access = await validateWorkflowAccess(request, workflowId, false)
+  if (access.error) {
+    return NextResponse.json({ error: access.error.message }, { status: access.error.status })
+  }
+
+  const statusFilter = request.nextUrl.searchParams.get('status') || undefined
+
+  const pausedExecutions = await PauseResumeManager.listPausedExecutions({
+    workflowId,
+    status: statusFilter,
+  })
+
+  return NextResponse.json({ pausedExecutions })
+}

apps/sim/app/resume/[workflowId]/[executionId]/[contextId]/page.tsx

@@ -0,0 +1,15 @@
+import { redirect } from 'next/navigation'
+
+export const runtime = 'nodejs'
+export const dynamic = 'force-dynamic'
+
+interface PageParams {
+  workflowId: string
+  executionId: string
+  contextId: string
+}
+
+export default async function ResumePage({ params }: { params: Promise<PageParams> }) {
+  const { workflowId, executionId, contextId } = await params
+  redirect(`/resume/${workflowId}/${executionId}?contextId=${contextId}`)
+}