Human-in-the-Loop Patterns in simstudioai/sim: How the Executor Pauses for User Input

The simstudioai/sim workflow executor pauses execution at Human-in-the-Loop (HiTL) blocks by creating a PausePoint, persisting a snapshot to the pausedExecutions table, and waiting for user input via a FIFO resumeQueue before rewiring the DAG and continuing.

The simstudioai/sim repository implements a robust Human-in-the-Loop (HiTL) system that allows workflows to stop mid-execution and wait for human intervention. This pattern relies on a transactional pause-resume mechanism that preserves execution state, queues user submissions, and safely rewires the workflow DAG to continue processing once input is received.

Core Architecture of HiTL Patterns

The executor implements three coordinated concepts to manage human-in-the-loop patterns:

• Pause Points – Markers created when the executor hits a block of type "human_in_the_loop". Each PausePoint (defined in apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts) captures the contextId, blockId, current snapshot, and a resumeStatus flag set to "paused".
• Paused Execution Records – Database rows in the pausedExecutions table (@sim/db/schema) that store the full executionSnapshot and a JSONB map of pause points, enabling the UI to list paused runs and track queue position.
• Resume Queue – A FIFO queue implemented in the resumeQueue table that guarantees only one resume runs at a time per execution while allowing multiple users to claim the same pause point.

Detailed Execution Flow

When the executor encounters a HiTL block, it follows a precise sequence to pause safely and resume transactionally.

Detecting the HiTL Block

Each block executor checks its block type against the registry. When the type matches "human_in_the_loop", the executor creates a PausePoint and returns a paused ExecutionResult instead of completing. This triggers the pause handling logic in human-in-the-loop-manager.ts.

Persisting the Pause State

The PauseResumeManager.persistPauseResult method (lines 26–84 in human-in-the-loop-manager.ts) handles the transaction:

1. Stores the executionSnapshot and a map of pausePoints.
2. Executes an INSERT … ON CONFLICT DO UPDATE against the pausedExecutions table, setting status = "paused" and incrementing totalPauseCount.
3. Immediately calls processQueuedResumes to check for any already-queued user inputs awaiting processing.

Simultaneously, the system streams an execution:paused event to the client containing the block output, pause ID, and optional resumeLinks.

Queueing User Input

User submissions arrive via the HTTP endpoint POST /api/v1/human-in-the-loop/:executionId/:contextId implemented in apps/sim/executor/handlers/human-in-the-loop/human-in-the-loop-handler.ts. The handler delegates to PauseResumeManager.enqueueOrStartResume (lines 86–123):

• If a resume is already active, the request inserts into resumeQueue with status = "pending" and returns a queuePosition to the user.
• If no resume is active, the entry inserts with status = "claimed" and the manager immediately invokes startResumeExecution.

Resuming and Rewiring the Workflow

The startResumeExecution method orchestrates the continuation:

1. Load Snapshot: Retrieves the stored pausedExecution.executionSnapshot from the database.
2. Merge Input: Combines the user’s payload into the pause block’s output, creating fields submission, submittedAt, _resumed, and _pauseDurationMs (lines 526–590).
3. Rewire DAG: Removes edges originating from the completed pause block by updating stateCopy.remainingEdges and completedPauseContexts (lines 670–720).
4. Update State: Sets pauseBlockState.executed = true and updates execution time metrics.
5. Continue Execution: Creates a fresh ExecutionSnapshot (resumeSnapshot) and passes it to executeWorkflowCore, which runs exactly like a fresh execution using the updated DAG.

When the resumed run finishes, markResumeCompleted (or markResumeFailed) updates the record and triggers processQueuedResumes again, allowing the next queued entry to start.

Cancellation Support

The PauseResumeManager.cancelPausedExecution method can abort a paused workflow, clearing both the pausedExecutions row and the workflow execution log, effectively resetting the state as if the pause never occurred.

Code Implementation Details

Creating a Pause Point

Inside a block executor when detecting a HiTL block:

// Simplified excerpt from the executor core
const pausePoint: PausePoint = {
  contextId,
  blockId,
  response: blockOutput.response,
  resumeStatus: 'paused',
  registeredAt: new Date(),
  snapshotReady: true,
};

return { 
  status: 'paused', 
  pausePoints: [pausePoint], 
  snapshotSeed 
};

Persisting to the Database

The manager persists the pause atomically:

await PauseResumeManager.persistPauseResult({
  workflowId,
  executionId,
  pausePoints: result.pausePoints,
  snapshotSeed: result.snapshotSeed,
  executorUserId,
});

Enqueueing or Starting Resume

Handling concurrent user submissions:

const enqueueResult = await PauseResumeManager.enqueueOrStartResume({
  executionId,
  contextId,
  resumeInput,
  userId,
});

// Returns either:
// { status: 'queued', queuePosition: number }
// { status: 'starting', resumeEntryId: string, ... }

Merging User Input and Continuing

During runResumeExecution, the system merges inputs and continues:

const mergedOutput = {
  ...existingOutput,
  response: mergedResponse,
  submission: submissionPayload,
  submittedAt: new Date(),
  _resumed: true,
  _pauseDurationMs: pauseDurationMs,
};

pauseBlockState.output = mergedOutput;
pauseBlockState.executed = true;

// Remove edges from the completed pause block
stateCopy.remainingEdges = edgesToRemove;
stateCopy.completedPauseContexts = Array.from(completedPauseContexts);

// Continue execution
const result = await executeWorkflowCore({
  snapshot: resumeSnapshot,
  callbacks,
  loggingSession,
  abortSignal,
});

Key Source Files

Path	Role
apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts	Central pause/resume logic, PauseResumeManager class, DB persistence, and queue handling.
apps/sim/executor/handlers/human-in-the-loop/human-in-the-loop-handler.ts	HTTP API endpoint receiving user submissions and delegating to the manager.
apps/sim/executor/execution/executor.ts	Entry point for workflow execution that routes to executeWorkflowCore.
apps/sim/executor/execution/core.ts	Detects pause points and returns paused ExecutionResult to upstream callers.
apps/sim/executor/execution/engine.ts	DAG execution engine that processes the updated snapshot after resume.
apps/sim/db/schema.ts	Defines pausedExecutions and resumeQueue tables with JSONB support for snapshots.

Summary

• Pause Creation: HiTL blocks generate PausePoint objects that stop execution and trigger persistPauseResult to save snapshots to the pausedExecutions table.
• Event Streaming: The executor emits execution:paused events containing block output and resume metadata for UI consumption.
• Input Queueing: The resumeQueue table manages FIFO ordering of user submissions, ensuring single active execution per workflow with status tracking (pending vs claimed).
• State Restoration: The startResumeExecution method reloads snapshots, merges submission data into block outputs, and rewires the DAG by removing completed pause edges.
• Core Reuse: Resumed workflows call executeWorkflowCore with a fresh resumeSnapshot, ensuring consistent behavior between initial and resumed executions.

Frequently Asked Questions

How does the executor ensure only one resume runs at a time per execution?

The PauseResumeManager uses the resumeQueue table with a status field. When a user submits input, enqueueOrStartResume checks for existing claimed entries. If found, the new request inserts with status = "pending" and receives a queuePosition. The processQueuedResumes worker only starts the next resume after the current one calls markResumeCompleted, ensuring atomic, sequential processing.

What data structure stores the paused workflow state?

The pausedExecutions table stores a JSONB executionSnapshot containing the complete DAG state, variable context, and a pausePointsRecord map keyed by contextId. Each PausePoint includes blockId, response, resumeStatus, and registeredAt timestamps, allowing full state restoration during startResumeExecution.

How is user input merged into the workflow when resuming?

During runResumeExecution, the manager creates a mergedOutput object that spreads the existing block output and adds the user's submission payload alongside metadata fields: submittedAt, _resumed: true, and _pauseDurationMs. This merged output attaches to the pauseBlockState before the DAG edges are removed and executeWorkflowCore continues processing downstream blocks.

Can a paused execution be cancelled before user input arrives?

Yes. The PauseResumeManager.cancelPausedExecution method deletes the pausedExecutions row and clears the associated workflow execution log. This effectively discards the snapshot and pause points, preventing any future resumes from that state and freeing the execution ID for fresh runs.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}