Workflow Engine

Why a DAG?

A workflow is a set of processing nodes with dependencies between them. The classifier must run before the generator (because the generator needs the classification). But the classifier and summarizer are independent — they can run at the same time.

This is a directed acyclic graph (DAG). Nodes are vertices. Dependencies are directed edges. "Acyclic" means no circular dependencies — a node can't depend on its own output.

DAG execution gives you two things that sequential execution doesn't:

Topological Execution

The DAG executor processes nodes in topological order — a node only executes when all its dependencies have completed. The algorithm:

For the support ticket workflow:

Total time ≈ max(classifier, summarizer, extractor) + generator, not the sum of all four.

Conditional Routing

Not every event should follow the same path. A billing ticket needs different handling than a feature request. A critical-urgency event needs immediate escalation.

Edge conditions control routing:

edges: [
  { from: "classifier", to: "escalation_handler",
    condition: "classifier.urgency === 'critical'" },
  { from: "classifier", to: "standard_handler",
    condition: "classifier.urgency !== 'critical'" },
]

The executor evaluates conditions by reading the actual output of completed nodes. If the classifier returned urgency: "critical", the event routes to the escalation handler. Otherwise, it goes to the standard handler.

This is data-driven routing — the workflow adapts to each event based on the AI's analysis, not hardcoded rules.

Routing Rules

For more complex routing, the ConditionalRouter class provides a rules engine:

router.addRule({
  id: "critical_escalation",
  name: "Critical Escalation",
  condition: (results) => results["classifier"]?.output?.urgency === "critical",
  targetNodes: ["page_oncall", "notify_manager", "generator"],
  priority: 100,
});

Rules are evaluated by priority (highest first). The first matching rule determines the routing. This is the chain of responsibility pattern — adding a new route is just one addRule() call.

Parallel Execution

When multiple nodes are ready simultaneously, the parallel runner executes them concurrently with two safety mechanisms:

await runParallel(nodes, input, {
  maxConcurrency: 5,
  timeoutMs: 30_000,
});

Join Strategies

When parallel branches converge, how long do you wait?

The join handler reports which branches are complete, which are pending, and whether the join condition is satisfied. The dashboard uses this to show real-time progress.

What's Next

In Module 4, you'll add reliability to this engine. What happens when a node fails? When the AI API times out? When a webhook delivers the same event twice? Retry policies, circuit breakers, dead letter queues, and idempotency guards handle all of these.