Graph-Based Sales Workflow DSL

Overview

Graph-based workflow DSLs represent business processes as directed graphs where nodes encode discrete operations and edges define control flow between them. For AI-powered sales automation, this approach lets teams compose complex prospect journeys from reusable building blocks — trigger nodes that fire on CRM events, filter nodes that evaluate lead scoring rules, AI analysis nodes that summarize contact history or draft outreach copy, and queue steps that hand off work to human reps. The entire workflow is serialized as JSON, making it storable in a database, versionable in git, and renderable in a visual editor.

The schema design draws heavily from ReactFlow’s node/edge model: every node carries a unique id, a type discriminator, a data payload specific to that type, and a position object with x/y coordinates for the canvas. Edges link a source node to a target node and can carry a label (e.g., "true" / "false" for conditional branches). This dual concern — runtime execution semantics and visual layout metadata — lives in a single JSON document, so the same artifact drives both the workflow engine and the drag-and-drop builder.

Storing positional metadata alongside execution logic is a deliberate trade-off. It couples the visual representation to the workflow definition, but it eliminates a separate layout store and ensures that what the user sees in the builder is exactly what the engine executes. Teams building sales automation tools benefit from this because product managers can inspect, annotate, and modify workflows visually while engineers trust that the serialized JSON is the authoritative source of truth for runtime behavior.

Core Concepts

How It Works

Executing a graph-based workflow involves three stages: deserializing the JSON document into an in-memory graph, traversing the graph from trigger nodes forward, and dispatching each node to its type-specific handler.

Graph Serialization to JSON

When a user saves a workflow in the visual builder, the frontend serializes the ReactFlow state into the canonical JSON format and POSTs it to the backend. The backend stores the document in a workflows table with a definition JSONB column. On load, the backend returns the raw JSON and the frontend hydrates the ReactFlow graph directly — no transformation needed. Versioning is handled by storing each save as an immutable snapshot with an incrementing version number; the active version is a separate pointer. This lets teams roll back a workflow to a previous version without losing history.

Workflow Execution Engine Traversal

The execution engine builds an adjacency list from the edges array, mapping each node id to its outbound neighbors (with edge labels preserved for conditional routing). Execution begins by enqueuing all trigger nodes. A BFS/DFS loop dequeues a node, executes its handler, receives a result that may include a nextHandle string (e.g., "true" or "false"), and then enqueues the neighbors connected via edges whose sourceHandle matches nextHandle. For unconditional nodes, all outbound edges are followed. The engine maintains a context map that accumulates outputs from each node; this context is passed to every handler and is also persisted to a workflow_runs table so that queue steps can resume a paused run after human action.

Node Type Dispatch

Each node type is registered in a handler registry — a map from type string to an async function that accepts the node’s data payload and the current context, and returns an updated context plus an optional nextHandle. Adding a new node type means registering a new handler; the traversal loop requires no changes. The registry pattern also makes testing straightforward: each handler is a pure async function that can be unit-tested in isolation without running the full graph traversal. Integration tests exercise the full engine against fixture workflow JSON documents, asserting on the final context state and the sequence of nodes visited.

Implementation Patterns

Node & Edge Type Definitions

Define discriminated unions for each node type so TypeScript enforces that the correct data shape accompanies each type discriminator.

// types/workflow.ts

export interface Position {
  x: number;
  y: number;
}

// --- Node data payloads ---

export interface ManualTriggerData {
  label: string;
}

export interface ScheduleTriggerData {
  label: string;
  cron: string; // e.g. "0 9 * * 1-5"
  timezone: string;
}

export interface FilterNodeData {
  label: string;
  condition: {
    field: string;
    op: "eq" | "neq" | "gt" | "gte" | "lt" | "lte" | "contains";
    value: unknown;
  };
}

export interface AIAnalysisNodeData {
  label: string;
  model: string;           // e.g. "gpt-4o"
  prompt_template: string; // mustache: "Summarize {{prospect.name}}"
  output_key: string;      // context key to store LLM response
  temperature?: number;
}

export interface QueueStepData {
  label: string;
  queue_name: string;
  task_template: string;
  timeout_hours: number;
}

// --- Discriminated union ---

export type WorkflowNode =
  | { id: string; type: "trigger_manual";   data: ManualTriggerData;   position: Position }
  | { id: string; type: "trigger_schedule"; data: ScheduleTriggerData; position: Position }
  | { id: string; type: "filter";           data: FilterNodeData;      position: Position }
  | { id: string; type: "ai_analysis";      data: AIAnalysisNodeData;  position: Position }
  | { id: string; type: "queue_step";       data: QueueStepData;       position: Position };

export interface WorkflowEdge {
  id: string;
  source: string;
  target: string;
  sourceHandle?: string; // "true" | "false" | undefined
  targetHandle?: string;
  label?: string;
}

export interface WorkflowDefinition {
  id: string;
  name: string;
  version: number;
  nodes: WorkflowNode[];
  edges: WorkflowEdge[];
}

// Example serialized workflow (abbreviated):
// {
//   "id": "wf_prospect_nurture",
//   "name": "High-Score Prospect Nurture",
//   "version": 3,
//   "nodes": [
//     { "id": "n1", "type": "trigger_schedule",
//       "data": { "label": "Daily 9am", "cron": "0 9 * * 1-5", "timezone": "America/New_York" },
//       "position": { "x": 80, "y": 40 } },
//     { "id": "n2", "type": "filter",
//       "data": { "label": "High Score?",
//                 "condition": { "field": "lead_score", "op": "gte", "value": 75 } },
//       "position": { "x": 80, "y": 180 } },
//     { "id": "n3", "type": "ai_analysis",
//       "data": { "label": "Draft Email", "model": "gpt-4o",
//                 "prompt_template": "Draft a follow-up email for {{prospect.name}}",
//                 "output_key": "draft_email" },
//       "position": { "x": 80, "y": 320 } },
//     { "id": "n4", "type": "queue_step",
//       "data": { "label": "Rep Review", "queue_name": "email_review",
//                 "task_template": "Review draft: {{draft_email}}",
//                 "timeout_hours": 24 },
//       "position": { "x": 80, "y": 460 } }
//   ],
//   "edges": [
//     { "id": "e1", "source": "n1", "target": "n2" },
//     { "id": "e2", "source": "n2", "target": "n3", "sourceHandle": "true" },
//     { "id": "e3", "source": "n3", "target": "n4" }
//   ]
// }

Workflow Execution Engine

Traverse the graph from trigger nodes, dispatch each node type to its registered handler, and accumulate context through the run.

// engine/executor.ts
import type { WorkflowDefinition, WorkflowNode, WorkflowEdge } from "../types/workflow";

export type WorkflowContext = Record<string, unknown>;

export interface NodeResult {
  context: WorkflowContext;
  nextHandle?: string; // which output port to follow ("true"/"false"/undefined)
}

export type NodeHandler = (
  node: WorkflowNode,
  context: WorkflowContext
) => Promise<NodeResult>;

// --- Handler registry ---

const handlers = new Map<string, NodeHandler>();

export function registerHandler(type: string, handler: NodeHandler): void {
  handlers.set(type, handler);
}

// --- Built-in handlers ---

registerHandler("trigger_manual", async (_node, ctx) => ({ context: ctx }));

registerHandler("trigger_schedule", async (_node, ctx) => ({ context: ctx }));

registerHandler("filter", async (node, ctx) => {
  if (node.type !== "filter") return { context: ctx };
  const { field, op, value } = node.data.condition;
  const actual = (ctx as Record<string, unknown>)[field];
  let passed = false;
  switch (op) {
    case "gte": passed = (actual as number) >= (value as number); break;
    case "gt":  passed = (actual as number) >  (value as number); break;
    case "lte": passed = (actual as number) <= (value as number); break;
    case "lt":  passed = (actual as number) <  (value as number); break;
    case "eq":  passed = actual === value; break;
    case "neq": passed = actual !== value; break;
    case "contains":
      passed = typeof actual === "string" && actual.includes(value as string);
      break;
  }
  return { context: ctx, nextHandle: passed ? "true" : "false" };
});

registerHandler("ai_analysis", async (node, ctx) => {
  if (node.type !== "ai_analysis") return { context: ctx };
  const { prompt_template, output_key, model, temperature } = node.data;

  // Interpolate {{variable}} placeholders from context
  const prompt = prompt_template.replace(/\{\{(\w+(?:\.\w+)*)\}\}/g, (_match, path) => {
    const val = path.split(".").reduce((o: unknown, k: string) =>
      (o as Record<string, unknown>)?.[k], ctx);
    return val !== undefined ? String(val) : "";
  });

  // Call LLM (stub — replace with real SDK call)
  const response = await callLLM({ model, prompt, temperature: temperature ?? 0.7 });

  return { context: { ...ctx, [output_key]: response } };
});

registerHandler("queue_step", async (node, ctx) => {
  if (node.type !== "queue_step") return { context: ctx };
  // Enqueue task for human review; execution resumes asynchronously
  await enqueueHumanTask({
    queueName: node.data.queue_name,
    taskTemplate: node.data.task_template,
    context: ctx,
    timeoutHours: node.data.timeout_hours,
  });
  // Return a sentinel so the engine knows to suspend this run
  return { context: { ...ctx, __suspended: true, __resume_node: node.id } };
});

// --- Graph traversal ---

export async function executeWorkflow(
  definition: WorkflowDefinition,
  initialContext: WorkflowContext
): Promise<WorkflowContext> {
  // Build adjacency list: nodeId -> [{targetId, sourceHandle}]
  const adjacency = new Map<string, Array<{ targetId: string; sourceHandle?: string }>>();
  for (const edge of definition.edges) {
    const neighbors = adjacency.get(edge.source) ?? [];
    neighbors.push({ targetId: edge.target, sourceHandle: edge.sourceHandle });
    adjacency.set(edge.source, neighbors);
  }

  const nodeMap = new Map<string, WorkflowNode>(
    definition.nodes.map((n) => [n.id, n])
  );

  // Seed queue with trigger nodes
  const queue: string[] = definition.nodes
    .filter((n) => n.type === "trigger_manual" || n.type === "trigger_schedule")
    .map((n) => n.id);

  let ctx: WorkflowContext = { ...initialContext };
  const visited = new Set<string>();

  while (queue.length > 0) {
    const nodeId = queue.shift()!;
    if (visited.has(nodeId)) continue;
    visited.add(nodeId);

    const node = nodeMap.get(nodeId);
    if (!node) continue;

    const handler = handlers.get(node.type);
    if (!handler) throw new Error(`No handler registered for node type: ${node.type}`);

    const result = await handler(node, ctx);
    ctx = result.context;

    // Suspend if a queue step requested it
    if (ctx.__suspended) break;

    // Enqueue outbound neighbors that match the output handle
    const neighbors = adjacency.get(nodeId) ?? [];
    for (const { targetId, sourceHandle } of neighbors) {
      const matchesHandle =
        result.nextHandle === undefined ||
        sourceHandle === undefined ||
        sourceHandle === result.nextHandle;
      if (matchesHandle) {
        queue.push(targetId);
      }
    }
  }

  return ctx;
}

// --- Stubs (replace with real implementations) ---
async function callLLM(_opts: { model: string; prompt: string; temperature: number }): Promise<string> {
  return "[LLM response placeholder]";
}
async function enqueueHumanTask(_task: {
  queueName: string;
  taskTemplate: string;
  context: WorkflowContext;
  timeoutHours: number;
}): Promise<void> {
  // Insert into task queue table
}

Common Pitfalls

• Cycles in the graph causing infinite loops — Without cycle detection, a workflow with a loop-back edge (e.g., a retry path that reconnects to an earlier node) will run forever. Track visited nodes or enforce a maximum step count, and validate that the graph is a DAG before accepting a save unless cycles are explicitly supported with a loop-counter guard.
• Positional metadata drift after programmatic edits — If the execution engine or a migration script modifies the nodes array without preserving position fields, the visual builder will render nodes at (0, 0) and overlap them. Always treat position as a required field in the schema and include it in validation so drift is caught at save time rather than discovered when a designer opens the workflow.
• Context key collisions between AI analysis nodes — When two AI analysis nodes in the same workflow share the same output_key, the second node silently overwrites the first node’s output. Enforce uniqueness of output_key values across all nodes during workflow validation, and consider namespacing outputs by node id (e.g., "n3.draft_email" instead of "draft_email").
• Missing sourceHandle on conditional edges — A filter node has two output ports (true / false), but if an edge is saved without a sourceHandle, the engine treats it as an unconditional edge and follows it regardless of the filter result. Validate that every edge leaving a filter node carries an explicit sourceHandle, and reject workflow saves that omit it.
• Queue step resumption re-executing prior nodes — When a paused workflow resumes after human review, the engine must restart from the queue step node — not from the trigger. Persisting the full traversal state (visited node set and current context) to the database before suspending and restoring it on resume prevents partial re-execution. Failing to do this runs earlier nodes again, double-sending emails or double-charging credits.
• Storing large LLM outputs in the context map — AI analysis nodes can produce kilobytes of text that get carried through the entire downstream context object. Over a long workflow with many AI nodes, the context can balloon to hundreds of kilobytes, making every database read/write for context persistence expensive. Offload large values to blob storage and store only a reference key in the context map.