The Workflow State Machine — Events In, Actions Out

This is the most important page of the chapter. You build pkg/statemachine: the package that decides, for every event, what happens next in a run. It has no I/O — no SQL, no Redis, no clock. That is what makes it small enough to get right and easy to test.

The whole package in one picture

You write it in six files: events.go, state.go, actions.go, machine.go, routing.go and waits.go.

Step 1: Define the events

An event is something that happened. Create pkg/statemachine/events.go:

// pkg/statemachine/events.go
package statemachine

import "encoding/json"

// Event is something that happened to a workflow run. Machine.Apply consumes
// one event at a time.
type Event interface{ isEvent() }

// RunStarted: a new run was created. StartNodeID optionally overrides the
// definition's entry node.
type RunStarted struct {
    StartNodeID string
}

// NodeCompleted: a worker finished a task attempt successfully.
type NodeCompleted struct {
    TaskID  string
    NodeID  string
    Attempt int
    Output  json.RawMessage
}

// NodeFailed: a worker finished a task attempt with an error (or gave up on
// it, e.g. after too many deliveries).
type NodeFailed struct {
    TaskID  string
    NodeID  string
    Attempt int
    Error   string
}

// TimerFired: a waiting node's wake-up time passed — a WAIT node's timer, or
// a WAIT_FOR_EVENT node's timeout.
type TimerFired struct {
    TaskID string
    NodeID string
}

// SignalReceived: an external event (signal) was sent to the run.
type SignalReceived struct {
    Name    string
    Payload json.RawMessage
}

// RunRetried: the run is restarted from a node, on the definition version the
// machine was built with, optionally with new input or execution options.
type RunRetried struct {
    FromNodeID string
    Version    int
    Input      json.RawMessage
    Config     json.RawMessage
}

// RunCancelled: the run was cancelled by a user.
type RunCancelled struct {
    Reason string
}

func (RunStarted) isEvent()     {}
func (NodeCompleted) isEvent()  {}
func (NodeFailed) isEvent()     {}
func (TimerFired) isEvent()     {}
func (SignalReceived) isEvent() {}
func (RunRetried) isEvent()     {}
func (RunCancelled) isEvent()   {}

The private isEvent() method closes the set: only types in this package can be events. External events sent to WAIT_FOR_EVENT nodes are called signals, to keep them apart from the machine's own events.

Step 2: Define the run snapshot

The machine decides against a snapshot of the run. Keep it minimal: per node, only what routing needs. Create pkg/statemachine/state.go:

// pkg/statemachine/state.go
package statemachine

// Run is the snapshot of one workflow run that the machine decides on. The
// machine updates it as it emits actions, so after Apply it reflects the
// state the persisted actions produce.
type Run struct {
    ID              string
    WorkflowID      string
    WorkflowVersion int
    Status          model.WorkflowInstanceStatus
    // Context holds the run's input and execution options. Its Tasks map is a
    // cache of outputs loaded on demand through the Store.
    Context *model.RunContext
    // Nodes holds every node that has been reached, i.e. that has a task.
    Nodes map[string]*NodeState
}

// NodeState is the state of a reached node: its task's current status.
type NodeState struct {
    TaskID  string
    Status  model.TaskInstanceStatus
    Attempt int
    // Branches lists the successors a completed SWITCH selected.
    Branches []string
    // AwaitedSignal is the signal a WAITING WAIT_FOR_EVENT node waits for.
    AwaitedSignal string
}

// Signal is an external event buffered for a run before any node waited for it.
type Signal struct {
    ID      string
    Payload json.RawMessage
}

// NewRun returns the snapshot of a run that has just been created.
func NewRun(id, workflowID string, version int, ctx *model.RunContext) *Run {
    return &Run{
        ID:              id,
        WorkflowID:      workflowID,
        WorkflowVersion: version,
        Status:          model.WorkflowInstanceStatusRunning,
        Context:         ctx,
        Nodes:           map[string]*NodeState{},
    }
}

Notice that NodeState has no input or output. Outputs are fetched only when a template needs them (you set that up on the previous page with TaskRefs).

Add the small predicates the machine uses everywhere:

// isTaskActive: the node's task is with (or on its way to) a worker.
func (n *NodeState) isTaskActive() bool {
    switch n.Status {
    case model.TaskInstanceStatusPending, model.TaskInstanceStatusQueued, model.TaskInstanceStatusRunning:
        return true
    }
    return false
}

// isActive: the node still has to finish before the run can.
func (n *NodeState) isActive() bool {
    return n.isTaskActive() || n.Status == model.TaskInstanceStatusWaiting
}

// matches reports whether a result is for this node's current task attempt.
func (n *NodeState) matches(taskID string, attempt int) bool {
    return n != nil && n.TaskID == taskID && n.Attempt == attempt
}

func (r *Run) counts() (active, failed int) {
    for _, n := range r.Nodes {
        switch {
        case n.isActive():
            active++
        case n.Status == model.TaskInstanceStatusFailed:
            failed++
        }
    }
    return active, failed
}

matches is the fencing check from the concepts page, in three lines.

Step 3: Define the actions

An action is a change that must be saved. Each action also knows how to update the in-memory snapshot, so the machine can keep deciding within the same event (a FORK creates two tasks, and the second decision must see the first).

Create pkg/statemachine/actions.go:

// pkg/statemachine/actions.go
package statemachine

// Action is a state change the machine decided on. The shell persists actions
// in order; apply evolves the in-memory snapshot the same way.
type Action interface{ apply(*Run) }

// Transition is the outcome of applying one event.
type Transition struct {
    Actions []Action
    // Ignored explains why the event changed nothing, e.g. a duplicate or a
    // stale report from a superseded task attempt.
    Ignored string
    // SignalDelivered reports, for SignalReceived, whether a waiting node
    // consumed the signal (false: it was buffered).
    SignalDelivered bool
}

The full set of actions, grouped by purpose:

// CreateTask records a reached node as a task in the given state: PENDING
// (followed by DispatchTask), WAITING, or already resolved (COMPLETED for
// control nodes, SKIPPED, FAILED when its input cannot be resolved).
type CreateTask struct {
    TaskID        string
    Node          *model.Node
    Status        model.TaskInstanceStatus
    Input         json.RawMessage
    Output        json.RawMessage
    Error         string
    Branches      []string   // SWITCH decision
    WakeAt        *time.Time // WAITING: timer or timeout
    AwaitedSignal string     // WAITING on WAIT_FOR_EVENT
}

// DispatchTask sends a PENDING task attempt to a worker, no earlier than NotBefore.
type DispatchTask struct {
    TaskID    string
    Node      *model.Node
    Attempt   int
    Input     json.RawMessage
    NotBefore time.Time
}

// CompleteTask: an active task attempt finished successfully.
type CompleteTask struct {
    TaskID  string
    NodeID  string
    Attempt int
    Output  json.RawMessage
}

// FailTask: an active task attempt failed for good.
type FailTask struct {
    TaskID  string
    NodeID  string
    Attempt int
    Reason  string
}

// RetryTask: a failed attempt is retried; the task returns to PENDING with
// Attempt+1 and is re-dispatched (with its stored input) at NotBefore.
type RetryTask struct {
    TaskID    string
    NodeID    string
    Attempt   int // the attempt that failed
    Reason    string
    NotBefore time.Time
}

// ResolveWait: a WAITING node completed (timer fired, signal arrived) or
// failed (event timeout).
type ResolveWait struct {
    TaskID string
    NodeID string
    Status model.TaskInstanceStatus
    Output json.RawMessage
    Reason string
}

// BufferSignal keeps a signal no node is waiting for yet.
type BufferSignal struct {
    Name    string
    Payload json.RawMessage
}

// ConsumeSignal marks a buffered signal as used by a WAIT_FOR_EVENT node.
type ConsumeSignal struct {
    SignalID string
}

// ReopenRun puts a finished run back to RUNNING for a retry: the reset nodes'
// tasks are deleted and the run moves to Version with the given context.
type ReopenRun struct {
    Version    int
    Context    *model.RunContext
    ResetNodes []string
}

// CancelTasks fails every task that is still active.
type CancelTasks struct {
    Reason string
}

// FinishRun moves the run to a terminal status.
type FinishRun struct {
    Status model.WorkflowInstanceStatus
    Reason string
}

And how each one evolves the snapshot:

func (a CreateTask) apply(r *Run) {
    r.Nodes[a.Node.NodeID] = &NodeState{
        TaskID: a.TaskID, Status: a.Status, Branches: a.Branches, AwaitedSignal: a.AwaitedSignal,
    }
}

func (DispatchTask) apply(*Run) {}

func (a CompleteTask) apply(r *Run) { r.Nodes[a.NodeID].Status = model.TaskInstanceStatusCompleted }

func (a FailTask) apply(r *Run) { r.Nodes[a.NodeID].Status = model.TaskInstanceStatusFailed }

func (a RetryTask) apply(r *Run) {
    n := r.Nodes[a.NodeID]
    n.Status, n.Attempt = model.TaskInstanceStatusPending, a.Attempt+1
}

func (a ResolveWait) apply(r *Run) {
    n := r.Nodes[a.NodeID]
    n.Status, n.AwaitedSignal = a.Status, ""
}

func (BufferSignal) apply(*Run) {}

func (ConsumeSignal) apply(*Run) {}

func (a ReopenRun) apply(r *Run) {
    tasks := r.Context.Tasks
    for _, id := range a.ResetNodes {
        delete(r.Nodes, id)
        delete(tasks, id)
    }
    r.Context = a.Context
    r.Context.Tasks = tasks
    r.WorkflowVersion = a.Version
    r.Status = model.WorkflowInstanceStatusRunning
}

func (CancelTasks) apply(r *Run) {
    for _, n := range r.Nodes {
        if n.isActive() {
            n.Status = model.TaskInstanceStatusFailed
        }
    }
}

func (a FinishRun) apply(r *Run) { r.Status = a.Status }

DispatchTask, BufferSignal and ConsumeSignal change nothing the machine routes on, so their apply is empty. They exist only for the shell to persist.

Step 4: The machine and its dependencies

The machine needs four things from the outside world. Instead of importing them, it declares them. Create pkg/statemachine/machine.go:

// pkg/statemachine/machine.go
package statemachine

var (
    // ErrInvalidRequest: the event cannot apply to this run or definition
    // (unknown node, missing signal name, unresolvable start node, …).
    ErrInvalidRequest = errors.New("invalid request")
    // ErrRunFinished: the run is already terminal and cannot take the event.
    ErrRunFinished = errors.New("workflow run is not active")
    // ErrRunActive: a retry targets a run that still has tasks in flight.
    ErrRunActive = errors.New("workflow run still has active tasks")
)

// StoreError wraps a failure of the Store. Unlike the machine's own errors it
// is transient: the caller should retry the event rather than give up.
type StoreError struct{ Err error }

func (e *StoreError) Error() string { return "statemachine store: " + e.Err.Error() }
func (e *StoreError) Unwrap() error { return e.Err }

// TemplateResolver evaluates the CEL templates in node specs and conditions.
type TemplateResolver interface {
    ResolveTemplates(spec any, ctx *model.RunContext) (any, error)
}

// Store gives the machine read access to run data it only loads on demand.
type Store interface {
    // Outputs returns the outputs of the given completed nodes.
    Outputs(nodeIDs []string) (map[string]json.RawMessage, error)
    // PendingSignal returns the oldest buffered signal with that name, if any.
    PendingSignal(name string) (*Signal, error)
}

// Deps are the machine's collaborators.
type Deps struct {
    Templates TemplateResolver
    Store     Store
    Now       func() time.Time
    NewID     func() string
}

Now and NewID are injected so a test can fix the clock and get predictable task IDs. Store is read-only: the machine may ask for data, never write it.

Two kinds of error matter. Machine errors (ErrInvalidRequest, …) are permanent — retrying the same event gives the same answer. A *StoreError is transient — the database hiccupped, try again.

// Machine decides how one event moves one run forward. Create a Machine per
// event; it is not safe for concurrent use, and the caller must serialize
// events per run (the coordinator holds the run's row lock).
type Machine struct {
    def  *model.WorkflowDAG
    run  *Run
    deps Deps

    tr     *Transition
    now    time.Time
    loaded map[string]bool // nodes whose output is in run.Context.Tasks
}

// New returns a machine for run, governed by its workflow definition.
func New(def *model.WorkflowDAG, run *Run, deps Deps) *Machine {
    if run.Context.Tasks == nil {
        run.Context.Tasks = map[string]*model.StepResult{}
    }
    loaded := make(map[string]bool, len(run.Context.Tasks))
    for id := range run.Context.Tasks {
        loaded[id] = true
    }
    return &Machine{def: def, run: run, deps: deps, loaded: loaded}
}

Step 5: Apply — the single entry point

// Apply consumes one event and returns the resulting transition. Errors leave
// nothing to persist: ErrInvalidRequest, ErrRunFinished and ErrRunActive
// reject the event; a *StoreError is transient.
func (m *Machine) Apply(ev Event) (*Transition, error) {
    m.tr, m.now = &Transition{}, m.deps.Now()

    var err error
    switch ev := ev.(type) {
    case RunStarted:
        err = m.onRunStarted(ev)
    case NodeCompleted:
        err = m.onNodeCompleted(ev)
    case NodeFailed:
        err = m.onNodeFailed(ev)
    case TimerFired:
        err = m.onTimerFired(ev)
    case SignalReceived:
        err = m.onSignalReceived(ev)
    case RunRetried:
        err = m.onRunRetried(ev)
    case RunCancelled:
        err = m.onRunCancelled(ev)
    default:
        err = fmt.Errorf("%w: unknown event %T", ErrInvalidRequest, ev)
    }
    if err != nil {
        return nil, err
    }
    if m.tr.Ignored == "" {
        m.finishIfDone()
    }
    return m.tr, nil
}

// emit records an action and applies it to the snapshot.
func (m *Machine) emit(a Action) {
    m.tr.Actions = append(m.tr.Actions, a)
    a.apply(m.run)
}

func (m *Machine) ignore(format string, args ...any) error {
    m.tr.Ignored = fmt.Sprintf(format, args...)
    return nil
}

func (m *Machine) terminal() bool { return m.run.Status.IsTerminal() }

Every handler follows the same shape: check the event is still relevant, emit actions, and maybe route forward. After any handler, finishIfDone checks whether the run is over:

// finishIfDone finishes the run once no task is active: FAILED if any failed,
// SUCCEEDED otherwise.
func (m *Machine) finishIfDone() {
    if m.terminal() {
        return
    }
    active, failed := m.run.counts()
    switch {
    case active > 0:
    case failed > 0:
        m.emit(FinishRun{Status: model.WorkflowInstanceStatusFailed,
            Reason: fmt.Sprintf("workflow has %d failed task(s)", failed)})
    default:
        m.emit(FinishRun{Status: model.WorkflowInstanceStatusSucceeded})
    }
}

Run lifecycle

Note the order: a run fails only when nothing is active. If one branch fails while another is still running, the other branch finishes first. That keeps the run's final state consistent — no worker is still executing a task of a run that already reports "done".

Step 6: Handle task results (and fence stale ones)

func (m *Machine) onNodeCompleted(ev NodeCompleted) error {
    n := m.run.Nodes[ev.NodeID]
    if !n.matches(ev.TaskID, ev.Attempt) || !n.isTaskActive() {
        return m.ignore("stale or duplicate completion of %s attempt %d", ev.NodeID, ev.Attempt)
    }
    m.emit(CompleteTask{TaskID: ev.TaskID, NodeID: ev.NodeID, Attempt: ev.Attempt, Output: ev.Output})
    if m.terminal() {
        return nil // e.g. cancelled while running: keep the output, schedule nothing
    }
    return m.advanceFrom(ev.NodeID, ev.Output)
}

The first line rejects three cases with one check:

Report	Verdict
for attempt 0 while the task is on attempt 1	stale — ignored
for a task ID the node no longer has	stale — ignored
a second report for an already `COMPLETED` task	duplicate — ignored

Failures either retry or fail the task for good:

func (m *Machine) onNodeFailed(ev NodeFailed) error {
    n := m.run.Nodes[ev.NodeID]
    if !n.matches(ev.TaskID, ev.Attempt) || !n.isTaskActive() {
        return m.ignore("stale or duplicate failure of %s attempt %d", ev.NodeID, ev.Attempt)
    }
    if node := m.def.GetNode(ev.NodeID); node != nil && !m.terminal() && ev.Attempt+1 < node.MaxAttempts() {
        m.emit(RetryTask{TaskID: ev.TaskID, NodeID: ev.NodeID, Attempt: ev.Attempt, Reason: ev.Error,
            NotBefore: m.now.Add(node.RetryDelay(ev.Attempt + 1))})
        return nil
    }
    m.emit(FailTask{TaskID: ev.TaskID, NodeID: ev.NodeID, Attempt: ev.Attempt, Reason: ev.Error})
    return nil
}

Attempts are 0-based, so with max_attempts: 3 the attempts are 0, 1 and 2. A retry does not create a new task; it bumps Attempt on the same task. That bump is what makes any report for the old attempt stale.

Step 7: Route forward

When a node completes, its successors might become ready. Write the routing in pkg/statemachine/routing.go.

The one readiness rule

This is the rule that makes fork/join, diamonds and SWITCH re-convergence all work the same way:

A node becomes ready once every incoming edge is resolved. An edge is active when its source completed (and, for a SWITCH, selected this target). An edge is inactive when its source was skipped, or the SWITCH chose another branch. A ready node runs if at least one incoming edge is active, and is skipped otherwise.

Incoming edges	Readiness
active, unresolved	waiting (a predecessor is still running)
active, active	run
active, inactive	run (the SWITCH merge case)
inactive, inactive	skip (and the skip propagates)

type readiness int

const (
    nodeWaiting readiness = iota // some incoming edge is unresolved
    nodeRun                      // all edges resolved, at least one active
    nodeSkip                     // all edges resolved, none active
)

func (m *Machine) readiness(n *model.Node) readiness {
    anyActive := false
    for _, edge := range m.def.IncomingEdges(n.NodeID) {
        active, resolved := m.edgeState(edge)
        if !resolved {
            return nodeWaiting
        }
        anyActive = anyActive || active
    }
    if anyActive {
        return nodeRun
    }
    return nodeSkip
}

// edgeState reports whether an edge is resolved and, if so, whether it is active.
func (m *Machine) edgeState(edge *model.Edge) (active, resolved bool) {
    from := m.run.Nodes[edge.From]
    switch {
    case from == nil:
        return false, false
    case from.Status == model.TaskInstanceStatusSkipped:
        return false, true
    case !from.Status.IsCompleted():
        return false, false // pending, running, waiting or failed
    }
    if node := m.def.GetNode(edge.From); node == nil || node.Type != model.NodeTypeSwitch {
        return true, true
    }
    return slices.Contains(m.switchBranches(edge.From, from), edge.To), true
}

A failed source leaves its edge unresolved, so nothing downstream of a failure runs. The run then ends as FAILED once the other branches settle.

Propagate

propagate is a breadth-first walk from a set of candidate nodes. Skips cascade down untaken branches; control nodes resolve inline and keep walking; worker nodes and real waits stop the walk:

// propagate walks forward from candidates, running or skipping each node whose
// incoming edges are all resolved. Nodes still waiting on a predecessor are
// left alone; a later event re-examines them.
func (m *Machine) propagate(candidates []*model.Node) error {
    queue := slices.Clone(candidates)
    for len(queue) > 0 {
        n := queue[0]
        queue = queue[1:]
        if _, reached := m.run.Nodes[n.NodeID]; reached {
            continue
        }

        switch m.readiness(n) {
        case nodeWaiting:
            continue
        case nodeSkip:
            m.emit(CreateTask{TaskID: m.deps.NewID(), Node: n, Status: model.TaskInstanceStatusSkipped})
            queue = append(queue, m.def.Successors(n.NodeID)...)
        case nodeRun:
            next, err := m.activate(n)
            if err != nil {
                return err
            }
            queue = append(queue, next...)
        }
    }
    return nil
}

Trace it on the switch example from the model page, after p completes with {"n": 5}:

One propagation walk after p completes with {n: 5}

The walk, step by step:

queue=[s] — s is ready, the SWITCH completes inline with branches=[hi]; its successors hi and lo join the queue.
hi is ready (edge s→hi is active) — it is scheduled, and the walk stops on this branch.
lo has only an inactive edge — it is skipped, and lo2 joins the queue.
lo2 has only an inactive edge — skipped, and merge joins the queue.
merge still has an unresolved edge from hi — it waits.

Later, NodeCompleted(hi) makes hi→merge active and lo2→merge is already inactive, so merge runs.

Activate a ready node

// activate starts a ready node and returns the nodes to consider next:
// control nodes and waits that are already satisfied complete inline and pass
// routing on to their successors; worker nodes and real waits stop it.
func (m *Machine) activate(n *model.Node) ([]*model.Node, error) {
    switch {
    case n.Type == model.NodeTypeSwitch:
        return m.activateSwitch(n)
    case n.Type.IsControlFlow(): // FORK, JOIN
        m.emit(CreateTask{TaskID: m.deps.NewID(), Node: n, Status: model.TaskInstanceStatusCompleted,
            Output: json.RawMessage("{}")})
        return m.def.Successors(n.NodeID), nil
    case n.Type.IsWait():
        return m.park(n)
    default:
        return nil, m.schedule(n)
    }
}

A FORK is trivially "complete and continue to all successors". There is no special join logic anywhere: the readiness rule already makes a node with two incoming edges wait for both.

SWITCH: decide once, record the decision

// activateSwitch passes its predecessor's output through and records which
// branches its conditions select.
func (m *Machine) activateSwitch(n *model.Node) ([]*model.Node, error) {
    if err := m.ensureOutputs(n); err != nil {
        return nil, err
    }
    output := m.def.PredecessorOutput(n.NodeID, m.run.Context.Tasks)
    if output == nil {
        output = json.RawMessage("{}")
    }
    branches, err := m.selectBranches(n)
    if err != nil {
        return nil, fmt.Errorf("switch %s: %w", n.NodeID, err)
    }
    m.emit(CreateTask{TaskID: m.deps.NewID(), Node: n, Status: model.TaskInstanceStatusCompleted,
        Output: output, Branches: branches})
    m.setOutput(n.NodeID, output)
    return m.def.Successors(n.NodeID), nil
}

// selectBranches evaluates a SWITCH's edge conditions. Unconditional edges are
// always selected. The result is non-nil so an empty decision is still recorded.
func (m *Machine) selectBranches(n *model.Node) ([]string, error) {
    branches := []string{}
    for _, edge := range m.def.OutgoingEdges(n.NodeID) {
        if edge.Condition != "" {
            ok, err := m.evalCondition(edge.Condition)
            if err != nil {
                return nil, fmt.Errorf("condition eval for edge %s→%s: %w", edge.From, edge.To, err)
            }
            if !ok {
                continue
            }
        }
        branches = append(branches, edge.To)
    }
    return branches, nil
}

The selected branches are saved on the task (Branches, a TEXT[] column). Why save them instead of re-evaluating the conditions later? Because conditions can read system.now or data that changes. A join that runs an hour later must see the same decision the SWITCH made. Recording it makes the decision a fact.

// switchBranches returns the successors a completed SWITCH selected. Runs
// persisted before decisions were recorded fall back to re-evaluation.
func (m *Machine) switchBranches(nodeID string, s *NodeState) []string {
    if s.Branches != nil {
        return s.Branches
    }
    node := m.def.GetNode(nodeID)
    if err := m.ensureOutputs(node); err != nil {
        return nil
    }
    branches, err := m.selectBranches(node)
    if err != nil {
        return nil
    }
    return branches
}

Conditions are coerced to booleans in a forgiving way:

// evalCondition evaluates a CEL condition against the run context.
func (m *Machine) evalCondition(condition string) (bool, error) {
    val, err := m.deps.Templates.ResolveTemplates(condition, m.run.Context)
    if err != nil {
        return false, fmt.Errorf("failed to evaluate condition: %w", err)
    }
    switch v := val.(type) {
    case bool:
        return v, nil
    case string:
        return v != "", nil
    case float64:
        return v != 0, nil
    case nil:
        return false, nil
    default:
        return true, nil
    }
}

Schedule a worker node

// schedule creates a worker node's task and dispatches it. A node whose input
// cannot be resolved is recorded as FAILED instead.
func (m *Machine) schedule(n *model.Node) error {
    input, err := m.resolveInput(n)
    if err != nil {
        if isStoreError(err) {
            return err
        }
        m.emit(CreateTask{TaskID: m.deps.NewID(), Node: n, Status: model.TaskInstanceStatusFailed, Error: err.Error()})
        return nil
    }
    id := m.deps.NewID()
    m.emit(CreateTask{TaskID: id, Node: n, Status: model.TaskInstanceStatusPending, Input: input})
    m.emit(DispatchTask{TaskID: id, Node: n, Input: input, NotBefore: m.now})
    return nil
}

The input is resolved once, here, and stored on the task. A retry reuses the stored input. The worker receives plain JSON and never sees a template.

// resolveInput resolves the CEL templates in node.Spec. A *StoreError is
// transient; any other error means the node's templates cannot be resolved.
func (m *Machine) resolveInput(node *model.Node) (json.RawMessage, error) {
    if err := m.ensureOutputs(node); err != nil {
        return nil, err
    }
    var spec any
    if len(node.Spec) > 0 {
        if err := json.Unmarshal(node.Spec, &spec); err != nil {
            spec = string(node.Spec) // treat non-JSON specs as a plain string
        }
    }
    resolved, err := m.deps.Templates.ResolveTemplates(spec, m.run.Context)
    if err != nil {
        return nil, fmt.Errorf("CEL resolution for node %s: %w", node.NodeID, err)
    }
    out, err := json.Marshal(resolved)
    if err != nil {
        return nil, fmt.Errorf("marshal resolved input for node %s: %w", node.NodeID, err)
    }
    return out, nil
}

Load outputs only when needed

Back in machine.go, ensureOutputs uses the TaskRefs index you built on the previous page:

// setOutput caches a node's output for templates that reference it.
func (m *Machine) setOutput(nodeID string, output json.RawMessage) {
    m.run.Context.Tasks[nodeID] = &model.StepResult{Status: model.TaskInstanceStatusCompleted, Output: output}
    m.loaded[nodeID] = true
}

// ensureOutputs loads the outputs node's templates may read, if not loaded yet.
func (m *Machine) ensureOutputs(node *model.Node) error {
    refs, ok := m.def.TaskRefs(node.NodeID)
    if !ok {
        refs = make([]string, 0, len(m.run.Nodes))
        for id := range m.run.Nodes {
            refs = append(refs, id)
        }
    }
    var missing []string
    for _, id := range refs {
        n := m.run.Nodes[id]
        if !m.loaded[id] && n != nil && n.Status.IsCompleted() && !slices.Contains(missing, id) {
            missing = append(missing, id)
        }
    }
    if len(missing) == 0 {
        return nil
    }
    outputs, err := m.deps.Store.Outputs(missing)
    if err != nil {
        return &StoreError{Err: err}
    }
    for _, id := range missing {
        m.setOutput(id, outputs[id])
    }
    return nil
}

Starting and advancing

Two entry points use propagate. begin starts a node unconditionally (the start node, or a retry's restart node). advanceFrom routes past a node that just finished:

// begin activates node unconditionally — the start node, or the node a retry
// restarts from — and everything that becomes ready as a result.
func (m *Machine) begin(node *model.Node) error {
    next, err := m.activate(node)
    if err == nil {
        err = m.propagate(next)
    }
    if err != nil && !isStoreError(err) {
        return fmt.Errorf("%w: %v", ErrInvalidRequest, err)
    }
    return err
}

// advanceFrom routes past a node that completed with output. Routing errors
// are permanent (a bad SWITCH condition, a node missing from the definition),
// so they fail the run instead of being returned for endless retries.
func (m *Machine) advanceFrom(nodeID string, output json.RawMessage) error {
    m.setOutput(nodeID, output)
    if m.def.GetNode(nodeID) == nil {
        return m.failRun(fmt.Sprintf("node %s not found in workflow definition", nodeID))
    }
    if err := m.propagate(m.def.Successors(nodeID)); err != nil {
        if isStoreError(err) {
            return err
        }
        return m.failRun(fmt.Sprintf("routing after node %s failed: %v", nodeID, err))
    }
    return nil
}

func (m *Machine) failRun(reason string) error {
    m.emit(FinishRun{Status: model.WorkflowInstanceStatusFailed, Reason: reason})
    return nil
}

func isStoreError(err error) bool {
    var se *StoreError
    return errors.As(err, &se)
}

func (m *Machine) onRunStarted(ev RunStarted) error {
    start, err := m.def.StartNode(ev.StartNodeID)
    if err != nil {
        return fmt.Errorf("%w: %v", ErrInvalidRequest, err)
    }
    return m.begin(start)
}

The error handling in advanceFrom is deliberate. If a SWITCH condition is broken (tasks.a.output.missing.deeper), returning an error would make the worker's report fail, the worker would retry, and the same broken condition would fail again — forever. So a permanent routing error fails the run with a clear reason, while a transient StoreError is returned so the event is retried.

Step 8: Wait nodes

WAIT and WAIT_FOR_EVENT never reach a worker. They park as WAITING and are resolved by TimerFired or SignalReceived. Write them in pkg/statemachine/waits.go:

// pkg/statemachine/waits.go

// waitSpec is a wait node's resolved spec. Numbers may also be given as
// strings, so templated values like "{{input.delay}}" work.
type waitSpec struct {
    DurationSeconds json.RawMessage `json:"duration_seconds"`
    Until           string          `json:"until"` // RFC 3339
    Event           string          `json:"event"`
    TimeoutSeconds  json.RawMessage `json:"timeout_seconds"`
}

// park records a wait node as WAITING. A wait that is already satisfied — a
// zero duration, or a buffered signal — completes now and returns the node's
// successors so routing continues past it.
func (m *Machine) park(n *model.Node) ([]*model.Node, error) {
    input, err := m.resolveInput(n)
    if err != nil {
        if isStoreError(err) {
            return nil, err
        }
        return nil, m.failNode(n, nil, err.Error())
    }
    var spec waitSpec
    if err := json.Unmarshal(input, &spec); err != nil {
        return nil, m.failNode(n, input, fmt.Sprintf("invalid wait spec: %v", err))
    }

    task := CreateTask{TaskID: m.deps.NewID(), Node: n, Input: input, Status: model.TaskInstanceStatusWaiting}
    switch n.Type {
    case model.NodeTypeWait:
        wakeAt, err := spec.wakeTime(m.now)
        if err != nil {
            return nil, m.failNode(n, input, err.Error())
        }
        if !wakeAt.After(m.now) {
            return m.completeNow(task, wokeAt(m.now)), nil
        }
        task.WakeAt = &wakeAt

    case model.NodeTypeWaitForEvent:
        if spec.Event == "" {
            return nil, m.failNode(n, input, "WAIT_FOR_EVENT requires spec.event")
        }
        signal, err := m.deps.Store.PendingSignal(spec.Event)
        if err != nil {
            return nil, &StoreError{Err: err}
        }
        if signal != nil {
            m.emit(ConsumeSignal{SignalID: signal.ID})
            return m.completeNow(task, signalOutput(signal.Payload)), nil
        }
        timeout, err := parseSeconds(spec.TimeoutSeconds)
        if err != nil {
            return nil, m.failNode(n, input, fmt.Sprintf("timeout_seconds: %v", err))
        }
        if timeout > 0 {
            deadline := m.now.Add(timeout)
            task.WakeAt = &deadline
        }
        task.AwaitedSignal = spec.Event
    }

    m.emit(task)
    return nil, nil
}

// completeNow records a wait that is already satisfied and continues past it.
func (m *Machine) completeNow(task CreateTask, output json.RawMessage) []*model.Node {
    task.Status, task.Output = model.TaskInstanceStatusCompleted, output
    m.emit(task)
    m.setOutput(task.Node.NodeID, output)
    return m.def.Successors(task.Node.NodeID)
}

func (m *Machine) failNode(n *model.Node, input json.RawMessage, reason string) error {
    m.emit(CreateTask{TaskID: m.deps.NewID(), Node: n, Status: model.TaskInstanceStatusFailed, Input: input, Error: reason})
    return nil
}

Signals can arrive before the run reaches the waiting node (a webhook fires early). So a signal with no waiter is buffered, and park checks the buffer first:

Early signals are buffered, never lost

Both resolutions come in through events. A timer:

func (m *Machine) onTimerFired(ev TimerFired) error {
    n := m.run.Nodes[ev.NodeID]
    if n == nil || n.TaskID != ev.TaskID || n.Status != model.TaskInstanceStatusWaiting || m.terminal() {
        return m.ignore("timer for %s no longer pending", ev.NodeID)
    }
    if node := m.def.GetNode(ev.NodeID); node != nil && node.Type == model.NodeTypeWaitForEvent {
        m.emit(ResolveWait{TaskID: ev.TaskID, NodeID: ev.NodeID, Status: model.TaskInstanceStatusFailed,
            Reason: fmt.Sprintf("timed out waiting for event %q", n.AwaitedSignal)})
        return nil
    }
    output := wokeAt(m.now)
    m.emit(ResolveWait{TaskID: ev.TaskID, NodeID: ev.NodeID, Status: model.TaskInstanceStatusCompleted, Output: output})
    return m.advanceFrom(ev.NodeID, output)
}

A timer on a WAIT means "done waiting". A timer on a WAIT_FOR_EVENT means "the event never came" — a timeout, so the node fails. And a signal:

func (m *Machine) onSignalReceived(ev SignalReceived) error {
    if ev.Name == "" {
        return fmt.Errorf("%w: event name is required", ErrInvalidRequest)
    }
    if len(ev.Payload) > 0 && !json.Valid(ev.Payload) {
        return fmt.Errorf("%w: payload must be valid JSON", ErrInvalidRequest)
    }
    if m.terminal() {
        return fmt.Errorf("%w: run is %s", ErrRunFinished, m.run.Status)
    }

    nodeID, n := m.waiterFor(ev.Name)
    if n == nil {
        m.emit(BufferSignal{Name: ev.Name, Payload: ev.Payload})
        return nil
    }
    output := signalOutput(ev.Payload)
    m.emit(ResolveWait{TaskID: n.TaskID, NodeID: nodeID, Status: model.TaskInstanceStatusCompleted, Output: output})
    m.tr.SignalDelivered = true
    return m.advanceFrom(nodeID, output)
}

// waiterFor returns the oldest node waiting for the signal (task IDs are time-ordered).
func (m *Machine) waiterFor(name string) (string, *NodeState) {
    var bestID string
    var best *NodeState
    for id, n := range m.run.Nodes {
        if n.Status == model.TaskInstanceStatusWaiting && n.AwaitedSignal == name &&
            (best == nil || n.TaskID < best.TaskID) {
            bestID, best = id, n
        }
    }
    return bestID, best
}

Task IDs are UUIDv7, so comparing them as strings orders them by creation time. "Oldest waiter first" costs no extra column.

The small helpers at the bottom of waits.go:

// wakeTime returns when a WAIT node fires.
func (s waitSpec) wakeTime(now time.Time) (time.Time, error) {
    if s.Until != "" {
        t, err := time.Parse(time.RFC3339, s.Until)
        if err != nil {
            return time.Time{}, fmt.Errorf("until must be an RFC 3339 time: %w", err)
        }
        return t, nil
    }
    if len(s.DurationSeconds) == 0 {
        return time.Time{}, errors.New("WAIT requires spec.duration_seconds or spec.until")
    }
    d, err := parseSeconds(s.DurationSeconds)
    if err != nil {
        return time.Time{}, fmt.Errorf("duration_seconds: %w", err)
    }
    return now.Add(d), nil
}

// parseSeconds reads a JSON number or numeric string as a duration in seconds.
func parseSeconds(raw json.RawMessage) (time.Duration, error) {
    if len(raw) == 0 || string(raw) == "null" {
        return 0, nil
    }
    var s string
    if json.Unmarshal(raw, &s) == nil {
        raw = json.RawMessage(s)
    }
    f, err := strconv.ParseFloat(string(raw), 64)
    if err != nil || f < 0 {
        return 0, fmt.Errorf("must be a non-negative number, got %s", raw)
    }
    return time.Duration(f * float64(time.Second)), nil
}

func wokeAt(t time.Time) json.RawMessage {
    out, _ := json.Marshal(map[string]string{"woke_at": t.UTC().Format(time.RFC3339Nano)})
    return out
}

func signalOutput(payload json.RawMessage) json.RawMessage {
    if len(payload) == 0 || string(payload) == "null" {
        return json.RawMessage("{}")
    }
    return payload
}

Step 9: Retry and cancel a run

Back in machine.go, the two user-driven events:

func (m *Machine) onRunRetried(ev RunRetried) error {
    if active, _ := m.run.counts(); active > 0 {
        return fmt.Errorf("%w: %d task(s) in flight; cancel the run before retrying", ErrRunActive, active)
    }
    restart := m.def.GetNode(ev.FromNodeID)
    if restart == nil {
        return fmt.Errorf("%w: node %s not in workflow definition", ErrInvalidRequest, ev.FromNodeID)
    }

    ctx := m.run.Context.Persisted()
    if len(ev.Input) > 0 {
        ctx.Input = ev.Input
    }
    if len(ev.Config) > 0 {
        ctx.Config = ev.Config
    }
    reset := append(m.def.Descendants(ev.FromNodeID), ev.FromNodeID)
    for _, id := range reset {
        delete(m.loaded, id)
    }
    m.emit(ReopenRun{Version: ev.Version, Context: ctx, ResetNodes: reset})
    return m.begin(restart)
}

func (m *Machine) onRunCancelled(ev RunCancelled) error {
    if m.terminal() {
        return fmt.Errorf("%w: run is %s", ErrRunFinished, m.run.Status)
    }
    reason := ev.Reason
    if reason == "" {
        reason = "cancelled"
    }
    m.emit(CancelTasks{Reason: reason})
    m.emit(FinishRun{Status: model.WorkflowInstanceStatusFailed, Reason: reason})
    return nil
}

Retry resets the restart node and everything downstream of it

Retry refuses to run while tasks are in flight. Mixing a live old attempt with a fresh restart would make the run's history impossible to reason about, so the user must cancel first.

Step 10: Test every rule without a database

This is where the pure design pays off. Create pkg/statemachine/machine_test.go with a small harness that plays the coordinator's role:

// pkg/statemachine/machine_test.go
package statemachine_test

// harness plays the coordinator's role: it keeps the run snapshot between
// events, persists what actions write (outputs, buffered signals), and builds
// a fresh Machine per event with an empty output cache, so outputs are only
// reachable through the Store — exactly as after a reload from the database.
type harness struct {
    t       *testing.T
    def     *model.WorkflowDAG
    tmpl    *template.Resolver
    run     *sm.Run
    now     time.Time
    ids     int
    outputs map[string]json.RawMessage
    signals []bufferedSignal
    loads   [][]string
}

func (h *harness) try(ev sm.Event) (*sm.Transition, error) {
    h.run.Context.Tasks = nil // reloaded snapshot: no outputs cached
    m := sm.New(h.def, h.run, sm.Deps{
        Templates: h.tmpl,
        Store:     h,
        Now:       func() time.Time { return h.now },
        NewID:     func() string { h.ids++; return fmt.Sprintf("t%03d", h.ids) },
    })
    tr, err := m.Apply(ev)
    if err == nil {
        h.persist(tr)
    }
    return tr, err
}

The harness implements Store with two maps, uses a fixed clock, and generates IDs t001, t002, …. Add tiny helpers so each test reads like a story:

func (h *harness) start() *sm.Transition { return h.apply(sm.RunStarted{}) }

func (h *harness) complete(nodeID, output string) *sm.Transition {
    h.t.Helper()
    n := h.run.Nodes[nodeID]
    return h.apply(sm.NodeCompleted{TaskID: n.TaskID, NodeID: nodeID, Attempt: n.Attempt, Output: json.RawMessage(output)})
}

func task(id string) model.Node { return model.Node{NodeID: id, Type: model.NodeTypeHTTP} }
func fork(id string) model.Node { return model.Node{NodeID: id, Type: model.NodeTypeFork} }
func sw(id string) model.Node   { return model.Node{NodeID: id, Type: model.NodeTypeSwitch} }
func edge(from, to string) model.Edge { return model.Edge{From: from, To: to} }
func cond(from, to, c string) model.Edge {
    return model.Edge{From: from, To: to, Condition: c}
}

Now a fork/join with branches of different lengths:

func TestForkJoinWaitsForMultiNodeBranches(t *testing.T) {
    // fork → a1 → a2 ─┐
    //      → b1 ──────┴→ join
    h := newHarness(t,
        []model.Node{fork("f"), task("a1"), task("a2"), task("b1"), task("join")},
        []model.Edge{edge("f", "a1"), edge("a1", "a2"), edge("f", "b1"), edge("a2", "join"), edge("b1", "join")},
        `{}`)

    tr := h.start()
    expect(t, "fork completes inline", created(tr, model.TaskInstanceStatusCompleted), "f")
    expect(t, "branches", dispatched(tr), "a1", "b1")
    expect(t, "after a1", dispatched(h.complete("a1", `{}`)), "a2")
    expect(t, "after b1 (join waits for a2)", dispatched(h.complete("b1", `{}`)))
    expect(t, "after a2", dispatched(h.complete("a2", `{}`)), "join")
}

A SWITCH that skips a two-node branch and merges again:

func TestSwitchSkipsUntakenBranchAndReconverges(t *testing.T) {
    // p → s ─(n > 1)→ hi ─────────┐
    //       ─(n <= 1)→ lo → lo2 ──┴→ merge
    h := newHarness(t,
        []model.Node{task("p"), sw("s"), task("hi"), task("lo"), task("lo2"), task("merge")},
        []model.Edge{
            edge("p", "s"),
            cond("s", "hi", "{{tasks.p.output.n > 1}}"),
            cond("s", "lo", "{{tasks.p.output.n <= 1}}"),
            edge("lo", "lo2"), edge("hi", "merge"), edge("lo2", "merge"),
        }, `{}`)

    h.start()
    tr := h.complete("p", `{"n": 5}`)
    expect(t, "switch completes inline", created(tr, model.TaskInstanceStatusCompleted), "s")
    expect(t, "taken branch", dispatched(tr), "hi")
    expect(t, "skipped branch", created(tr, model.TaskInstanceStatusSkipped), "lo", "lo2")
    expect(t, "recorded decision", h.run.Nodes["s"].Branches, "hi")
    expect(t, "merge after taken branch", dispatched(h.complete("hi", `{}`)), "merge")
}

That the recorded decision wins over re-evaluation:

func TestJoinUsesRecordedSwitchDecision(t *testing.T) {
    h := newHarness(t,
        []model.Node{fork("f"), sw("s"), task("slow"), task("x"), task("j")},
        []model.Edge{edge("f", "s"), edge("f", "slow"), cond("s", "x", "{{input.flag}}"), edge("x", "j"), edge("slow", "j")},
        `{"flag": true}`)

    expect(t, "start", dispatched(h.start()), "slow", "x")
    h.run.Context.Input = json.RawMessage(`{"flag": false}`) // re-evaluation would now deselect x
    expect(t, "after slow", dispatched(h.complete("slow", `{}`)))
    expect(t, "after x", dispatched(h.complete("x", `{}`)), "j")
}

That outputs load lazily — only a is fetched for c, never the huge output of b:

func TestOutputsAreLoadedOnlyWhenReferenced(t *testing.T) {
    c := task("c")
    c.Spec = json.RawMessage(`{"url": "{{tasks.a.output.url}}"}`)
    h := newHarness(t, []model.Node{task("a"), task("b"), c},
        []model.Edge{edge("a", "b"), edge("b", "c")}, `{}`)

    h.start()
    h.complete("a", `{"url": "http://x"}`)
    h.loads = nil
    tr := h.complete("b", `{"huge": true}`)

    if len(h.loads) != 1 || !slices.Equal(h.loads[0], []string{"a"}) {
        t.Fatalf("expected exactly one load of [a], got %v", h.loads)
    }
    for _, a := range tr.Actions {
        if d, ok := a.(sm.DispatchTask); ok && string(d.Input) != `{"url":"http://x"}` {
            t.Fatalf("c's input not resolved from a: %s", d.Input)
        }
    }
}

And fencing:

func TestStaleAndDuplicateReportsAreIgnored(t *testing.T) {
    h := newHarness(t, []model.Node{task("a"), task("b")}, []model.Edge{edge("a", "b")}, `{}`)
    h.start()
    taskID := h.run.Nodes["a"].TaskID

    if tr := h.apply(sm.NodeCompleted{TaskID: taskID, NodeID: "a", Attempt: 3}); tr.Ignored == "" {
        t.Fatal("a report for another attempt must be ignored")
    }
    if tr := h.apply(sm.NodeCompleted{TaskID: "other", NodeID: "a"}); tr.Ignored == "" {
        t.Fatal("a report for another task must be ignored")
    }
    h.complete("a", `{}`)
    if tr := h.apply(sm.NodeCompleted{TaskID: taskID, NodeID: "a"}); tr.Ignored == "" || len(tr.Actions) != 0 {
        t.Fatal("a duplicate completion must change nothing")
    }
}

Run them:

go test ./pkg/statemachine/ -v

These tests run in milliseconds and cover every rule: fencing, retries, joins, skips, waits, signals, and finishing the run. You have a working workflow brain. It cannot remember anything yet — the next page gives it a database.

The Workflow State Machine — Events In, Actions Out

#Step 1: Define the events

#Step 2: Define the run snapshot

#Step 3: Define the actions

#Step 4: The machine and its dependencies

#Step 5: Apply — the single entry point

#Step 6: Handle task results (and fence stale ones)

#Step 7: Route forward

#The one readiness rule

#Propagate

#Activate a ready node

#SWITCH: decide once, record the decision

#Schedule a worker node

#Load outputs only when needed

#Starting and advancing

#Step 8: Wait nodes

#Step 9: Retry and cancel a run

#Step 10: Test every rule without a database