fix(hitl): fix stream endpoint, pause persistence, and resume page

apps/docs/content/docs/en/blocks/human-in-the-loop.mdx

@@ -93,17 +93,36 @@ Access resume data in downstream blocks using `<blockId.resumeInput.fieldName>`.
  <Tab>
  ### REST API
 
- Programmatically resume workflows:
+ Programmatically resume workflows using the resume endpoint. The `contextId` is available from the block's `resumeEndpoint` output or from the paused execution detail.
 
  ```bash
- POST /api/workflows/{workflowId}/executions/{executionId}/resume/{blockId}
+ POST /api/resume/{workflowId}/{executionId}/{contextId}
+ Content-Type: application/json
 
  {
- "approved": true,
- "comments": "Looks good to proceed"
+ "input": {
+ "approved": true,
+ "comments": "Looks good to proceed"
+ }
  }
  ```
 
+ The response includes a new `executionId` for the resumed execution:
+
+ ```json
+ {
+ "status": "started",
+ "executionId": "<resumeExecutionId>",
+ "message": "Resume execution started."
+ }
+ ```
+
+ To poll execution progress after resuming, connect to the SSE stream:
+
+ ```bash
+ GET /api/workflows/{workflowId}/executions/{resumeExecutionId}/stream
+ ```
+
  Build custom approval UIs or integrate with existing systems.
  </Tab>
  <Tab>

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

@@ -3,6 +3,7 @@ import { type NextRequest, NextResponse } from 'next/server'
 import { AuthType } from '@/lib/auth/hybrid'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { generateId } from '@/lib/core/utils/uuid'
+import { setExecutionMeta } from '@/lib/execution/event-buffer'
 import { preprocessExecution } from '@/lib/execution/preprocessing'
 import { PauseResumeManager } from '@/lib/workflows/executor/human-in-the-loop-manager'
 import { getWorkspaceBilledAccountUserId } from '@/lib/workspaces/utils'
@@ -125,14 +126,43 @@ export async function POST(
  })
  }
 
- PauseResumeManager.startResumeExecution({
+ await setExecutionMeta(enqueueResult.resumeExecutionId, {
+ status: 'active',
+ userId,
+ workflowId,
+ })
+
+ 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,
+ status: result.status ?? (result.success ? 'completed' : 'failed'),
+ 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

@@ -41,7 +41,7 @@ import {
 } from '@/lib/uploads/utils/user-file-base64.server'
 import { executeWorkflowCore } from '@/lib/workflows/executor/execution-core'
 import { type ExecutionEvent, encodeSSEEvent } from '@/lib/workflows/executor/execution-events'
-import { PauseResumeManager } from '@/lib/workflows/executor/human-in-the-loop-manager'
+import { handlePostExecutionPauseState } from '@/lib/workflows/executor/pause-persistence'
 import {
  DIRECT_WORKFLOW_JOB_NAME,
  type QueuedWorkflowExecutionPayload,
@@ -903,6 +903,8 @@ async function handleExecutePost(
  abortSignal: timeoutController.signal,
  })
 
+ await handlePostExecutionPauseState({ result, workflowId, executionId, loggingSession })
+
  if (
  result.status === 'cancelled' &&
  timeoutController.isTimedOut() &&
@@ -1359,31 +1361,7 @@ async function handleExecutePost(
  runFromBlock: resolvedRunFromBlock,
  })
 
- if (result.status === 'paused') {
- if (!result.snapshotSeed) {
- reqLogger.error('Missing snapshot seed for paused execution')
- await loggingSession.markAsFailed('Missing snapshot seed for paused execution')
- } else {
- try {
- await PauseResumeManager.persistPauseResult({
- workflowId,
- executionId,
- pausePoints: result.pausePoints || [],
- snapshotSeed: result.snapshotSeed,
- executorUserId: result.metadata?.userId,
- })
- } catch (pauseError) {
- reqLogger.error('Failed to persist pause result', {
- error: pauseError instanceof Error ? pauseError.message : String(pauseError),
- })
- await loggingSession.markAsFailed(
- `Failed to persist pause state: ${pauseError instanceof Error ? pauseError.message : String(pauseError)}`
- )
- }
- }
- } else {
- await PauseResumeManager.processQueuedResumes(executionId)
- }
+ await handlePostExecutionPauseState({ result, workflowId, executionId, loggingSession })
 
  if (result.status === 'cancelled') {
  if (timeoutController.isTimedOut() && timeoutController.timeoutMs) {
@@ -1422,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/app/api/workflows/[id]/executions/[executionId]/stream/route.ts

@@ -1,6 +1,6 @@
 import { createLogger } from '@sim/logger'
 import { type NextRequest, NextResponse } from 'next/server'
-import { checkHybridAuth } from '@/lib/auth/hybrid'
+import { getSession } from '@/lib/auth'
 import { SSE_HEADERS } from '@/lib/core/utils/sse'
 import {
  type ExecutionStreamStatus,
@@ -29,14 +29,14 @@ export async function GET(
  const { id: workflowId, executionId } = await params
 
  try {
- const auth = await checkHybridAuth(req, { requireWorkflowId: false })
- if (!auth.success || !auth.userId) {
- return NextResponse.json({ error: auth.error || 'Unauthorized' }, { status: 401 })
+ const session = await getSession()
+ if (!session?.user?.id) {
+ return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
  }
 
  const workflowAuthorization = await authorizeWorkflowByWorkspacePermission({
  workflowId,
- userId: auth.userId,
+ userId: session.user.id,
  action: 'read',
  })
  if (!workflowAuthorization.allowed) {
@@ -46,16 +46,6 @@ export async function GET(
  )
  }
 
- if (
- auth.apiKeyType === 'workspace' &&
- workflowAuthorization.workflow?.workspaceId !== auth.workspaceId
- ) {
- return NextResponse.json(
- { error: 'API key is not authorized for this workspace' },
- { status: 403 }
- )
- }
-
  const meta = await getExecutionMeta(executionId)
  if (!meta) {
  return NextResponse.json({ error: 'Execution buffer not found or expired' }, { status: 404 })

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

@@ -6,6 +6,7 @@ import { useRouter } from 'next/navigation'
 import {
  Badge,
  Button,
+ Code,
  Input,
  Label,
  Table,
@@ -155,14 +156,54 @@ function getBlockNameFromSnapshot(
  const parsed = JSON.parse(executionSnapshot.snapshot)
  const workflowState = parsed?.workflow
  if (!workflowState?.blocks || !Array.isArray(workflowState.blocks)) return null
- // Blocks are stored as an array of serialized blocks with id and metadata.name
  const block = workflowState.blocks.find((b: { id: string }) => b.id === blockId)
  return block?.metadata?.name || null
  } catch {
  return null
  }
 }
 
+function renderStructuredValuePreview(value: unknown) {
+ if (value === null || value === undefined) {
+ return <span style={{ fontSize: '12px', color: 'var(--text-muted)' }}>—</span>
+ }
+
+ if (typeof value === 'object') {
+ return (
+ <div style={{ minWidth: '220px' }}>
+ <Code.Viewer
+ code={JSON.stringify(value, null, 2)}
+ language='json'
+ wrapText
+ className='max-h-[220px]'
+ />
+ </div>
+ )
+ }
+
+ const stringValue = String(value)
+ return (
+ <div
+ style={{
+ display: 'inline-flex',
+ maxWidth: '100%',
+ borderRadius: '6px',
+ border: '1px solid var(--border)',
+ background: 'var(--surface-5)',
+ padding: '4px 8px',
+ whiteSpace: 'pre-wrap',
+ wordBreak: 'break-word',
+ fontFamily: 'var(--font-mono, monospace)',
+ fontSize: '12px',
+ lineHeight: '16px',
+ color: 'var(--text-primary)',
+ }}
+ >
+ {stringValue}
+ </div>
+ )
+}
+
 export default function ResumeExecutionPage({
  params,
  initialExecutionDetail,
@@ -874,8 +915,11 @@ export default function ResumeExecutionPage({
  <Tooltip.Trigger asChild>
  <Button
  variant='outline'
+ size='sm'
  onClick={refreshExecutionDetail}
  disabled={refreshingExecution}
+ className='gap-1.5 px-2.5'
+ aria-label='Refresh execution details'
  >
  <RefreshCw
  style={{
@@ -884,6 +928,7 @@ export default function ResumeExecutionPage({
  animation: refreshingExecution ? 'spin 1s linear infinite' : undefined,
  }}
  />
+ Refresh
  </Button>
  </Tooltip.Trigger>
  <Tooltip.Content>Refresh</Tooltip.Content>
@@ -1123,11 +1168,7 @@ export default function ResumeExecutionPage({
  <TableRow key={row.id}>
  <TableCell>{row.name}</TableCell>
  <TableCell>{row.type}</TableCell>
- <TableCell>
- <code style={{ fontSize: '12px' }}>
- {formatStructureValue(row.value)}
- </code>
- </TableCell>
+ <TableCell>{renderStructuredValuePreview(row.value)}</TableCell>
  </TableRow>
  ))}
  </TableBody>
@@ -1243,6 +1284,8 @@ export default function ResumeExecutionPage({
  }}
  placeholder='{"example": "value"}'
  rows={6}
+ spellCheck={false}
+ className='min-h-[180px] border-[var(--border-1)] bg-[var(--surface-3)] font-mono text-[12px] leading-5'
  />
  </div>
  </div>
@@ -1267,10 +1310,10 @@ export default function ResumeExecutionPage({
  {/* Footer */}
  <div
  style={{
- marginTop: '32px',
- padding: '16px',
+ maxWidth: '1200px',
+ margin: '24px auto 0',
+ padding: '0 24px 24px',
  textAlign: 'center',
- borderTop: '1px solid var(--border)',
  fontSize: '13px',
  color: 'var(--text-muted)',
  }}