Help us shape the future of Go! Take the 2026 Go Developer Survey

workflow

package
v1.4.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 31, 2026 License: MIT Imports: 22 Imported by: 0

Documentation

Overview

Package workflow composes nodes such as agents and functions into graphs with conditional branches, loops and dynamic fan-out. Independent nodes run concurrently and share a State. Runs can be streamed, persisted after every node with a Checkpointer, suspended for human input with Suspend and resumed, also after a crash.

Index

Constants

View Source 
const (
	Start = "__start__"
	End   = "__end__"
)

Start and End are the implicit entry and exit of a graph.

View Source 
const CheckpointVersion = 1

CheckpointVersion is the version of the Checkpoint format written by this package.

View Source 
const DefaultMaxSteps = 1000

DefaultMaxSteps bounds the node executions of a run of a workflow with loops when MaxSteps is not set.

Variables

View Source 
var ErrCheckpointNotFound = fmt.Errorf("workflow: %w", loop.ErrCheckpointNotFound)

ErrCheckpointNotFound is returned when a workflow checkpoint does not exist. It wraps loop.ErrCheckpointNotFound, so errors.Is matches either: check this error for workflow runs only, or loop's for any checkpoint.

View Source 
var ErrMaxIterations = errors.New("workflow: loop reached its iteration limit")

ErrMaxIterations is returned by Loop when the iteration limit is reached while the condition still holds.

View Source 
var ErrNodeTimeout = errors.New("workflow: node attempt timed out")

ErrNodeTimeout is matched, together with context.DeadlineExceeded, by the error of a node attempt that exceeded its Timeout.

View Source 
var ErrStepLimit = errors.New("workflow: step limit reached")

ErrStepLimit is returned when a run would start more node executions than MaxSteps allows.

View Source 
var ErrSuspended = errors.New("workflow: suspended")

ErrSuspended is returned by Suspend when the node must stop to wait for a resume value. Suspension matches it with errors.Is.

Functions

func Emit 

func Emit(ctx context.Context, v any) bool

Emit sends v to the consumers of the run as a CustomEvent of the current node. It reports false outside a workflow node.

func ErrorKey 

func ErrorKey(node string) string

ErrorKey returns the state key under which OnError stores the error of node.

func SubRunID 

func SubRunID(runID, node string, attempt, index int) string

SubRunID returns the run ID used for the agent of an AgentNode, or the nested workflow of AsNode, in the given attempt of node in workflow run runID; index is the Map item index, or -1. It is "<runID>-<node>-<attempt>[-i<index>]", where node is replaced by "_" and a hash of it unless it is at most 24 letters, digits and inner underscores. Runtime checkpointers do not accept dots in run IDs, so '-' separates the parts; the node part never contains one, so IDs of different runs, nodes or attempts never collide.

func Suspend 

func Suspend(ctx context.Context, payload any) (json.RawMessage, error)

Suspend pauses the run for human input, or any value from outside the run, describing what it needs with payload, which must be encodable as JSON.

The first time, Suspend returns ErrSuspended, which the node should return: the node stops, its writes are discarded and, once the nodes still running finish, the run ends with status suspended and a Pending entry for the node. Workflow.Resume with a value for the entry's key runs the node again from its start, and this time Suspend returns the value encoded as JSON. A node that calls Suspend several times gets the values in order, so it must call Suspend in the same order each time, and code before a Suspend runs again on resume: keep it free of side effects or idempotent.

Outside a workflow node Suspend returns an error.

func Template 

func Template(tmpl string) func(*State) string

Template returns a prompt function that replaces {{key}} with state values.

func Transient 

func Transient(err error) bool

Transient reports whether a node error may not recur: an attempt that exceeded its Timeout, or an error model.IsRetryable accepts, such as a rate limit, a server error or a dropped connection.

Types

type BranchFunc 

type BranchFunc func(ctx context.Context, st *State) ([]string, error)

BranchFunc chooses the successors to run after a node. It returns names among the branch targets; other targets are skipped. Returning End or no names skips all targets.

type Checkpoint 

type Checkpoint struct {
	Version  int    `json:"version"`
	RunID    string `json:"run_id"`
	Workflow string `json:"workflow,omitempty"`
	Status   Status `json:"status"`
	Error    string `json:"error,omitempty"`
	// Values holds the committed state values.
	Values map[string]json.RawMessage `json:"values"`
	// Done lists the nodes that finished, Skipped those skipped, Running
	// those that started and had not finished, Failed those that failed
	// and Suspended those waiting for a resume value.
	Done      []string `json:"done,omitempty"`
	Skipped   []string `json:"skipped,omitempty"`
	Running   []string `json:"running,omitempty"`
	Failed    []string `json:"failed,omitempty"`
	Suspended []string `json:"suspended,omitempty"`
	// Routes holds the successors each finished node, and Start, took.
	Routes map[string][]string `json:"routes,omitempty"`
	// Activated lists loop targets activated by a node that looped back.
	Activated []string `json:"activated,omitempty"`
	// Attempts counts every attempt of each node in the run, Tries the
	// attempts of its current execution and Visits its executions.
	Attempts map[string]int `json:"attempts,omitempty"`
	Tries    map[string]int `json:"tries,omitempty"`
	Visits   map[string]int `json:"visits,omitempty"`
	// Steps counts the node executions started; see MaxSteps.
	Steps int `json:"steps"`
	// Writers lists, per key, the nodes of the current executions that set
	// or deleted it without a reducer, for conflict detection.
	Writers map[string][]string `json:"writers,omitempty"`
	// Suspends holds, per pending key, the suspensions of the current
	// execution in the order Suspend was called.
	Suspends map[string][]*SuspendPoint `json:"suspends,omitempty"`
	// Maps holds the progress of Map nodes.
	Maps map[string]*MapProgress `json:"maps,omitempty"`
	// Writes holds the net writes committed by a nested workflow run
	// (see AsNode), replayed onto the parent's state when it resumes.
	Writes    map[string]ItemWrite `json:"writes,omitempty"`
	UpdatedAt time.Time            `json:"updated_at"`
}

Checkpoint is the persisted state of a workflow run.

Node sets describe the current execution of each node: a node that loops back is reset to pending together with the nodes reachable from its loop target.

type Checkpointer 

type Checkpointer interface {
	Save(ctx context.Context, cp *Checkpoint) error
	Load(ctx context.Context, runID string) (*Checkpoint, error)
	Delete(ctx context.Context, runID string) error
}

Checkpointer persists workflow runs by run ID: 1 to 64 ASCII letters, digits, '_' and '-'; the built-in checkpointers reject others.

type ConflictError 

type ConflictError struct {
	Key   string
	Nodes []string
}

ConflictError reports that nodes not ordered by the graph, such as parallel branches or items of a Map, wrote the same key without a Reducer.

func (*ConflictError) Error 

func (e *ConflictError) Error() string

type Event 

type Event struct {
	Type     EventType
	Workflow string
	// RunID identifies the run.
	RunID string
	// Node is the node the event belongs to. Events of a workflow nested
	// with AsNode are forwarded with Node set to "parent/child".
	Node string
	// Index is the item index for events of Map items, and -1 otherwise.
	Index int
	// Attempt counts the attempts of the node in its current execution,
	// starting at 1; see Retry.
	Attempt  int
	Err      error
	Duration time.Duration
	// Step is the number of node executions the run has started; see
	// MaxSteps.
	Step int

	Delta   map[string]any
	Removed []string
	// Pending describes the suspension for NodeSuspended.
	Pending *Pending
	// Agent is the forwarded event for AgentEvent.
	Agent *loop.Event
	// Custom is the value passed to Emit for CustomEvent.
	Custom any

	// Status and, for a suspended run, Suspended are set on RunEnd.
	Status    Status
	Suspended []*Pending
}

Event reports the progress of a run.

type EventType 

type EventType string

EventType identifies a workflow event. 

const (
	// RunStart is the first event of a run or a resumed run.
	RunStart    EventType = "run_start"
	NodeStart   EventType = "node_start"
	NodeEnd     EventType = "node_end"
	NodeSkipped EventType = "node_skipped"
	NodeFailed  EventType = "node_failed"
	// NodeSuspended reports that a node called Suspend; Pending describes
	// it.
	NodeSuspended EventType = "node_suspended"
	// StateUpdated reports the writes a node committed: Delta holds the new
	// values of the keys that changed and Removed the keys deleted.
	StateUpdated EventType = "state_updated"
	// AgentEvent forwards an event of the agent an AgentNode runs.
	AgentEvent EventType = "agent_event"
	// CustomEvent carries a value a node passed to Emit.
	CustomEvent EventType = "custom"
	// RunEnd is the last event; Status says how the run ended.
	RunEnd EventType = "run_end"
)

type Graph 

type Graph struct {
	// contains filtered or unexported fields
}

Graph builds a workflow. Methods record the first error, which Compile returns, so calls can be chained.

func NewGraph 

func NewGraph() *Graph

NewGraph returns an empty graph.

func (*Graph) AddBranch 

func (g *Graph) AddBranch(from string, cond BranchFunc, targets ...string) *Graph

AddBranch runs the targets chosen by cond after from completes.

A target may be a node from which from can be reached, which makes a loop: choosing it routes back. See Compile for which edges loop back and Workflow.Run for how loops run.

func (*Graph) AddChain 

func (g *Graph) AddChain(nodes ...string) *Graph

AddChain adds edges between consecutive nodes.

func (*Graph) AddCondition 

func (g *Graph) AddCondition(from string, choose func(ctx context.Context, st *State) (string, error), routes map[string]string) *Graph

AddCondition is a branch that picks one target by key: choose returns a key of routes and the matching node runs.

func (*Graph) AddEdge 

func (g *Graph) AddEdge(from, to string) *Graph

AddEdge runs to after from completes.

func (*Graph) AddFunc 

func (g *Graph) AddFunc(name string, fn func(ctx context.Context, st *State) error, opts ...NodeOption) *Graph

AddFunc adds a function node.

func (*Graph) AddNode 

func (g *Graph) AddNode(name string, n Node, opts ...NodeOption) *Graph

AddNode adds a named node.

func (*Graph) Compile 

func (g *Graph) Compile(opts ...Option) (*Workflow, error)

Compile validates the graph: every edge must join known nodes and every node must be reachable from Start.

Edges are walked depth first from Start, in the order they were added; an edge into a node on the current path loops back. Only branch edges, from AddBranch or AddCondition, may loop back: a cycle closed by AddEdge or OnError is an error.

func (*Graph) MustCompile 

func (g *Graph) MustCompile(opts ...Option) *Workflow

MustCompile is like Compile but panics on error.

type Info 

type Info struct {
	RunID string
	Node  string
	// Index is the Map item index, or -1.
	Index int
	// Attempt counts all attempts of the node in the run, including those
	// of earlier executions in loops; it is part of SubRunID. Try counts
	// the attempts of the current execution and Visit the executions.
	Attempt, Try, Visit int
}

Info describes the node attempt a context belongs to.

func NodeInfo 

func NodeInfo(ctx context.Context) (Info, bool)

NodeInfo returns information about the node attempt running with ctx. It reports false outside a workflow node.

type ItemWrite 

type ItemWrite struct {
	Key     string `json:"key"`
	Value   any    `json:"value,omitempty"`
	Deleted bool   `json:"deleted,omitempty"`
	// Replace marks a value produced by State.Update, which is stored
	// without its reducer.
	Replace bool `json:"replace,omitempty"`
}

ItemWrite is a write of a finished Map item.

type MapOption 

type MapOption func(*mapNode)

MapOption configures a Map node.

func MapConcurrency 

func MapConcurrency(n int) MapOption

MapConcurrency bounds the number of items running at once. Zero means unbounded.

func MapKeys 

func MapKeys(item, index string) MapOption

MapKeys sets the state keys under which each item sees its element and its index; they default to "item" and "index".

type MapProgress 

type MapProgress struct {
	// Items are the elements of the list, captured when the execution
	// started.
	Items []json.RawMessage `json:"items"`
	// Done holds the writes of each finished item.
	Done map[int][]ItemWrite `json:"done,omitempty"`
	// contains filtered or unexported fields
}

MapProgress is the progress of a Map node's current execution.

type MemoryCheckpointer 

type MemoryCheckpointer struct {
	// contains filtered or unexported fields
}

MemoryCheckpointer keeps checkpoints in process memory, encoded as JSON like a persistent checkpointer, so a state that cannot be stored fails here too. The zero value is ready to use.

func NewMemoryCheckpointer 

func NewMemoryCheckpointer() *MemoryCheckpointer

NewMemoryCheckpointer returns an empty in-memory checkpointer.

func (*MemoryCheckpointer) Delete 

func (c *MemoryCheckpointer) Delete(_ context.Context, runID string) error

Delete implements Checkpointer.

func (*MemoryCheckpointer) Load 

func (c *MemoryCheckpointer) Load(_ context.Context, runID string) (*Checkpoint, error)

Load implements Checkpointer.

func (*MemoryCheckpointer) Save 

func (c *MemoryCheckpointer) Save(_ context.Context, cp *Checkpoint) error

Save implements Checkpointer.

type Node 

type Node interface {
	Run(ctx context.Context, st *State) error
}

Node is a unit of work in a workflow.

func AgentNode 

func AgentNode(a agent.Agent, prompt func(*State) string, output string) Node

AgentNode runs an agent. The prompt is built from the state by prompt, and the agent's answer is stored under output.

In a workflow run the agent's events are forwarded as AgentEvent, and its run has the ID SubRunID of the node attempt. When the agent is interrupted for tool approvals, the node suspends with the loop.Interrupt as payload (see Pending.Interrupt); resuming the node's key with a map[string]loop.Approval resumes the agent from the interrupt with those decisions, from its state if the agent implements agent.StateResumer and otherwise with agent.Resumer. If the agent is interrupted again, the node suspends again.

When an attempt runs again after a crash or cancellation, an agent with a Runtime and a loop.Checkpointer, such as an agent.LLMAgent with agent.WithCheckpointer, resumes its checkpointed run instead of starting over, so finished tool calls are not repeated and cut-off calls with side effects are not run twice; a run that was interrupted suspends the node again with its pending requests. Other agents start over.

Outside a workflow run, an interrupted agent fails the node.

func Loop 

func Loop(name string, body Node, cond func(ctx context.Context, st *State) (bool, error), limit int) Node

Loop runs body while cond returns true, at most limit times. The condition is checked before each iteration. The current iteration, starting at 0, is stored in the state under "<name>.iteration" when name is set.

func Map 

func Map(key string, body Node, opts ...MapOption) Node

Map returns a node that runs body once per element of the list stored under key, for example to process documents a previous node produced.

Each item runs on its own view of the state, in which "item" holds the element and "index" its position (see MapKeys); those two keys are never committed. Items run concurrently, at most MapConcurrency at once, and do not see each other's writes. Once every item succeeded, their writes are applied in item order, so the result does not depend on timing: to collect results, have each item Set a key with a Reducer such as Append, which yields the results in item order. Items that set or delete the same key without a reducer fail the node with a ConflictError. The first failing item cancels the others and fails the node; retries rerun only the items that did not finish.

As a node of a graph, Map saves each finished item's writes with the run's Checkpointer and captures the list when its execution starts, so a resumed run reruns only unfinished items. Items may call Suspend; their Pending keys are "node[i]", and the node resumes once all its items have values. AgentNode bodies get the item index in their SubRunID. Items of a Map nested in another Map's item run without these features.

func Sequence 

func Sequence(nodes ...Node) Node

Sequence runs nodes one after another.

type NodeError 

type NodeError struct {
	Node string
	Err  error
}

NodeError reports the failure of a node.

func (*NodeError) Error 

func (e *NodeError) Error() string

func (*NodeError) Unwrap 

func (e *NodeError) Unwrap() error

type NodeFunc 

type NodeFunc func(ctx context.Context, st *State) error

NodeFunc adapts a function to Node.

func (NodeFunc) Run 

func (f NodeFunc) Run(ctx context.Context, st *State) error

Run calls f.

type NodeOption 

type NodeOption func(*nodeSpec)

NodeOption configures how a graph runs a node.

func OnError 

func OnError(fallback string) NodeOption

OnError routes to fallback when the node fails after its retries, instead of failing the run. The node's writes are discarded, its error message is stored under "<node>.error" and only fallback is taken among its successors. Cancellation and suspensions do not fall back. The fallback edge counts as an edge of the graph: it may not close a cycle.

func Retry 

func Retry(attempts int, backoff time.Duration) NodeOption

Retry retries a failed node up to attempts-1 more times, waiting backoff multiplied by the attempt number between attempts. Every error is retried. WithRetry offers exponential backoff and error classification.

func Timeout 

func Timeout(d time.Duration) NodeOption

Timeout bounds each attempt of a node. An attempt that exceeds it fails with an error matching ErrNodeTimeout and context.DeadlineExceeded.

func WithRetry 

func WithRetry(p RetryPolicy) NodeOption

WithRetry retries a failed node according to p, with exponential backoff and jitter. Cancellation of the run and suspensions are never retried.

type Option 

type Option func(*Workflow)

Option configures a compiled workflow. Options are also RunOptions: passed to Run, Stream or Resume they apply to that run only.

func MaxConcurrency 

func MaxConcurrency(n int) Option

MaxConcurrency bounds the number of nodes running at once. Zero means unbounded.

func MaxSteps 

func MaxSteps(n int) Option

MaxSteps bounds the number of node executions a run may start, retries excluded; the run fails with ErrStepLimit when a node would exceed it. Zero means DefaultMaxSteps for workflows with loops and no bound otherwise. Resumed runs keep counting.

func Named 

func Named(name string) Option

Named sets the workflow name reported in events.

func OnEvent 

func OnEvent(fn func(Event)) Option

OnEvent registers a callback for node events: NodeStart, NodeEnd, NodeSkipped, NodeFailed and NodeSuspended. Stream delivers every event. Callbacks run one at a time, in the order of the events, on the goroutine consuming the run.

func Reducer 

func Reducer(key string, fn ReducerFunc) Option

Reducer declares how writes to key are merged. A node's Set of key is an update passed to fn with the current value, so writes of parallel nodes combine instead of conflicting. Without a reducer, a Set replaces the value, and two nodes that the graph does not order, such as parallel branches, that both set or delete the same key fail the run with a ConflictError, whichever finishes second. Nodes ordered by edges may overwrite each other's values. Updates made with State.Update are replayed in commit order and never conflict.

Writes are committed when a node succeeds, in the order the node made them; Map items are committed in item order. The reducer of a nested workflow applies to its own writes, falling back to the parent's.

func WithCheckpointer 

func WithCheckpointer(cp Checkpointer) Option

WithCheckpointer persists runs with cp: after every node finishes, when a node attempt starts, after every Map item and when the run suspends or fails. State values must then be encodable as JSON; a run fails with an error naming the key otherwise. Failure to save a terminal checkpoint or delete a completed one also fails the run, preserving any execution error. Resumed runs see the decoded values, so read them with State.Value or State.Decode for typed access.

type Pending 

type Pending struct {
	// Key identifies the suspension in the map passed to Resume: the node
	// name, or "node[i]" for item i of a Map node.
	Key   string `json:"key"`
	Node  string `json:"node"`
	Index int    `json:"index"`
	// Payload is the JSON encoding of the value passed to Suspend.
	Payload json.RawMessage `json:"payload,omitempty"`
	// Interrupt is set when an AgentNode suspended because its agent needs
	// tool approvals. Resume the key with a map[string]loop.Approval
	// keyed by call ID; calls without a decision are denied.
	Interrupt *loop.Interrupt `json:"interrupt,omitempty"`
}

Pending is a suspension waiting for a resume value.

func (*Pending) Decode 

func (p *Pending) Decode(dst any) error

Decode decodes the payload into dst.

type ReducerFunc 

type ReducerFunc func(current, update any) any

ReducerFunc merges an update written to a key into the key's current value, which is nil if the key has no value.

func Append 

func Append() ReducerFunc

Append returns a reducer that appends updates to a list. If the current value is a slice, an update of the same slice type is concatenated and an update assignable to its element type is appended, keeping the slice type. Otherwise the result is a []any: a []any update is concatenated, any other update is appended as one element, and a current value that is not a slice becomes the first element.

type RetryPolicy 

type RetryPolicy struct {
	// MaxAttempts includes the first attempt. Defaults to 3.
	MaxAttempts int
	// InitialBackoff is the wait before the first retry, 500ms by default;
	// each further wait is Multiplier, 2 by default, times longer, up to
	// MaxBackoff, 30s by default. A provider's Retry-After is honored, up
	// to MaxBackoff.
	InitialBackoff time.Duration
	MaxBackoff     time.Duration
	Multiplier     float64
	// Jitter randomizes each wait by up to this fraction. Defaults to 0.2;
	// set a negative value to disable.
	Jitter float64
	// RetryIf decides whether an error is retried. By default every error
	// is; Transient retries only errors that may not recur.
	RetryIf func(error) bool
	// contains filtered or unexported fields
}

RetryPolicy configures how a failed node is retried; see WithRetry.

type RunOption 

type RunOption interface {
	// contains filtered or unexported methods
}

RunOption configures a single run; see Option and WithRunID.

func WithRunID 

func WithRunID(id string) RunOption

WithRunID sets the ID of a new run. IDs consist of letters, digits, '_' and '-', at most 64 of them. Without one, a random ID is generated.

type State 

type State struct {
	// contains filtered or unexported fields
}

State is the data shared by the nodes of a workflow run. Its map operations are safe for concurrent use. Values are shared without copying: treat them as immutable and replace them through Set or Update. Mutating a stored map, slice or pointer in place bypasses synchronization and staged writes.

Inside a workflow run each node attempt works on its own view of the state: it reads the values committed so far together with its own writes, and its writes reach the shared state only when the node succeeds, all at once. Writes of a failed or suspended attempt are discarded, and nodes running in parallel do not see each other's writes until they finish. See Reducer for how the writes of parallel nodes are merged.

func NewState 

func NewState(init map[string]any) *State

NewState returns a state holding a shallow copy of init.

func (*State) Decode 

func (s *State) Decode(key string, dst any) (bool, error)

Decode converts the value for key into dst through JSON.

func (*State) Delete 

func (s *State) Delete(key string)

Delete removes a value.

func (*State) Get 

func (s *State) Get(key string) (any, bool)

Get returns the value for key.

func (*State) Set 

func (s *State) Set(key string, value any)

Set stores a value. Inside a workflow node whose workflow declares a Reducer for key, value is an update that the reducer merges into the current value; Get then returns the merged value.

func (*State) Snapshot 

func (s *State) Snapshot() map[string]any

Snapshot returns a shallow copy of all values.

func (*State) String 

func (s *State) String(key string) string

String returns the value for key formatted as a string, or "".

func (*State) Update 

func (s *State) Update(fn func(values map[string]any))

Update calls fn with exclusive access to the state's map for read-modify-write operations. fn must replace mutable values rather than modify them in place, must not retain the map, and must not call other methods of the same State.

Inside a workflow node, fn first runs on the node's view, so the node sees its effect, and runs again when the node's writes are committed, on the values current then. Updates of parallel nodes therefore compose instead of conflicting. fn must not have other side effects. In Map items, whose writes are stored per item, the values fn produced are committed instead, without running it again.

func (*State) Value 

func (s *State) Value[T any](key string) (T, bool, error)

Value returns the value for key as a T, converting through JSON as Decode does, so values restored from checkpoints, which are plain JSON values, read as the type they were stored as. It reports false if the state has no such value.

type Status 

type Status string

Status is the state of a workflow run. 

const (
	StatusRunning   Status = "running"
	StatusSuspended Status = "suspended"
	StatusCompleted Status = "completed"
	StatusFailed    Status = "failed"
)

type SuspendPoint 

type SuspendPoint struct {
	// Agent is set for the suspensions of an AgentNode, whose payload is
	// a loop.Interrupt.
	Agent   bool            `json:"agent,omitempty"`
	Payload json.RawMessage `json:"payload,omitempty"`
	Resume  json.RawMessage `json:"resume,omitempty"`
}

SuspendPoint is a call of Suspend and, once given, its resume value.

type Suspension 

type Suspension struct {
	RunID   string
	Pending []*Pending
}

Suspension is the error Run and Resume return for a run that suspended. Resume it with Workflow.Resume and a value for each pending key.

func (*Suspension) Error 

func (s *Suspension) Error() string

func (*Suspension) Is 

func (s *Suspension) Is(target error) bool

Is reports whether target is ErrSuspended.

type Workflow 

type Workflow struct {
	// contains filtered or unexported fields
}

Workflow is a compiled graph. It is safe for concurrent runs.

func (*Workflow) AsNode 

func (w *Workflow) AsNode() Node

AsNode returns the workflow as a node of another workflow, sharing the parent's state; the nested run's writes are committed with the node. Its events are forwarded to the parent run with Node set to "parent/child".

Inside a workflow run, the nested run has the ID SubRunID of the node's attempt and uses the parent's Checkpointer, so a parent resumed after a crash resumes the nested run where it stopped. When the nested run suspends, the node suspends with the nested run's []*Pending as payload; resume the node's key with a map from the nested keys to their values.

func (*Workflow) Discard 

func (w *Workflow) Discard(ctx context.Context, runID string, opts ...RunOption) error

Discard deletes the checkpoint of a run that will not be resumed, or the copy kept in memory without a Checkpointer. A missing run is not an error.

func (*Workflow) Mermaid 

func (w *Workflow) Mermaid() string

Mermaid renders the workflow as a Mermaid flowchart. Edges added with AddEdge are solid, branch edges dashed, edges that loop back are labeled "loop" and OnError fallbacks "on error".

func (*Workflow) Resume 

func (w *Workflow) Resume(ctx context.Context, runID string, resume map[string]any, opts ...RunOption) (*State, error)

Resume continues a run from its checkpoint, after a crash, a failure or a suspension, and returns the state; see ResumeStream.

func (*Workflow) ResumeStream 

func (w *Workflow) ResumeStream(ctx context.Context, runID string, resume map[string]any, opts ...RunOption) iter.Seq2[*Event, error]

ResumeStream is like Resume but returns the events of the resumed run; see Stream.

resume maps the keys of pending suspensions (see Pending) to their values, which are encoded as JSON unless they are a json.RawMessage. A node runs again once all its suspensions have values; keys left out stay pending. Nodes that were running when the run stopped run again with the same attempt number; failed nodes run again with fresh retries.

Without a Checkpointer, only runs that suspended in this process can be resumed: the Workflow keeps them in memory until they complete or are discarded.

func (*Workflow) Run 

func (w *Workflow) Run(ctx context.Context, st *State, opts ...RunOption) (*State, error)

Run executes the workflow on st, which may be nil, and returns the state. It consumes Stream; see there for the semantics. A run that suspends returns a *Suspension error, which matches ErrSuspended.

func (*Workflow) Stream 

func (w *Workflow) Stream(ctx context.Context, st *State, opts ...RunOption) iter.Seq2[*Event, error]

Stream executes the workflow on st, which may be nil, and returns its events. The last event is RunEnd; a failed run then yields its error.

A node runs once all its predecessors have finished or been skipped and at least one incoming edge was taken; otherwise it is skipped. Edges that loop back do not count as predecessors. The first failure, after retries and fallbacks, cancels the remaining nodes and fails the run once they returned.

When a node's branch routes back to a loop target, the run waits until no node reachable from the target is running or suspended, resets those nodes to pending and runs the target again; they run again as they are reached. A node either loops back or continues: choosing a loop target together with other targets is an error. Every node execution counts as a step; see MaxSteps.

With a Checkpointer the run is saved when a node attempt starts and after each node, so Resume continues it after a crash. Node code runs at least once: an attempt cut off by a crash runs again, with the same attempt number. State writes take effect exactly once, because they are committed with the node's completion. If the consumer stops iterating, running nodes are cancelled and the run can be resumed.

Source Files

View all Source files

Directories

Path Synopsis
Package checkpointtest checks workflow.Checkpointer implementations against the contract the built-in checkpointers follow.
 Package checkpointtest checks workflow.Checkpointer implementations against the contract the built-in checkpointers follow.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.