Workflow Engine

The NetPad Workflow Engine powers all node-based automation, agentic flows, and plugin execution. This section covers its architecture, extensibility, and usage for advanced developers.

1. Architecture Overview

• Event-driven: Workflows are triggered and managed via events (using Inngest).
• Node runners: Each node type has a dedicated runner for its logic.
• Agentic/conditional edges: Enable dynamic, AI-driven, and conditional branching.
• Pausing/resuming: Supports user input, prompt nodes, and resumable execution.
• Trace/logging: Every step is logged for debugging and analysis.

2. Execution Engine

• File: src/app/engine/executionEngine.js
• Orchestrates node-by-node execution, manages context, memory, and agentic logic.
• Handles pausing (for user input), resuming, and error handling.
• Maintains a trace of all steps for debugging.

Example:

import { runWorkflow } from '@/app/engine/executionEngine';
 
const result = await runWorkflow({
  nodes, // array of node definitions
  edges, // array of edge definitions
  startNodeId, // id of the start node
  initialContext: {},
  logStep: (step) => console.log(step),
  onUserInputNeeded: (prompt, nodeId) => {/* ... */}
});

3. Node Runners

• Each node type has a runner in /src/app/runners/ (e.g., agentNodeRunner.js, toolNodeRunner.js).
• Runners receive helpers for input/output, context, logging, and triggering downstream nodes.
• To add a new node type, create a runner and register it in the engine.

Example runner:

// runners/textNodeRunner.js
export async function runTextNode(node, { getInput, setOutput, context, log }) {
  const [input] = await getInput('input1');
  setOutput('output1', { value: input.value.toUpperCase() });
}

4. Agentic & Conditional Edges

• Edges can be conditional, agentic (AI-driven), or default.
• The engine evaluates edge conditions to determine execution flow.
• Supports advanced patterns: error recovery, iteration, fallback, memory flow, etc.

5. Memory & Context

• Memory is injected for agent nodes and can be persisted across runs.
• Context is passed to all runners and can include user/session data, workflow state, etc.

6. Inngest Integration

• Workflows can be started, paused, and resumed via Inngest events.
• See src/app/api/workflow/start-inngest/route.js and inngestFunctions.js.

Example:

// Start a workflow via API
await fetch('/api/workflow/start-inngest', {
  method: 'POST',
  body: JSON.stringify({ workflow }),
  headers: { 'Content-Type': 'application/json' }
});

7. API Usage

• Start: POST /api/workflow/start-inngest
• Input: POST /api/workflow/input-inngest
• Run (sync): POST /api/workflow/run
• Resume: POST /api/workflow/run/[runId]/resume
• Status: GET /api/workflow/status/[runId]

8. Debugging & Tracing

• Every workflow run produces a trace (step-by-step log).
• Errors, pauses, and user input are all recorded.
• Use the trace for debugging, analytics, and audit.

9. Extending the Engine

• Add new node types by creating a runner and registering it.
• Use the helpers provided to interact with the engine (input, output, context, trigger).
• Follow best practices for error handling and logging.

For more, see the source code or reach out to the core team for advanced integration support.