The Workflow Model — Nodes, Edges, the Indexed DAG and CEL Templates

Start at the bottom: the data types every other package uses. In this page you write pkg/model (what a workflow is) and pkg/template (how a node reads data from earlier nodes). Neither package does any I/O.

#Step 1: Write a workflow in YAML first

Before you write Go, write down what users will type. Design the data model to fit it.

nodes:
  - node_id: fetch_order          # unique id, used in templates as tasks.fetch_order
    name: Fetch order
    type: HTTP
    retry: {max_attempts: 3, backoff_seconds: 2}
    spec:
      url: "https://shop.example.com/orders/{{input.order_id}}"
      method: GET

  - node_id: is_large
    name: Large order?
    type: SWITCH

  - node_id: approval
    name: Manager approval
    type: WAIT_FOR_EVENT
    spec: {event: "approve-{{input.order_id}}", timeout_seconds: 86400}

  - node_id: ship
    name: Ship
    type: CODE
    runtime: python
    spec:
      code: |
        def handler(input):
            return {"shipped": True, "order": input["order"]}
      order: "{{tasks.fetch_order.output}}"

edges:
  - {from: fetch_order, to: is_large}
  - {from: is_large, to: approval, condition: "{{tasks.fetch_order.output.total > 1000}}"}
  - {from: is_large, to: ship,     condition: "{{tasks.fetch_order.output.total <= 1000}}"}
  - {from: approval, to: ship}

As a picture:

Small orders ship immediately. Large orders wait up to a day for an approval event, then ship. ship has two incoming edges, but only one is ever taken; the other branch is skipped, so ship runs exactly once either way. You will build that rule on the next page.

Three things to notice in the YAML:

• A node has an id, a type, and a free-form spec. The spec's shape depends on the type.
• Any string can contain {{ … }} templates that read input or the output of earlier tasks.
• Edges can carry a condition. Only edges leaving a SWITCH use it.

#Step 2: Statuses and node types

Create pkg/model/workflow_types.go. Start with run and task statuses:

// pkg/model/workflow_types.go
package model

// WorkflowInstanceStatus represents the status of a workflow instance
type WorkflowInstanceStatus string

const (
    WorkflowInstanceStatusPending   WorkflowInstanceStatus = "PENDING"
    WorkflowInstanceStatusRunning   WorkflowInstanceStatus = "RUNNING"
    WorkflowInstanceStatusSucceeded WorkflowInstanceStatus = "SUCCEEDED"
    WorkflowInstanceStatusCompleted WorkflowInstanceStatus = "COMPLETED"
    WorkflowInstanceStatusFailed    WorkflowInstanceStatus = "FAILED"
)

// IsTerminal reports whether no further nodes should be scheduled for the run.
func (s WorkflowInstanceStatus) IsTerminal() bool {
    switch s {
    case WorkflowInstanceStatusSucceeded, WorkflowInstanceStatusCompleted, WorkflowInstanceStatusFailed:
        return true
    }
    return false
}

// TaskInstanceStatus represents the status of a task instance
type TaskInstanceStatus string

const (
    TaskInstanceStatusPending   TaskInstanceStatus = "PENDING"
    TaskInstanceStatusQueued    TaskInstanceStatus = "QUEUED"
    TaskInstanceStatusRunning   TaskInstanceStatus = "RUNNING"
    TaskInstanceStatusSucceeded TaskInstanceStatus = "SUCCEEDED"
    TaskInstanceStatusCompleted TaskInstanceStatus = "COMPLETED"
    TaskInstanceStatusFailed    TaskInstanceStatus = "FAILED"
    // TaskInstanceStatusSkipped marks a node on a branch that was not taken
    // (an unselected SWITCH edge, or downstream of one).
    TaskInstanceStatusSkipped TaskInstanceStatus = "SKIPPED"
    // TaskInstanceStatusWaiting marks a WAIT or WAIT_FOR_EVENT node parked
    // until its timer fires or its event arrives.
    TaskInstanceStatusWaiting TaskInstanceStatus = "WAITING"
)

// IsCompleted reports whether the task finished successfully.
func (s TaskInstanceStatus) IsCompleted() bool {
    return s == TaskInstanceStatusCompleted || s == TaskInstanceStatusSucceeded
}

The task lifecycle you are encoding:

Now the node types, in the same file:

// NodeType represents the type of a workflow node
type NodeType string

const (
    // CODE - Execute code in a language runtime (requires runtime field)
    NodeTypeCode NodeType = "CODE"
    // HTTP - Execute HTTP request
    NodeTypeHTTP NodeType = "HTTP"
    // SWITCH - Conditional branching based on conditions
    NodeTypeSwitch NodeType = "SWITCH"
    // FORK - Fork execution into parallel branches
    NodeTypeFork NodeType = "FORK"
    // JOIN - Explicit join point; any node with several incoming edges already
    // waits for all of them, so JOIN is a no-op pass-through.
    NodeTypeJoin NodeType = "JOIN"
    // WAIT - Pause the branch for a duration or until a point in time
    NodeTypeWait NodeType = "WAIT"
    // WAIT_FOR_EVENT - Pause the branch until a named event is signalled to the run
    NodeTypeWaitForEvent NodeType = "WAIT_FOR_EVENT"
)

// IsControlFlow reports whether the coordinator resolves this node inline
// instead of dispatching it to a worker.
func (t NodeType) IsControlFlow() bool {
    return t == NodeTypeSwitch || t == NodeTypeFork || t == NodeTypeJoin
}

// IsWait reports whether the node parks in WAITING until a timer or event
// resolves it, without ever being dispatched to a worker.
func (t NodeType) IsWait() bool {
    return t == NodeTypeWait || t == NodeTypeWaitForEvent
}

IsControlFlow and IsWait are how the state machine decides between "send to a worker", "resolve now" and "park".

#Step 3: Nodes, edges and the definition spec

Create pkg/model/workflow_definition.go. The spec is exactly the YAML from Step 1, converted to JSON and stored in a JSONB column:

// pkg/model/workflow_definition.go
package model

// WorkflowDefinitionSpec represents the workflow definition structure (the spec JSON)
// This is what gets stored in the spec_json column
// Note: workflow_id and name are stored in the workflow_definitions table, not in the spec
type WorkflowDefinitionSpec struct {
    Nodes []Node `json:"nodes"`
    Edges []Edge `json:"edges"`
}

// Node represents a node in the workflow DAG
type Node struct {
    NodeID  string          `json:"node_id"`
    Name    string          `json:"name"`
    Type    NodeType        `json:"type"`              // CODE, HTTP, SWITCH, FORK, JOIN, etc
    Runtime *string         `json:"runtime,omitempty"` // Optional: python/js/node (only for CODE type nodes)
    Spec    json.RawMessage `json:"spec"`              // task implementation definition (http spec, code reference etc)
    Retry   *RetryPolicy    `json:"retry,omitempty"`   // Optional: retry failed executions of this node
    // TaskQueue optionally routes the node to a dedicated worker pool. By
    // default a node goes to the queue of its runtime ("http", "python", "js").
    TaskQueue string `json:"queue,omitempty"`
}

// Edge represents an edge in the workflow DAG
type Edge struct {
    From      string `json:"from"`
    To        string `json:"to"`
    Condition string `json:"condition,omitempty"` // CEL expression for conditional edges (used in SWITCH nodes)
}

Spec stays a json.RawMessage. The model does not need to understand an HTTP spec or a code spec — the worker's executor does. Keeping it raw means you can add new node types without touching this struct.

#Routing a node to a queue

Workers subscribe to queues. A node goes to the queue of its runtime unless it names one:

// Queue returns the task queue a worker node is dispatched on.
func (n *Node) Queue() string {
    if n.TaskQueue != "" {
        return strings.ToLower(n.TaskQueue)
    }
    key := strings.ToLower(n.GetNodeTypeForRouting())
    switch key {
    case "py":
        return "python"
    case "javascript", "node":
        return "js"
    }
    return key
}

// GetNodeTypeForRouting returns the node type string to use for routing to executors
// For CODE nodes, returns the runtime (python/js/node)
// For other nodes, returns the type string (HTTP/SWITCH/etc)
func (n *Node) GetNodeTypeForRouting() string {
    if n.Type == NodeTypeCode {
        // For CODE nodes, use runtime
        if n.Runtime != nil && *n.Runtime != "" {
            return *n.Runtime
        }
        // Fallback to "python" if runtime not specified
        return "python"
    }
    return string(n.Type)
}

Node	Queue
HTTP node	http
CODE node, runtime: python	python
CODE node, runtime: node	js
any node with queue: partner	partner (a dedicated worker pool)

Dedicated queues let you isolate a slow partner API onto its own workers, so it cannot starve everything else.

#Retry policy

// RetryPolicy controls how often a failed worker node is re-executed.
type RetryPolicy struct {
    // MaxAttempts is the total number of executions, including the first.
    MaxAttempts int `json:"max_attempts"`
    // BackoffSeconds is the base delay; attempt n waits BackoffSeconds * 2^(n-1).
    BackoffSeconds int `json:"backoff_seconds,omitempty"`
}

const maxRetryBackoff = 10 * time.Minute

// MaxAttempts returns how many times the node may execute (at least 1).
func (n *Node) MaxAttempts() int {
    if n.Retry == nil || n.Retry.MaxAttempts < 1 {
        return 1
    }
    return n.Retry.MaxAttempts
}

// RetryDelay returns the backoff before the given retry attempt (1-based).
func (n *Node) RetryDelay(attempt int) time.Duration {
    if n.Retry == nil || n.Retry.BackoffSeconds <= 0 || attempt < 1 {
        return 0
    }
    delay := time.Duration(n.Retry.BackoffSeconds) * time.Second << (attempt - 1)
    if delay <= 0 || delay > maxRetryBackoff {
        return maxRetryBackoff
    }
    return delay
}

With backoff_seconds: 5 the waits are 5 s, 10 s, 20 s, 40 s … capped at 10 minutes. The delay <= 0 check catches overflow from a large shift.

#Step 4: Index the graph

The state machine asks graph questions constantly: "what are the successors of a?", "what edges enter join?". Scanning the edge list each time is O(edges). Build the indexes once per definition version and cache the result.

Create pkg/model/dag.go:

// pkg/model/dag.go
package model

// WorkflowDAG is an indexed, read-only view of a WorkflowDefinitionSpec.
// Pre-built adjacency maps make all graph queries O(1) or O(degree) instead of O(nodes+edges).
type WorkflowDAG struct {
    nodes     map[string]*Node
    outgoing  map[string][]*Edge // nodeID → outgoing edges
    incoming  map[string][]*Edge // nodeID → incoming edges
    entryNode *Node              // first node (in spec order) with no incoming edges
    // taskRefs lists, per node, the task outputs its templates can read; a
    // missing entry means the references could not be determined statically.
    taskRefs map[string][]string
}

BuildDAG fills the maps:

// BuildDAG constructs a WorkflowDAG from a spec, pre-building all index structures.
func BuildDAG(spec *WorkflowDefinitionSpec) *WorkflowDAG {
    dag := &WorkflowDAG{
        nodes:    make(map[string]*Node, len(spec.Nodes)),
        outgoing: make(map[string][]*Edge, len(spec.Nodes)),
        incoming: make(map[string][]*Edge, len(spec.Nodes)),
    }

    for i := range spec.Nodes {
        dag.nodes[spec.Nodes[i].NodeID] = &spec.Nodes[i]
    }

    for i := range spec.Edges {
        e := &spec.Edges[i]
        dag.outgoing[e.From] = append(dag.outgoing[e.From], e)
        dag.incoming[e.To] = append(dag.incoming[e.To], e)
    }

    for i := range spec.Nodes {
        if len(dag.incoming[spec.Nodes[i].NodeID]) == 0 {
            dag.entryNode = &spec.Nodes[i]
            break
        }
    }

    dag.taskRefs = make(map[string][]string, len(spec.Nodes))
    for id, n := range dag.nodes {
        texts := jsonStrings(n.Spec)
        var implicit []string
        if n.Type == NodeTypeSwitch {
            // A SWITCH reads its conditions and passes a predecessor's output through.
            for _, e := range dag.outgoing[id] {
                texts = append(texts, e.Condition)
            }
            for _, e := range dag.incoming[id] {
                implicit = append(implicit, e.From)
            }
        }
        if refs, ok := templateTaskRefs(texts...); ok {
            dag.taskRefs[id] = append(refs, implicit...)
        }
    }

    return dag
}

The last loop needs an explanation. That is the next step.

#Step 5: Know which outputs a node reads

A run with 200 nodes can have megabytes of task outputs. When node c is about to run, you do not want to load all of them from the database just to resolve {{tasks.a.output.url}}.

So, at build time, scan every template in the node's spec and record which task IDs it mentions. At run time, load only those outputs.

// TaskRefs returns the node IDs whose outputs nodeID's templates may read.
// ok is false when they cannot be determined, meaning every output may be read.
func (d *WorkflowDAG) TaskRefs(nodeID string) (refs []string, ok bool) {
    refs, ok = d.taskRefs[nodeID]
    return refs, ok
}

var (
    templateExprRe = regexp.MustCompile(`{{\s*([^}]+)\s*}}`)
    tasksIdentRe   = regexp.MustCompile(`\btasks\b`)
    taskRefRe      = regexp.MustCompile(`\btasks\s*(?:\.\s*([A-Za-z_][A-Za-z0-9_]*)|\[\s*["']([^"']+)["']\s*\])`)
)

// templateTaskRefs extracts tasks.<id> and tasks["<id>"] references from the
// {{ }} expressions in texts. It reports false if any expression uses tasks in
// another way (e.g. tasks[someVar]), which could read any output.
func templateTaskRefs(texts ...string) ([]string, bool) {
    refs := []string{}
    for _, text := range texts {
        for _, expr := range templateExprRe.FindAllStringSubmatch(text, -1) {
            body := expr[1]
            matches := taskRefRe.FindAllStringSubmatch(body, -1)
            if len(matches) != len(tasksIdentRe.FindAllStringIndex(body, -1)) {
                return nil, false
            }
            for _, m := range matches {
                if m[1] != "" {
                    refs = append(refs, m[1])
                } else {
                    refs = append(refs, m[2])
                }
            }
        }
    }
    return refs, true
}

The safety trick is the count comparison. Every occurrence of the word tasks must be a static reference (tasks.x or tasks["x"]). If someone writes tasks[input.key], the counts differ, the function returns false, and the machine falls back to loading every completed output. Correct first, fast when possible.

jsonStrings walks the spec's JSON and collects every string (keys too), so templates nested deep in a body are found:

// jsonStrings returns every string value in a JSON document (so escaped quotes
// are decoded), or the raw text if it is not valid JSON.
func jsonStrings(raw json.RawMessage) []string {
    var v any
    if len(raw) == 0 {
        return nil
    }
    if err := json.Unmarshal(raw, &v); err != nil {
        return []string{string(raw)}
    }
    var out []string
    var walk func(any)
    walk = func(v any) {
        switch t := v.(type) {
        case string:
            out = append(out, t)
        case map[string]any:
            for k, val := range t {
                out = append(out, k)
                walk(val)
            }
        case []any:
            for _, val := range t {
                walk(val)
            }
        }
    }
    walk(v)
    return out
}

#Step 6: Graph queries

The rest of dag.go is small, fast lookups the state machine uses:

// GetNode returns the node with the given ID, or nil if not found.
func (d *WorkflowDAG) GetNode(id string) *Node {
    return d.nodes[id]
}

// StartNode returns the workflow entry node.
// If hint is non-empty, returns that specific node; otherwise returns the node with no incoming edges.
func (d *WorkflowDAG) StartNode(hint string) (*Node, error) {
    if hint != "" {
        n := d.nodes[hint]
        if n == nil {
            return nil, fmt.Errorf("start node %s not found in workflow definition", hint)
        }
        return n, nil
    }
    if d.entryNode != nil {
        return d.entryNode, nil
    }
    return nil, fmt.Errorf("no start node found: ensure workflow has a node with no incoming edges")
}

// Successors returns all nodes directly reachable from nodeID.
func (d *WorkflowDAG) Successors(nodeID string) []*Node {
    edges := d.outgoing[nodeID]
    nodes := make([]*Node, 0, len(edges))
    for _, e := range edges {
        if n := d.nodes[e.To]; n != nil {
            nodes = append(nodes, n)
        }
    }
    return nodes
}

// OutgoingEdges returns the edges leaving nodeID.
func (d *WorkflowDAG) OutgoingEdges(nodeID string) []*Edge {
    return d.outgoing[nodeID]
}

// IncomingEdges returns the edges entering nodeID.
func (d *WorkflowDAG) IncomingEdges(nodeID string) []*Edge {
    return d.incoming[nodeID]
}

Two more are used by SWITCH and by retries:

// PredecessorOutput returns the output of the first completed predecessor of nodeID.
// Used by SWITCH nodes which pass through their predecessor's output.
func (d *WorkflowDAG) PredecessorOutput(nodeID string, tasks map[string]*StepResult) json.RawMessage {
    for _, e := range d.incoming[nodeID] {
        if result, ok := tasks[e.From]; ok && result != nil &&
            result.Status.IsCompleted() && len(result.Output) > 0 {
            return result.Output
        }
    }
    return nil
}

// Descendants returns all node IDs reachable from nodeID via BFS (nodeID itself is excluded).
func (d *WorkflowDAG) Descendants(nodeID string) []string {
    visited := map[string]bool{nodeID: true}
    queue := []string{nodeID}
    var result []string
    for len(queue) > 0 {
        cur := queue[0]
        queue = queue[1:]
        for _, e := range d.outgoing[cur] {
            if !visited[e.To] {
                visited[e.To] = true
                queue = append(queue, e.To)
                result = append(result, e.To)
            }
        }
    }
    return result
}

Descendants powers "retry from node X": X and everything downstream of it are reset.

#Step 7: The run context

Templates read from a run context. Create pkg/model/run_context.go:

// pkg/model/run_context.go
package model

type StepResult struct {
    Status TaskInstanceStatus `json:"status"`
    Output json.RawMessage    `json:"output,omitempty"`
    Err    error              `json:"-"`
}

// RunContext captures workflow-level data available to node templates.
// Only Input and Config are persisted on the workflow run; Tasks is filled on
// demand from task_instances so a run's row never grows with its outputs.
type RunContext struct {
    Input  json.RawMessage        `json:"input"`
    Config json.RawMessage        `json:"config,omitempty"`
    Tasks  map[string]*StepResult `json:"tasks,omitempty"`
}

// NewRunContext returns an empty context for a new run.
func NewRunContext(input, config json.RawMessage) *RunContext {
    return &RunContext{
        Input:  input,
        Config: config,
        Tasks:  map[string]*StepResult{},
    }
}

// Persisted returns the part of the context stored in workflow_runs.context_json.
func (rc *RunContext) Persisted() *RunContext {
    return &RunContext{Input: rc.Input, Config: rc.Config}
}

The important design choice is in the comment: task outputs never go into the run row. If they did, every completed task would rewrite an ever-growing JSON blob on the run, and the hottest row in the system would grow with the workflow. Outputs live on their own task rows; Tasks is only an in-memory cache.

#Step 8: Resolve templates with CEL

Create pkg/template/resolver.go. A template is {{ expression }}, and the expression is CEL. Declare the four variables a template may use:

// pkg/template/resolver.go
package template

// Resolver resolves template expressions in workflow parameters using CEL (Common Expression Language).
// It supports expressions like {{ input.field }} or {{ tasks.task_id.output }}.
type Resolver struct {
    celEnv     *cel.Env
    templateRe *regexp.Regexp
}

// NewResolver creates a new template resolver with CEL environment configured.
func NewResolver() (*Resolver, error) {
    r := &Resolver{
        templateRe: regexp.MustCompile(`{{\s*([^}]+)\s*}}`),
    }
    env, err := cel.NewEnv(
        cel.Declarations(
            decls.NewVar("input", decls.Dyn),
            decls.NewVar("config", decls.Dyn),
            decls.NewVar("tasks", decls.Dyn),
            decls.NewVar("system", decls.Dyn),
        ),
    )
    if err != nil {
        return nil, err
    }
    r.celEnv = env
    return r, nil
}

Variable	Contents
input	the run's input
config	the run's execution_options
tasks.<node_id>.output / .status	a completed node's output and status
system.now	current time (RFC 3339)

ResolveTemplates walks any JSON-like value and resolves every string inside it:

// ResolveTemplates recursively resolves template expressions in the given node.
// It handles strings, maps, arrays, and other types.
// Template expressions use the format {{ expression }} where expression is a CEL expression.
func (r *Resolver) ResolveTemplates(node interface{}, rc *model.RunContext) (interface{}, error) {
    switch v := node.(type) {
    case string:
        return r.resolveStringTemplates(v, rc)
    case map[string]interface{}:
        m2 := make(map[string]interface{}, len(v))
        for k, vv := range v {
            res, err := r.ResolveTemplates(vv, rc)
            if err != nil {
                return nil, err
            }
            m2[k] = res
        }
        return m2, nil
    case []interface{}:
        arr := make([]interface{}, len(v))
        for i, vv := range v {
            res, err := r.ResolveTemplates(vv, rc)
            if err != nil {
                return nil, err
            }
            arr[i] = res
        }
        return arr, nil
    default:
        return v, nil
    }
}

One rule makes templates pleasant to use: a string that is exactly one template keeps the value's type.

  "{{input.count}}"           → 42                (a number, not "42")
  "{{tasks.fetch.output}}"    → {"total": 1200}   (an object)
  "order {{input.id}} ready"  → "order 7 ready"   (interpolated string)

func (r *Resolver) resolveStringTemplates(s string, rc *model.RunContext) (interface{}, error) {
    matches := r.templateRe.FindAllStringSubmatch(s, -1)
    if len(matches) == 0 {
        return s, nil
    }

    // Check if the entire string is just a single template expression (with optional whitespace)
    trimmed := strings.TrimSpace(s)
    if len(matches) == 1 {
        // Check if the string is exactly the template expression (with optional whitespace)
        fullMatch := strings.TrimSpace(matches[0][0])
        if trimmed == fullMatch {
            // Entire string is just the template - return the evaluated value directly
            rawExpr := strings.TrimSpace(matches[0][1])
            val, err := r.evalCELExpression(rawExpr, rc)
            if err != nil {
                return nil, err
            }
            return val, nil
        }
    }

    // Multiple templates or template is part of a larger string - do string replacement
    out := s
    for _, m := range matches {
        rawExpr := strings.TrimSpace(m[1])
        val, err := r.evalCELExpression(rawExpr, rc)
        if err != nil {
            return nil, err
        }
        // Convert val to string
        var valStr string
        switch t := val.(type) {
        case nil:
            valStr = ""
        case string:
            valStr = t
        default:
            b, err := json.Marshal(t)
            if err != nil {
                valStr = fmt.Sprintf("%v", t)
            } else {
                valStr = string(b)
            }
        }
        out = strings.Replace(out, m[0], valStr, 1)
    }
    return out, nil
}

Finally, evaluate one expression. Build the activation (the variable values) from the run context and run the CEL program:

func (r *Resolver) evalCELExpression(exprStr string, rc *model.RunContext) (interface{}, error) {
    decodeJSON := func(raw json.RawMessage) interface{} {
        if len(raw) == 0 {
            return map[string]interface{}{}
        }
        var v interface{}
        if err := json.Unmarshal(raw, &v); err != nil {
            // fall back to string representation
            return string(raw)
        }
        return v
    }

    ast, iss := r.celEnv.Parse(exprStr)
    if iss != nil && iss.Err() != nil {
        return nil, fmt.Errorf("cel parse error: %w", iss.Err())
    }

    checked, iss := r.celEnv.Check(ast)
    if iss != nil && iss.Err() != nil {
        return nil, fmt.Errorf("cel check error: %w", iss.Err())
    }

    prg, err := r.celEnv.Program(checked)
    if err != nil {
        return nil, err
    }

    // Prepare activation with available variables
    activation := map[string]interface{}{
        "input":  decodeJSON(rc.Input),
        "config": decodeJSON(rc.Config),
        "tasks":  r.buildStepsView(rc),
        "system": map[string]interface{}{"now": time.Now().UTC().Format(time.RFC3339)},
    }

    out, _, err := prg.Eval(activation)
    if err != nil {
        return nil, fmt.Errorf("cel eval error: %w", err)
    }
    return out.Value(), nil
}

CEL is a good fit here: expressions cannot loop forever, cannot do I/O, and cannot touch the host. A user's {{ … }} can only compute a value.

#Step 9: Test the model

Create pkg/model/dag_test.go. Test the parts with real logic — reference analysis and backoff:

// pkg/model/dag_test.go
func TestTaskRefs(t *testing.T) {
    node := func(id string, typ model.NodeType, spec string) model.Node {
        return model.Node{NodeID: id, Type: typ, Spec: json.RawMessage(spec)}
    }
    dag := model.BuildDAG(&model.WorkflowDefinitionSpec{
        Nodes: []model.Node{
            node("p", model.NodeTypeHTTP, `{}`),
            node("s", model.NodeTypeSwitch, `{}`),
            node("dyn", model.NodeTypeHTTP, `{"v": "{{tasks[input.key].output}}"}`),
            node("q", model.NodeTypeHTTP, `{"v": "{{tasks[\"my-node\"].output.x + tasks.other.output.y}}", "url": "/api/tasks/1"}`),
        },
        Edges: []model.Edge{
            {From: "p", To: "s"},
            {From: "s", To: "dyn", Condition: "{{tasks.p.output.n > 1}}"},
        },
    })

    if refs, ok := dag.TaskRefs("s"); !ok || !slices.Equal(refs, []string{"p", "p"}) {
        t.Fatalf("switch reads its conditions and passes its predecessor through: %v %v", refs, ok)
    }
    if _, ok := dag.TaskRefs("dyn"); ok {
        t.Fatal("dynamic task access must fall back to loading everything")
    }
    if refs, ok := dag.TaskRefs("q"); !ok || !slices.Equal(refs, []string{"my-node", "other"}) {
        t.Fatalf("quoted refs: %v %v", refs, ok)
    }
}

func TestNodeRetryDelay(t *testing.T) {
    n := model.Node{Retry: &model.RetryPolicy{MaxAttempts: 3, BackoffSeconds: 2}}
    if n.MaxAttempts() != 3 {
        t.Fatalf("max attempts: %d", n.MaxAttempts())
    }
    if d := n.RetryDelay(1).Seconds(); d != 2 {
        t.Fatalf("retry 1 delay %v", d)
    }
    if d := n.RetryDelay(3).Seconds(); d != 8 {
        t.Fatalf("retry 3 delay %v", d)
    }
    if (&model.Node{}).MaxAttempts() != 1 {
        t.Fatal("default max attempts should be 1")
    }
}

Notice "url": "/api/tasks/1" in node q: the word tasks outside {{ }} is ignored, as it should be.

go test ./pkg/model/ ./pkg/template/ -v

You now have a graph you can query in O(1) and templates you can resolve safely. Next you build the brain that walks this graph: the state machine.